vectordotdev · blt · Sep 16, 2021 · Aug 20, 2021 · Aug 20, 2021 · Aug 21, 2021
@@ -216,6 +216,7 @@ encoding_rs = { version = "0.8.28", features = ["serde"] }
 evmap = { version = "10.0.2", default-features = false, optional = true }
 exitcode = { version = "1.1.2", default-features = false }
 flate2 = { version = "1.0.21", default-features = false }
+futures-util = { version = "0.3.17", default-features = false }
 getset = { version = "0.1.1", default-features = false }
 glob = { version = "0.3.0", default-features = false }
 grok = { version = "1.2.0", default-features = false, optional = true }

@@ -260,7 +260,8 @@ pub enum BatchStatus {
 }
 
 impl BatchStatus {
-    /// Update this status with another batch's delivery status, and return the result.
+    /// Update this status with another batch's delivery status, and return the
+    /// result.
     #[allow(clippy::match_same_arms)] // False positive: https://github.com/rust-lang/rust-clippy/issues/860
     fn update(self, status: EventStatus) -> Self {
         match (self, status) {
@@ -281,21 +282,24 @@ impl BatchStatus {
 #[derivative(Default)]
 #[repr(u8)]
 pub enum EventStatus {
-    /// All copies of this event were dropped without being finalized (the default).
+    /// All copies of this event were dropped without being finalized (the
+    /// default).
     #[derivative(Default)]
     Dropped,
     /// All copies of this event were delivered successfully.
     Delivered,
     /// At least one copy of this event encountered a retriable error.
     Errored,
-    /// At least one copy of this event encountered a permanent failure or rejection.
+    /// At least one copy of this event encountered a permanent failure or
+    /// rejection.
     Failed,
     /// This status has been recorded and should not be updated.
     Recorded,
 }
 
 impl EventStatus {
-    /// Update this status with another event's finalization status and return the result.
+    /// Update this status with another event's finalization status and return
+    /// the result.
     ///
     /// # Panics
     ///

@@ -149,7 +149,7 @@ impl<B> Service<Request<B>> for HttpClient<B>
 where
     B: fmt::Debug + HttpBody + Send + 'static,
     B::Data: Send,
-    B::Error: Into<crate::Error>,
+    B::Error: Into<crate::Error> + Send,
 {
     type Response = http::Response<Body>;
     type Error = HttpError;

@@ -1,23 +1,29 @@
+use super::service::LogApiRetry;
 use crate::config::{DataType, GenerateConfig, SinkConfig, SinkContext};
 use crate::http::HttpClient;
 use crate::sinks::datadog::logs::healthcheck::healthcheck;
-use crate::sinks::datadog::logs::service;
-use crate::sinks::datadog::ApiKey;
+use crate::sinks::datadog::logs::service::LogApiService;
+use crate::sinks::datadog::logs::sink::LogSink;
 use crate::sinks::datadog::Region;
 use crate::sinks::util::encoding::EncodingConfigWithDefault;
-use crate::sinks::util::{
-    batch::{Batch, BatchError},
-    http::{HttpSink, PartitionHttpSink},
-    BatchConfig, BatchSettings, Compression, JsonArrayBuffer, PartitionBuffer,
-    PartitionInnerBuffer, TowerRequestConfig,
-};
+use crate::sinks::util::service::ServiceBuilderExt;
+use crate::sinks::util::Concurrency;
+use crate::sinks::util::{BatchConfig, Compression, TowerRequestConfig};
 use crate::sinks::{Healthcheck, VectorSink};
 use crate::tls::{MaybeTlsSettings, TlsConfig};
-use futures::{FutureExt, SinkExt};
+use futures::FutureExt;
 use indoc::indoc;
 use serde::{Deserialize, Serialize};
 use std::convert::TryFrom;
-use std::{sync::Arc, time::Duration};
+use std::sync::Arc;
+use std::time::Duration;
+use tower::ServiceBuilder;
+use vector_core::config::proxy::ProxyConfig;
+
+const DEFAULT_REQUEST_LIMITS: TowerRequestConfig = {
+    TowerRequestConfig::const_new(Concurrency::Fixed(50), Concurrency::Fixed(50))
+        .rate_limit_num(250)
+};
 
 #[derive(Deserialize, Serialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
@@ -81,82 +87,54 @@ impl DatadogLogsConfig {
             });
         http::Uri::try_from(endpoint).expect("URI not valid")
     }
+}
 
-    fn batch_settings<T: Batch>(&self) -> Result<BatchSettings<T>, BatchError> {
-        BatchSettings::default()
-            .bytes(bytesize::mib(5_u32))
-            .events(1_000)
-            .timeout(15)
-            .parse_config(self.batch)
-    }
-
-    /// Builds the required BatchedHttpSink.
-    /// Since the DataDog sink can create one of two different sinks, this
-    /// extracts most of the shared functionality required to create either sink.
-    fn build_sink<T, B, O>(
+impl DatadogLogsConfig {
+    pub fn build_processor(
         &self,
+        client: HttpClient,
         cx: SinkContext,
-        service: T,
-        batch: B,
-        timeout: Duration,
-    ) -> crate::Result<(VectorSink, Healthcheck)>
-    where
-        O: 'static,
-        B: Batch<Output = Vec<O>> + std::marker::Send + 'static,
-        B::Output: std::marker::Send + Clone,
-        B::Input: std::marker::Send,
-        T: HttpSink<
-                Input = PartitionInnerBuffer<B::Input, ApiKey>,
-                Output = PartitionInnerBuffer<B::Output, ApiKey>,
-            > + Clone,
-    {
-        let request_settings = self.request.unwrap_with(&TowerRequestConfig::default());
+    ) -> crate::Result<VectorSink> {
+        let default_api_key: Arc<str> = Arc::from(self.default_api_key.clone().as_str());
+        let request_limits = self.request.unwrap_with(&DEFAULT_REQUEST_LIMITS);
+        let batch_timeout = self.batch.timeout_secs.map(Duration::from_secs);
+
+        let service = ServiceBuilder::new()
+            .settings(request_limits, LogApiRetry)
+            .service(LogApiService::new(client, self.get_uri()));
+        let sink = LogSink::new(service, cx)
+            .batch_timeout(batch_timeout)
+            .encoding(self.encoding.clone())
+            .default_api_key(default_api_key)
+            .compression(self.compression.unwrap_or_default())
+            .log_schema(vector_core::config::log_schema())
+            .build();
+
+        Ok(VectorSink::Stream(Box::new(sink)))
+    }
+
+    pub fn build_healthcheck(&self, client: HttpClient) -> crate::Result<Healthcheck> {
+        let healthcheck = healthcheck(client, self.get_uri(), self.default_api_key.clone()).boxed();
+        Ok(healthcheck)
+    }
 
+    pub fn create_client(&self, proxy: &ProxyConfig) -> crate::Result<HttpClient> {
         let tls_settings = MaybeTlsSettings::from_config(
             &Some(self.tls.clone().unwrap_or_else(TlsConfig::enabled)),
             false,
         )?;
-
-        let client = HttpClient::new(tls_settings, cx.proxy())?;
-        let healthcheck = healthcheck(
-            service.clone(),
-            client.clone(),
-            self.default_api_key.clone(),
-        )
-        .boxed();
-        let sink = PartitionHttpSink::new(
-            service,
-            PartitionBuffer::new(batch),
-            request_settings,
-            timeout,
-            client,
-            cx.acker(),
-        )
-        .sink_map_err(|error| error!(message = "Fatal datadog_logs text sink error.", %error));
-        let sink = VectorSink::Sink(Box::new(sink));
-
-        Ok((sink, healthcheck))
+        Ok(HttpClient::new(tls_settings, proxy)?)
     }
 }
 
 #[async_trait::async_trait]
 #[typetag::serde(name = "datadog_logs")]
 impl SinkConfig for DatadogLogsConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let batch_settings = self.batch_settings()?;
-        let service = service::Service::builder()
-            .encoding(self.encoding.clone())
-            .compression(self.compression.unwrap_or_default())
-            .uri(self.get_uri())
-            .default_api_key(Arc::from(self.default_api_key.clone()))
-            .log_schema(vector_core::config::log_schema())
-            .build();
-        self.build_sink(
-            cx,
-            service,
-            JsonArrayBuffer::new(batch_settings.size),
-            batch_settings.timeout,
-        )
+        let client = self.create_client(&cx.proxy)?;
+        let healthcheck = self.build_healthcheck(client.clone())?;
+        let sink = self.build_processor(client, cx)?;
+        Ok((sink, healthcheck))
     }
 
     fn input_type(&self) -> DataType {

@@ -1,30 +1,20 @@
-use super::ApiKey;
-use crate::{
-    http::HttpClient,
-    sinks::util::{http::HttpSink, PartitionInnerBuffer},
-};
-use http::StatusCode;
+use crate::http::HttpClient;
+use http::{Request, Response, StatusCode, Uri};
 use hyper::body::Body;
-use std::sync::Arc;
 
 /// The healthcheck is performed by sending an empty request to Datadog and
 /// checking the return.
-pub async fn healthcheck<T, O>(sink: T, client: HttpClient, api_key: String) -> crate::Result<()>
-where
-    T: HttpSink<Output = PartitionInnerBuffer<Vec<O>, ApiKey>>,
-{
-    let req = sink
-        .build_request(PartitionInnerBuffer::new(
-            Vec::with_capacity(0),
-            Arc::from(api_key),
-        ))
-        .await?
-        .map(Body::from);
+pub async fn healthcheck(client: HttpClient, uri: Uri, api_key: String) -> crate::Result<()> {
+    let body = vec![];
+    let request: Request<Body> = Request::post(uri)
+        .header("Content-Type", "application/json")
+        .header("DD-API-KEY", &api_key[..])
+        .header("Content-Length", body.len())
+        .body(Body::from(body))?;
+    let response: Response<Body> = client.send(request).await?;
 
-    let res = client.send(req).await?;
-
-    let status = res.status();
-    let body = hyper::body::to_bytes(res.into_body()).await?;
+    let status = response.status();
+    let body = hyper::body::to_bytes(response.into_body()).await?;
 
     match status {
         StatusCode::OK => Ok(()),

@@ -1,11 +1,27 @@
-use super::ApiKey;
+//! The Datadog Logs [`VectorSink`]
+//!
+//! This module contains the [`VectorSink`] instance that is responsible for
+//! taking a stream of [`Event`] instances and getting them flung out to the
+//! Datadog Log API. The log API is relatively generous in terms of its
+//! constraints, excepting that:
+//!
+//!   * a 'payload' is comprised of no more than 1,000 array members
+//!   * a 'payload' may not be more than 5Mb in size, uncompressed and
+//!   * a 'payload' may not mix API keys
+//!
+//! Otherwise per [the
+//! docs](https://docs.datadoghq.com/api/latest/logs/#send-logs) there aren't
+//! other major constraints we have to follow in this implementation. The sink
+//! is careful to always send the maximum payload size excepting where we
+//! violate the size constraint.
 
 #[cfg(test)]
 mod tests;
 
 mod config;
 mod healthcheck;
 mod service;
+mod sink;
 
 use crate::config::SinkDescription;
 use crate::sinks::datadog::logs::config::DatadogLogsConfig;