chore(sinks): mark VectorSink::from_event_sink as deprecated (vectordotdev#17649)

This will help steer contributors away from writing new sinks in the
deprecated fashion.

See vectordotdev#9261 for more context.

<!--
**Your PR title must conform to the conventional commit spec!**

  <type>(<scope>)!: <description>

  * `type` = chore, enhancement, feat, fix, docs
  * `!` = OPTIONAL: signals a breaking change
* `scope` = Optional when `type` is "chore" or "docs", available scopes
https://github.com/vectordotdev/vector/blob/master/.github/semantic.yml#L20
  * `description` = short description of the change

Examples:

  * enhancement(file source): Add `sort` option to sort discovered files
  * feat(new source): Initial `statsd` source
  * fix(file source): Fix a bug discovering new files
  * chore(external docs): Clarify `batch_size` option
-->

Author: dsmith3197

lib/vector-core/src/sink.rs

@@ -61,6 +61,12 @@ impl VectorSink {
     }
 
     /// Converts an event sink into a `VectorSink`
+    ///
+    /// Deprecated in favor of `VectorSink::from_event_streamsink`. See [vector/9261]
+    /// for more info.
+    ///
+    /// [vector/9261]: https://github.com/vectordotdev/vector/issues/9261
+    #[deprecated]
     pub fn from_event_sink(sink: impl Sink<Event, Error = ()> + Send + Unpin + 'static) -> Self {
         VectorSink::Sink(Box::new(EventSink::new(sink)))
     }

src/config/unit_test/unit_test_components.rs

@@ -302,6 +302,7 @@ impl SinkConfig for UnitTestStreamSinkConfig {
         let sink = self.sink.lock().await.take().unwrap();
         let healthcheck = future::ok(()).boxed();
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/appsignal/mod.rs

@@ -136,6 +136,7 @@ impl SinkConfig for AppsignalSinkConfig {
         )
         .boxed();
 
+        #[allow(deprecated)]
         Ok((super::VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/aws_cloudwatch_metrics/mod.rs

@@ -252,6 +252,7 @@ impl CloudWatchMetricsSvc {
                 })
             });
 
+        #[allow(deprecated)]
         Ok(VectorSink::from_event_sink(sink))
     }
 

src/sinks/azure_monitor_logs.rs

@@ -204,6 +204,7 @@ impl SinkConfig for AzureMonitorLogsConfig {
         )
         .sink_map_err(|error| error!(message = "Fatal azure_monitor_logs sink error.", %error));
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 
@@ -471,6 +472,7 @@ mod tests {
         .sink_map_err(|error| error!(message = "Fatal azure_monitor_logs sink error.", %error));
 
         let event = Event::Log(LogEvent::from("simple message"));
+        #[allow(deprecated)]
         run_and_assert_sink_compliance(
             VectorSink::from_event_sink(sink),
             stream::once(ready(event)),

src/sinks/clickhouse/http_sink.rs

@@ -47,6 +47,7 @@ pub(crate) async fn build_http_sink(
 
     let healthcheck = healthcheck(client, config).boxed();
 
+    #[allow(deprecated)]
     Ok((VectorSink::from_event_sink(sink), healthcheck))
 }
 

src/sinks/gcp/pubsub.rs

@@ -135,6 +135,7 @@ impl SinkConfig for PubsubConfig {
         )
         .sink_map_err(|error| error!(message = "Fatal gcp_pubsub sink error.", %error));
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/gcp/stackdriver_logs.rs

@@ -237,6 +237,7 @@ impl SinkConfig for StackdriverConfig {
         )
         .sink_map_err(|error| error!(message = "Fatal gcp_stackdriver_logs sink error.", %error));
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/gcp/stackdriver_metrics.rs

@@ -126,6 +126,7 @@ impl SinkConfig for StackdriverConfig {
             |error| error!(message = "Fatal gcp_stackdriver_metrics sink error.", %error),
         );
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/honeycomb.rs

@@ -108,6 +108,7 @@ impl SinkConfig for HoneycombConfig {
 
         let healthcheck = healthcheck(self.clone(), client).boxed();
 
+        #[allow(deprecated)]
         Ok((super::VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/http.rs

@@ -251,6 +251,7 @@ impl SinkConfig for HttpSinkConfig {
         )
         .sink_map_err(|error| error!(message = "Fatal HTTP sink error.", %error));
 
+        #[allow(deprecated)]
         let sink = super::VectorSink::from_event_sink(sink);
 
         Ok((sink, healthcheck))

src/sinks/influxdb/logs.rs

@@ -232,6 +232,7 @@ impl SinkConfig for InfluxDbLogsConfig {
         )
         .sink_map_err(|error| error!(message = "Fatal influxdb_logs sink error.", %error));
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/influxdb/metrics.rs

@@ -193,6 +193,7 @@ impl InfluxDbSvc {
             })
             .sink_map_err(|error| error!(message = "Fatal influxdb sink error.", %error));
 
+        #[allow(deprecated)]
         Ok(VectorSink::from_event_sink(sink))
     }
 }

src/sinks/mezmo.rs

@@ -175,6 +175,7 @@ impl SinkConfig for MezmoConfig {
 
         let healthcheck = healthcheck(self.clone(), client).boxed();
 
+        #[allow(deprecated)]
         Ok((super::VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/prometheus/remote_write.rs

@@ -261,6 +261,7 @@ impl SinkConfig for RemoteWriteConfig {
                 )
         };
 
+        #[allow(deprecated)]
         Ok((sinks::VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/sinks/redis.rs

@@ -235,6 +235,7 @@ impl RedisSinkConfig {
             })
             .sink_map_err(|error| error!(message = "Sink failed to flush.", %error));
 
+        #[allow(deprecated)]
         Ok(super::VectorSink::from_event_sink(sink))
     }
 

src/sinks/sematext/metrics.rs

@@ -194,6 +194,7 @@ impl SematextMetricsService {
             })
             .sink_map_err(|error| error!(message = "Fatal sematext metrics sink error.", %error));
 
+        #[allow(deprecated)]
         Ok(VectorSink::from_event_sink(sink))
     }
 }

src/sinks/util/adaptive_concurrency/tests.rs

@@ -200,6 +200,7 @@ impl SinkConfig for TestConfig {
         );
         *self.controller_stats.lock().unwrap() = stats;
 
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(sink), healthcheck))
     }
 

src/test_util/mock/sinks/error.rs

@@ -30,6 +30,7 @@ impl_generate_config_from_default!(ErrorSinkConfig);
 #[async_trait]
 impl SinkConfig for ErrorSinkConfig {
     async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(ErrorSink), ok(()).boxed()))
     }
 

src/test_util/mock/sinks/panic.rs

@@ -30,6 +30,7 @@ impl_generate_config_from_default!(PanicSinkConfig);
 #[async_trait]
 impl SinkConfig for PanicSinkConfig {
     async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        #[allow(deprecated)]
         Ok((VectorSink::from_event_sink(PanicSink), ok(()).boxed()))
     }