feat(openobserve transform): Add OpenObserve as an officially supported sink

.github/actions/spelling/allow.txt

@@ -142,6 +142,7 @@ Nextbook
 Nextcloud
 OVH
 Odys
+openobserve
 Openpeak
 Oppo
 Ovi

Cargo.toml

@@ -709,6 +709,7 @@ sinks-logs = [
   "sinks-new_relic",
   "sinks-papertrail",
   "sinks-pulsar",
+  "sinks-openobserve",
   "sinks-redis",
   "sinks-sematext",
   "sinks-socket",
@@ -772,6 +773,7 @@ sinks-mqtt = ["dep:rumqttc"]
 sinks-nats = ["dep:async-nats", "dep:nkeys"]
 sinks-new_relic_logs = ["sinks-http"]
 sinks-new_relic = []
+sinks-openobserve = ["sinks-http"]
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["dep:base64", "dep:prost", "vector-lib/prometheus"]
 sinks-pulsar = ["dep:apache-avro", "dep:pulsar", "dep:lru"]
@@ -821,6 +823,7 @@ all-integration-tests = [
   "nats-integration-tests",
   "nginx-integration-tests",
   "opentelemetry-integration-tests",
+  "openobserve-integration-tests",
   "postgresql_metrics-integration-tests",
   "prometheus-integration-tests",
   "pulsar-integration-tests",

Makefile

@@ -367,7 +367,7 @@ test-integration: ## Runs all integration tests
 test-integration: test-integration-amqp test-integration-appsignal test-integration-aws test-integration-axiom test-integration-azure test-integration-chronicle test-integration-clickhouse
 test-integration: test-integration-databend test-integration-docker-logs test-integration-elasticsearch
 test-integration: test-integration-eventstoredb test-integration-fluent test-integration-gcp test-integration-greptimedb test-integration-humio test-integration-http-client test-integration-influxdb
-test-integration: test-integration-kafka test-integration-logstash test-integration-loki test-integration-mongodb test-integration-nats
+test-integration: test-integration-kafka test-integration-logstash test-integration-loki test-integration-mongodb test-integration-nats 
 test-integration: test-integration-nginx test-integration-opentelemetry test-integration-postgres test-integration-prometheus test-integration-pulsar
 test-integration: test-integration-redis test-integration-splunk test-integration-dnstap test-integration-datadog-agent test-integration-datadog-logs test-integration-e2e-datadog-logs
 test-integration: test-integration-datadog-traces test-integration-shutdown

src/sinks/mod.rs

@@ -82,6 +82,8 @@ pub mod nats;
 pub mod new_relic;
 #[cfg(feature = "sinks-webhdfs")]
 pub mod opendal_common;
+#[cfg(feature = "sinks-openobserve")]
+pub mod openobserve;
 #[cfg(feature = "sinks-papertrail")]
 pub mod papertrail;
 #[cfg(feature = "sinks-prometheus")]

src/sinks/openobserve.rs

@@ -0,0 +1,133 @@
+use http::Uri;
+use vector_lib::codecs::encoding::{FramingConfig, JsonSerializerConfig, SerializerConfig};
+use vector_lib::configurable::configurable_component;
+
+use crate::{
+    codecs::{EncodingConfig, EncodingConfigWithFraming},
+    config::{AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext},
+    http::{Auth, MaybeAuth},
+    sinks::{
+        http::config::{HttpMethod, HttpSinkConfig},
+        util::{
+            http::RequestConfig, BatchConfig, Compression, RealtimeSizeBasedDefaultBatchSettings,
+            UriSerde,
+        },
+        Healthcheck, VectorSink,
+    },
+    tls::TlsConfig,
+};
+
+/// Configuration for the `openobserve` sink.
+#[configurable_component(sink("openobserve", "Deliver log events to OpenObserve."))]
+#[derive(Clone, Debug)]
+pub struct OpenObserveConfig {
+    /// The OpenObserve endpoint to send data to.
+    #[serde(default = "default_endpoint")]
+    #[configurable(metadata(docs::examples = "http://localhost:5080/api/default/default/_json"))]
+    uri: UriSerde,
+
+    /// The user and password to authenticate with OpenObserve endpoint.
+    #[configurable(derived)]
+    auth: Option<Auth>,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    request: RequestConfig,
+
+    /// The compression algorithm to use.
+    #[configurable(derived)]
+    #[serde(default = "Compression::gzip_default")]
+    compression: Compression,
+
+    #[configurable(derived)]
+    encoding: EncodingConfig,
+
+    /// The batch settings for the sink.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub batch: BatchConfig<RealtimeSizeBasedDefaultBatchSettings>,
+
+    /// Controls how acknowledgements are handled for this sink.
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::is_default"
+    )]
+    acknowledgements: AcknowledgementsConfig,
+
+    /// The TLS settings for the connection.
+    ///
+    /// Optional, constrains TLS settings for this sink.
+    #[configurable(derived)]
+    tls: Option<TlsConfig>,
+}
+
+impl GenerateConfig for OpenObserveConfig {
+    fn generate_config() -> toml::Value {
+        toml::from_str(
+            r#"
+            uri = "http://localhost:5080/api/default/default/_json"
+            Auth = "user: [email protected], password: your_ingestion_password"
+            encoding.codec = "json"
+        "#,
+        )
+        .unwrap()
+    }
+}
+
+fn default_endpoint() -> UriSerde {
+    UriSerde {
+        uri: Uri::from_static("http://localhost:5080/api/default/default/_json"),
+        auth: None,
+    }
+}
+
+/// This sink wraps the Vector HTTP sink to provide official support for OpenObserve's  
+/// native HTTP ingest endpoint. By doing so, it maintains a distinct configuration for 
+/// the OpenObserve sink, separate from the Vector HTTP sink. This approach ensures  
+/// that future changes to OpenObserve's interface can be accommodated without impacting 
+/// the underlying Vector HTTP sink.  
+#[async_trait::async_trait]
+#[typetag::serde(name = "openobserve")]
+impl SinkConfig for OpenObserveConfig {
+    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let request = self.request.clone();
+        let http_sink_config = HttpSinkConfig {
+            uri: self.uri.clone(),
+            compression: self.compression,
+            auth: self.auth.choose_one(&self.uri.auth)?,
+            method: HttpMethod::Post,
+            tls: self.tls.clone(),
+            request,
+            acknowledgements: self.acknowledgements,
+            batch: self.batch,
+            headers: None,
+            encoding: EncodingConfigWithFraming::new(
+                Some(FramingConfig::Bytes),
+                SerializerConfig::Json(JsonSerializerConfig::default()),
+                self.encoding.transformer(),
+            ),
+            payload_prefix: "".into(), // Always newline delimited JSON
+            payload_suffix: "".into(), // Always newline delimited JSON
+        };
+
+        http_sink_config.build(cx).await
+    }
+
+    fn input(&self) -> Input {
+        Input::new(self.encoding.config().input_type() & DataType::Log)
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+}
+
+#[cfg(test)]
+mod test {
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<super::OpenObserveConfig>();
+    }
+}