chore: add sink prelude (#17595)

Generally our stream based sinks will share a lot of imports. This adds
a `crate::sinks::prelude` module that can be imported to bring in the
commonly used imports. The advantage is not only reducing the size of
our imports and making it easier to maintain (often changes to the
framework result in having to change the imports of all the sinks
modules which is tedious) but it also encourages and guides using the
shared components available.

This also updates the following sinks to use the prelude:

- amqp
- kafka
- aws_kinesis
- loki
- pulsar

---------

Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>

Author: StephenWakely

docs/tutorials/sinks/1_basic_sink.md

@@ -22,16 +22,7 @@ Provide some module level comments to explain what the sink does.
 Let's setup all the imports we will need for the tutorial:
 
 ```rust
-use super::Healthcheck;
-use crate::config::{GenerateConfig, SinkConfig, SinkContext};
-use futures::{stream::BoxStream, StreamExt};
-use vector_common::finalization::{EventStatus, Finalizable};
-use vector_config::configurable_component;
-use vector_core::{
-    config::{AcknowledgementsConfig, Input},
-    event::Event,
-    sink::{StreamSink, VectorSink},
-};
+use crate::prelude::*;
 ```
 
 # Configuration

docs/tutorials/sinks/2_http_sink.md

@@ -12,32 +12,11 @@ To start, update our imports to the following:
 use std::task::Poll;
 
 use crate::{
-    config::{GenerateConfig, SinkConfig, SinkContext},
+    sinks::prelude::*,
     http::HttpClient,
     internal_events::SinkRequestBuildError,
-    sinks::util::{
-        encoding::{write_all, Encoder},
-        metadata::RequestMetadataBuilder,
-        request_builder::EncodeResult,
-        Compression, RequestBuilder, SinkBuilderExt,
-    },
-    sinks::Healthcheck,
 };
 use bytes::Bytes;
-use futures::{future::BoxFuture, stream::BoxStream, StreamExt};
-use vector_common::{
-    finalization::{EventFinalizers, EventStatus, Finalizable},
-    internal_event::CountByteSize,
-    request_metadata::{MetaDescriptive, RequestMetadata},
-};
-use vector_config::configurable_component;
-use vector_core::{
-    config::{AcknowledgementsConfig, Input},
-    event::Event,
-    sink::{StreamSink, VectorSink},
-    stream::DriverResponse,
-    tls::TlsSettings,
-};
 ```
 
 # Configuration

src/internal_events/mod.rs

@@ -259,7 +259,7 @@ pub(crate) use self::unix::*;
 pub(crate) use self::websocket::*;
 #[cfg(windows)]
 pub(crate) use self::windows::*;
-pub(crate) use self::{
+pub use self::{
     adaptive_concurrency::*, batch::*, common::*, conditions::*, encoding_transcode::*,
     heartbeat::*, open::*, process::*, socket::*, tcp::*, template::*, udp::*,
 };

src/sinks/amqp/config.rs

@@ -1,17 +1,8 @@
 //! Configuration functionality for the `AMQP` sink.
-use crate::{
-    amqp::AmqpConfig,
-    codecs::EncodingConfig,
-    config::{DataType, GenerateConfig, Input, SinkConfig, SinkContext},
-    sinks::{Healthcheck, VectorSink},
-    template::Template,
-};
+use crate::{amqp::AmqpConfig, sinks::prelude::*};
 use codecs::TextSerializerConfig;
-use futures::FutureExt;
 use lapin::{types::ShortString, BasicProperties};
 use std::sync::Arc;
-use vector_config::configurable_component;
-use vector_core::config::AcknowledgementsConfig;
 
 use super::sink::AmqpSink;
 

src/sinks/amqp/encoder.rs

@@ -1,8 +1,5 @@
 //! Encoding for the `AMQP` sink.
-use crate::{
-    event::Event,
-    sinks::util::encoding::{write_all, Encoder},
-};
+use crate::sinks::prelude::*;
 use bytes::BytesMut;
 use std::io;
 use tokio_util::codec::Encoder as _;
@@ -13,7 +10,7 @@ pub(super) struct AmqpEncoder {
     pub(super) transformer: crate::codecs::Transformer,
 }
 
-impl Encoder<Event> for AmqpEncoder {
+impl encoding::Encoder<Event> for AmqpEncoder {
     fn encode_input(&self, mut input: Event, writer: &mut dyn io::Write) -> io::Result<usize> {
         let mut body = BytesMut::new();
         self.transformer.transform(&mut input);

src/sinks/amqp/request_builder.rs

@@ -1,22 +1,10 @@
 //! Request builder for the `AMQP` sink.
 //! Responsible for taking the event (which includes rendered template values) and turning
 //! it into the raw bytes and other data needed to send the request to `AMQP`.
-use crate::{
-    event::Event,
-    sinks::util::{
-        metadata::RequestMetadataBuilder, request_builder::EncodeResult, Compression,
-        RequestBuilder,
-    },
-};
+use crate::sinks::prelude::*;
 use bytes::Bytes;
 use lapin::BasicProperties;
 use std::io;
-use vector_common::{
-    finalization::{EventFinalizers, Finalizable},
-    json_size::JsonSize,
-    request_metadata::RequestMetadata,
-};
-use vector_core::EstimatedJsonEncodedSizeOf;
 
 use super::{encoder::AmqpEncoder, service::AmqpRequest, sink::AmqpEvent};
 

src/sinks/amqp/service.rs

@@ -1,6 +1,9 @@
 //! The main tower service that takes the request created by the request builder
 //! and sends it to `AMQP`.
-use crate::internal_events::sink::{AmqpAcknowledgementError, AmqpDeliveryError};
+use crate::{
+    internal_events::sink::{AmqpAcknowledgementError, AmqpDeliveryError},
+    sinks::prelude::*,
+};
 use bytes::Bytes;
 use futures::future::BoxFuture;
 use lapin::{options::BasicPublishOptions, BasicProperties};
@@ -9,14 +12,6 @@ use std::{
     sync::Arc,
     task::{Context, Poll},
 };
-use tower::Service;
-use vector_common::{
-    finalization::{EventFinalizers, EventStatus, Finalizable},
-    internal_event::CountByteSize,
-    json_size::JsonSize,
-    request_metadata::{MetaDescriptive, RequestMetadata},
-};
-use vector_core::stream::DriverResponse;
 
 /// The request contains the data to send to `AMQP` together
 /// with the information need to route the message.

src/sinks/amqp/sink.rs

@@ -1,19 +1,9 @@
 //! The sink for the `AMQP` sink that wires together the main stream that takes the
 //! event and sends it to `AMQP`.
-use crate::{
-    codecs::Transformer, event::Event, internal_events::TemplateRenderingError,
-    sinks::util::builder::SinkBuilderExt, template::Template,
-};
-use async_trait::async_trait;
-use futures::StreamExt;
-use futures_util::stream::BoxStream;
+use crate::sinks::prelude::*;
 use lapin::{options::ConfirmSelectOptions, BasicProperties};
 use serde::Serialize;
 use std::sync::Arc;
-use tower::ServiceBuilder;
-use vector_buffers::EventCount;
-use vector_common::json_size::JsonSize;
-use vector_core::{sink::StreamSink, ByteSizeOf, EstimatedJsonEncodedSizeOf};
 
 use super::{
     config::{AmqpPropertiesConfig, AmqpSinkConfig},

src/sinks/aws_kinesis/config.rs

@@ -1,19 +1,13 @@
 use std::marker::PhantomData;
 
-use tower::ServiceBuilder;
-use vector_config::configurable_component;
-use vector_core::{
-    config::{DataType, Input},
-    sink::VectorSink,
-    stream::BatcherSettings,
-};
+use vector_core::stream::BatcherSettings;
 
 use crate::{
     aws::{AwsAuthentication, RegionOrEndpoint},
-    codecs::{Encoder, EncodingConfig},
-    config::AcknowledgementsConfig,
-    sinks::util::{retries::RetryLogic, Compression, ServiceBuilderExt, TowerRequestConfig},
-    tls::TlsConfig,
+    sinks::{
+        prelude::*,
+        util::{retries::RetryLogic, TowerRequestConfig},
+    },
 };
 
 use super::{
@@ -78,7 +72,7 @@ impl KinesisSinkBaseConfig {
 }
 
 /// Builds an aws_kinesis sink.
-pub async fn build_sink<C, R, RR, E, RT>(
+pub fn build_sink<C, R, RR, E, RT>(
     config: &KinesisSinkBaseConfig,
     partition_key_field: Option<String>,
     batch_settings: BatcherSettings,

src/sinks/aws_kinesis/firehose/config.rs

@@ -141,8 +141,7 @@ impl SinkConfig for KinesisFirehoseSinkConfig {
             None,
             batch_settings,
             KinesisFirehoseClient { client },
-        )
-        .await?;
+        )?;
 
         Ok((sink, healthcheck))
     }

src/sinks/aws_kinesis/service.rs

@@ -5,16 +5,13 @@ use std::{
 
 use aws_smithy_client::SdkError;
 use aws_types::region::Region;
-use futures::future::BoxFuture;
-use tower::Service;
-use vector_common::{json_size::JsonSize, request_metadata::MetaDescriptive};
-use vector_core::{internal_event::CountByteSize, stream::DriverResponse};
+use vector_core::internal_event::CountByteSize;
 
 use super::{
     record::{Record, SendRecord},
     sink::BatchKinesisRequest,
 };
-use crate::event::EventStatus;
+use crate::{event::EventStatus, sinks::prelude::*};
 
 pub struct KinesisService<C, T, E> {
     pub client: C,

src/sinks/aws_kinesis/sink.rs

@@ -1,22 +1,13 @@
 use std::{borrow::Cow, fmt::Debug, marker::PhantomData, num::NonZeroUsize};
 
-use async_trait::async_trait;
-use futures::{future, stream::BoxStream, StreamExt};
 use rand::random;
-use tower::Service;
-use vector_common::{
-    finalization::{EventFinalizers, Finalizable},
-    request_metadata::{MetaDescriptive, RequestMetadata},
-};
-use vector_core::{
-    partition::Partitioner,
-    stream::{BatcherSettings, DriverResponse},
-};
 
 use crate::{
-    event::{Event, LogEvent},
     internal_events::{AwsKinesisStreamNoPartitionKeyError, SinkRequestBuildError},
-    sinks::util::{processed_event::ProcessedEvent, SinkBuilderExt, StreamSink},
+    sinks::{
+        prelude::*,
+        util::{processed_event::ProcessedEvent, StreamSink},
+    },
 };
 
 use super::{

src/sinks/aws_kinesis/streams/config.rs

@@ -148,8 +148,7 @@ impl SinkConfig for KinesisStreamsSinkConfig {
             self.partition_key_field.clone(),
             batch_settings,
             KinesisStreamClient { client },
-        )
-        .await?;
+        )?;
 
         Ok((sink, healthcheck))
     }

src/sinks/kafka/config.rs

@@ -5,20 +5,15 @@ use futures::FutureExt;
 use rdkafka::ClientConfig;
 use serde_with::serde_as;
 use vector_config::configurable_component;
-use vector_core::schema::Requirement;
 use vrl::value::Kind;
 
 use crate::{
-    codecs::EncodingConfig,
-    config::{AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext},
     kafka::{KafkaAuthConfig, KafkaCompression},
     serde::json::to_string,
     sinks::{
         kafka::sink::{healthcheck, KafkaSink},
-        util::{BatchConfig, NoDefaultsBatchSettings},
-        Healthcheck, VectorSink,
+        prelude::*,
     },
-    template::Template,
 };
 
 pub(crate) const QUEUED_MIN_MESSAGES: u64 = 100000;

src/sinks/kafka/service.rs

@@ -1,29 +1,17 @@
 use std::task::{Context, Poll};
 
 use bytes::Bytes;
-use futures::future::BoxFuture;
 use rdkafka::{
     error::KafkaError,
     message::OwnedHeaders,
     producer::{FutureProducer, FutureRecord},
     util::Timeout,
 };
-use tower::Service;
-use vector_common::{
-    json_size::JsonSize,
-    request_metadata::{MetaDescriptive, RequestMetadata},
-};
-use vector_core::{
-    internal_event::{
-        ByteSize, BytesSent, CountByteSize, InternalEventHandle as _, Protocol, Registered,
-    },
-    stream::DriverResponse,
+use vector_core::internal_event::{
+    ByteSize, BytesSent, CountByteSize, InternalEventHandle as _, Protocol, Registered,
 };
 
-use crate::{
-    event::{EventFinalizers, EventStatus, Finalizable},
-    kafka::KafkaStatisticsContext,
-};
+use crate::{kafka::KafkaStatisticsContext, sinks::prelude::*};
 
 pub struct KafkaRequest {
     pub body: Bytes,

src/sinks/kafka/sink.rs

@@ -1,5 +1,4 @@
-use async_trait::async_trait;
-use futures::{future, stream::BoxStream, StreamExt};
+use futures::future;
 use rdkafka::{
     consumer::{BaseConsumer, Consumer},
     error::KafkaError,
@@ -12,17 +11,11 @@ use tower::limit::ConcurrencyLimit;
 
 use super::config::{KafkaRole, KafkaSinkConfig};
 use crate::{
-    codecs::{Encoder, Transformer},
-    event::{Event, LogEvent},
     kafka::KafkaStatisticsContext,
-    sinks::{
-        kafka::{
-            config::QUEUED_MIN_MESSAGES, request_builder::KafkaRequestBuilder,
-            service::KafkaService,
-        },
-        util::{builder::SinkBuilderExt, StreamSink},
+    sinks::kafka::{
+        config::QUEUED_MIN_MESSAGES, request_builder::KafkaRequestBuilder, service::KafkaService,
     },
-    template::{Template, TemplateParseError},
+    sinks::prelude::*,
 };
 
 #[derive(Debug, Snafu)]

src/sinks/kafka/tests.rs

@@ -29,10 +29,8 @@ mod integration_test {
                 sink::KafkaSink,
                 *,
             },
-            util::{BatchConfig, NoDefaultsBatchSettings},
-            VectorSink,
+            prelude::*,
         },
-        template::Template,
         test_util::{
             components::{assert_sink_compliance, SINK_TAGS},
             random_lines_with_stream, random_string, wait_for,

src/sinks/loki/config.rs

@@ -1,21 +1,12 @@
 use std::collections::HashMap;
 
-use futures::future::FutureExt;
-use vector_config::configurable_component;
 use vrl::value::Kind;
 
 use super::{healthcheck::healthcheck, sink::LokiSink};
 use crate::{
-    codecs::EncodingConfig,
-    config::{AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext},
     http::{Auth, HttpClient, MaybeAuth},
     schema,
-    sinks::{
-        util::{BatchConfig, Compression, SinkBatchSettings, TowerRequestConfig, UriSerde},
-        VectorSink,
-    },
-    template::Template,
-    tls::{TlsConfig, TlsSettings},
+    sinks::{prelude::*, util::UriSerde},
 };
 
 /// Loki-specific compression.