feat(new sink): add Apache Doris sink support

Cargo.toml

@@ -411,7 +411,7 @@ seahash = { version = "4.1.0", default-features = false }
 smallvec = { version = "1", default-features = false, features = ["union", "serde"] }
 snap = { version = "1.1.1", default-features = false }
 socket2.workspace = true
-sqlx = { version = "0.8.6", default-features = false, features = ["derive", "postgres", "chrono", "runtime-tokio"], optional = true }
+sqlx = { version = "0.8.6", default-features = false, features = ["derive", "postgres", "mysql", "chrono", "runtime-tokio"], optional=true }
 stream-cancel = { version = "0.8.2", default-features = false }
 strip-ansi-escapes = { version = "0.2.1", default-features = false }
 syslog = { version = "6.1.1", default-features = false, optional = true }
@@ -436,6 +436,7 @@ heim = { git = "https://github.com/vectordotdev/heim.git", branch = "update-nix"
 mlua = { version = "0.10.5", default-features = false, features = ["lua54", "send", "vendored", "macros"], optional = true }
 sysinfo = "0.37.2"
 byteorder = "1.5.0"
+reqwest = { version = "0.11", features = ["blocking"] }
 
 [target.'cfg(windows)'.dependencies]
 windows-service = "0.8.0"
@@ -791,6 +792,7 @@ sinks-logs = [
   "sinks-datadog_events",
   "sinks-datadog_logs",
   "sinks-datadog_traces",
+  "sinks-doris",
   "sinks-elasticsearch",
   "sinks-file",
   "sinks-gcp",
@@ -858,6 +860,9 @@ sinks-datadog_events = []
 sinks-datadog_logs = []
 sinks-datadog_metrics = ["protobuf-build", "dep:prost", "dep:prost-reflect"]
 sinks-datadog_traces = ["protobuf-build", "dep:prost", "dep:rmpv", "dep:rmp-serde", "dep:serde_bytes"]
+sinks-doris = [
+  "dep:sqlx"
+]
 sinks-elasticsearch = ["transforms-metric_to_log"]
 sinks-file = ["dep:async-compression"]
 sinks-gcp = ["sinks-gcp-chronicle", "dep:base64", "gcp"]
@@ -911,6 +916,7 @@ all-integration-tests = [
   "datadog-traces-integration-tests",
   "dnstap-integration-tests",
   "docker-logs-integration-tests",
+  "doris-integration-tests",
   "es-integration-tests",
   "eventstoredb_metrics-integration-tests",
   "fluent-integration-tests",
@@ -976,6 +982,9 @@ datadog-logs-integration-tests = ["sinks-datadog_logs"]
 datadog-metrics-integration-tests = ["sinks-datadog_metrics", "dep:prost"]
 datadog-traces-integration-tests = ["sources-datadog_agent", "sinks-datadog_traces", "axum/tokio"]
 docker-logs-integration-tests = ["sources-docker_logs", "unix"]
+doris-integration-tests = [
+  "dep:sqlx"
+]
 es-integration-tests = ["sinks-elasticsearch", "aws-core"]
 eventstoredb_metrics-integration-tests = ["sources-eventstoredb_metrics"]
 fluent-integration-tests = ["docker", "sources-fluent"]

changelog.d/doris_sink_support.feature.md

@@ -0,0 +1,3 @@
+Add new Apache Doris sink for sending log data to Apache Doris databases using the Stream Load API. The sink supports configurable batching, custom HTTP headers for Doris-specific options, authentication, rate limiting, adaptive concurrency control, and includes comprehensive health checks.
+
+authors: bingquanzhao

scripts/integration/doris/compose.yaml

@@ -0,0 +1,13 @@
+version: "3"
+services:
+  doris:
+    image: docker.io/yagagagaga/doris-standalone:${CONFIG_VERSION}
+    container_name: doris
+    restart: unless-stopped
+    privileged: true
+    ports:
+      - "8030"
+      - "9030"
+      - "8040"
+    environment:
+      - SKIP_CHECK_ULIMIT=true

scripts/integration/doris/test.yaml

@@ -0,0 +1,17 @@
+features:
+- sinks-doris
+
+test_filter: '::doris::integration_test::'
+
+env:
+  DORIS_ADDRESS: http://doris:8030
+
+matrix:
+  version: ['2.1.7']
+
+# changes to these files/paths will invoke the integration test in CI
+# expressions are evaluated using https://github.com/micromatch/picomatch
+paths:
+- "src/sinks/doris/**"
+- "src/sinks/util/**"
+- "scripts/integration/doris/**"

src/internal_events/doris.rs

@@ -0,0 +1,44 @@
+use metrics::counter;
+use vector_lib::{NamedInternalEvent, internal_event::InternalEvent};
+
+/// Emitted when rows are successfully loaded into Doris.
+#[derive(Debug, NamedInternalEvent)]
+pub struct DorisRowsLoaded {
+    pub loaded_rows: i64,
+    pub load_bytes: i64,
+}
+
+impl InternalEvent for DorisRowsLoaded {
+    fn emit(self) {
+        trace!(
+            message = "Doris rows loaded successfully.",
+            loaded_rows = %self.loaded_rows,
+            load_bytes = %self.load_bytes,
+            internal_log_rate_limit = true
+        );
+
+        // Record the number of rows loaded
+        counter!("doris_rows_loaded_total").increment(self.loaded_rows as u64);
+
+        // Record the number of bytes loaded
+        counter!("doris_bytes_loaded_total").increment(self.load_bytes as u64);
+    }
+}
+
+/// Emitted when rows are filtered by Doris during loading.
+#[derive(Debug, NamedInternalEvent)]
+pub struct DorisRowsFiltered {
+    pub filtered_rows: i64,
+}
+
+impl InternalEvent for DorisRowsFiltered {
+    fn emit(self) {
+        warn!(
+            message = "Doris rows filtered during loading.",
+            filtered_rows = %self.filtered_rows,
+            internal_log_rate_limit = true
+        );
+
+        counter!("doris_rows_filtered_total").increment(self.filtered_rows as u64);
+    }
+}

src/internal_events/mod.rs

@@ -45,6 +45,8 @@ mod demo_logs;
 mod dnstap;
 #[cfg(feature = "sources-docker_logs")]
 mod docker_logs;
+#[cfg(feature = "sinks-doris")]
+mod doris;
 mod encoding_transcode;
 #[cfg(feature = "sources-eventstoredb_metrics")]
 mod eventstoredb_metrics;
@@ -197,6 +199,8 @@ pub(crate) use self::demo_logs::*;
 pub(crate) use self::dnstap::*;
 #[cfg(feature = "sources-docker_logs")]
 pub(crate) use self::docker_logs::*;
+#[cfg(feature = "sinks-doris")]
+pub(crate) use self::doris::*;
 #[cfg(feature = "sources-eventstoredb_metrics")]
 pub(crate) use self::eventstoredb_metrics::*;
 #[cfg(feature = "sources-exec")]