feat(opensearch sink): add first-class OpenSearch sink

Cargo.toml

@@ -930,6 +930,7 @@ sinks-logs = [
   "sinks-nats",
   "sinks-new_relic",
   "sinks-new_relic_logs",
+  "sinks-opensearch",
   "sinks-opentelemetry",
   "sinks-papertrail",
   "sinks-postgres",
@@ -1001,6 +1002,7 @@ sinks-mqtt = ["dep:rumqttc"]
 sinks-nats = ["dep:async-nats", "dep:nkeys"]
 sinks-new_relic_logs = ["sinks-http"]
 sinks-new_relic = []
+sinks-opensearch = ["sinks-elasticsearch"]
 sinks-opentelemetry = ["sinks-http", "codecs-opentelemetry"]
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["dep:base64", "dep:prost", "vector-lib/prometheus"]
@@ -1053,6 +1055,7 @@ all-integration-tests = [
   "mqtt-integration-tests",
   "nats-integration-tests",
   "nginx-integration-tests",
+  "opensearch-integration-tests",
   "opentelemetry-integration-tests",
   "postgresql_metrics-integration-tests",
   "postgres_sink-integration-tests",
@@ -1106,6 +1109,7 @@ datadog-traces-integration-tests = ["sources-datadog_agent", "sinks-datadog_trac
 docker-logs-integration-tests = ["sources-docker_logs"]
 doris-integration-tests = ["sinks-doris"]
 es-integration-tests = ["sinks-elasticsearch", "aws-core"]
+opensearch-integration-tests = ["es-integration-tests", "sinks-opensearch"]
 eventstoredb_metrics-integration-tests = ["sources-eventstoredb_metrics"]
 fluent-integration-tests = ["docker", "sources-fluent"]
 gcp-cloud-storage-integration-tests = ["sinks-gcp"]

changelog.d/11738_opensearch_sink.feature.md

@@ -0,0 +1,6 @@
+Add an `opensearch` sink for self-managed OpenSearch, Amazon OpenSearch Service,
+and Amazon OpenSearch Serverless. It shares the Elasticsearch sink's configuration
+options and delivery behavior. Automatic API version detection now recognizes
+OpenSearch in both sinks, so an explicit `api_version: v7` is no longer needed.
+
+authors: pront

src/sinks/elasticsearch/common.rs

@@ -394,6 +394,7 @@ async fn get_version(
     #[derive(Deserialize)]
     struct Version {
         number: Option<String>,
+        distribution: Option<String>,
     }
     #[derive(Deserialize)]
     struct ResponsePayload {
@@ -417,6 +418,11 @@ async fn get_version(
     let mut body = body.collect().await?.aggregate();
     let body = body.copy_to_bytes(body.remaining());
     let ResponsePayload { version } = serde_json::from_slice(&body)?;
+    // OpenSearch's own major version is unrelated to the Elasticsearch API version.
+    // Its bulk API is compatible with Elasticsearch 7 and must not include `_type`.
+    if version.as_ref().and_then(|v| v.distribution.as_deref()) == Some("opensearch") {
+        return Ok(7);
+    }
     if let Some(version) = version
         && let Some(number) = version.number
     {

src/sinks/elasticsearch/integration_tests.rs

@@ -15,10 +15,10 @@ use vrl::event_path;
 use super::{config::DATA_STREAM_TIMESTAMP_KEY, *};
 use crate::{
     aws::{ImdsAuthentication, RegionOrEndpoint},
-    config::{ProxyConfig, SinkConfig, SinkContext},
+    config::{BoxedSink, ProxyConfig, SinkContext},
     http::{HttpClient, ParameterValue, QueryParameterValue},
     sinks::{
-        HealthcheckError,
+        Healthcheck, HealthcheckError, VectorSink,
         util::{BatchConfig, Compression, HttpEndpoint, SinkBatchSettings, auth::Auth},
     },
     template::Template,
@@ -33,6 +33,19 @@ use crate::{
     tls::{self, TlsConfig},
 };
 
+// Exercise the registered sink type, including configuration deserialization, in both suites.
+async fn build_sink(
+    config: &ElasticsearchConfig,
+    cx: SinkContext,
+) -> crate::Result<(VectorSink, Healthcheck)> {
+    let mut value = serde_json::to_value(config)?;
+    value["type"] = std::env::var("ELASTICSEARCH_SINK_TYPE")
+        .unwrap_or_else(|_| "elasticsearch".into())
+        .into();
+    let config: BoxedSink = serde_json::from_value(value)?;
+    config.build(cx).await
+}
+
 fn http_username() -> String {
     std::env::var("ELASTICSEARCH_USERNAME").unwrap_or_else(|_| "elastic".into())
 }
@@ -242,7 +255,7 @@ async fn structures_events_correctly() {
     let base_url = common.base_url.clone();
 
     let cx = SinkContext::default();
-    let (sink, _hc) = config.build(cx.clone()).await.unwrap();
+    let (sink, _hc) = build_sink(&config, cx.clone()).await.unwrap();
 
     let (batch, mut receiver) = BatchNotifier::new_with_receiver();
     let mut input_event = LogEvent::from("raw log line").with_batch_notifier(&batch);
@@ -393,6 +406,46 @@ async fn insert_events_over_http() {
     .await;
 }
 
+#[tokio::test]
+async fn insert_events_with_auto_api_version_over_http() {
+    trace_init();
+    run_insert_tests(
+        ElasticsearchConfig {
+            endpoints: vec![http_server()],
+            api_version: ElasticsearchApiVersion::Auto,
+            batch: batch_settings(),
+            ..Default::default()
+        },
+        TestType::Normal,
+        BatchStatus::Delivered,
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn insert_events_with_auto_api_version_over_https() {
+    trace_init();
+    run_insert_tests(
+        ElasticsearchConfig {
+            endpoints: vec![https_server()],
+            auth: Some(ElasticsearchAuthConfig::Basic {
+                user: http_username(),
+                password: http_password().into(),
+            }),
+            tls: Some(TlsConfig {
+                ca_file: Some(tls::TEST_PEM_CA_PATH.into()),
+                ..Default::default()
+            }),
+            api_version: ElasticsearchApiVersion::Auto,
+            batch: batch_settings(),
+            ..Default::default()
+        },
+        TestType::Normal,
+        BatchStatus::Delivered,
+    )
+    .await;
+}
+
 #[tokio::test]
 async fn insert_events_with_data_volume() {
     trace_init();
@@ -716,8 +769,7 @@ async fn run_insert_tests_with_config(
     let base_url = common.base_url.clone();
 
     let cx = SinkContext::default();
-    let (sink, healthcheck) = config
-        .build(cx.clone())
+    let (sink, healthcheck) = build_sink(config, cx.clone())
         .await
         .expect("Building config failed");
 
@@ -835,8 +887,7 @@ async fn run_insert_tests_with_multiple_endpoints(config: &ElasticsearchConfig)
         ElasticsearchMode::Bulk => config.bulk.index.to_string(),
     };
 
-    let (sink, healthcheck) = config
-        .build(cx.clone())
+    let (sink, healthcheck) = build_sink(config, cx.clone())
         .await
         .expect("Building config failed");
 

src/sinks/elasticsearch/tests.rs

@@ -782,3 +782,45 @@ async fn test_parse_config_with_conflicting_auth() {
     // Should fail due to auth being specified in both places
     assert!(result.is_err());
 }
+
+#[tokio::test]
+async fn auto_version_distinguishes_opensearch_from_elasticsearch() {
+    use hyper::Body;
+
+    use crate::{sinks::util::test::build_test_server_generic, test_util::addr::next_addr};
+
+    for (version, suppress_type_name) in [
+        (serde_json::json!({"number": "6.8.23"}), false),
+        (serde_json::json!({"number": "7.17.0"}), true),
+        (serde_json::json!({"number": "8.0.0"}), true),
+        (
+            serde_json::json!({"number": "1.3.0", "distribution": "opensearch"}),
+            true,
+        ),
+        (
+            serde_json::json!({"number": "2.19.0", "distribution": "opensearch"}),
+            true,
+        ),
+        (
+            serde_json::json!({"number": "3.6.0", "distribution": "opensearch"}),
+            true,
+        ),
+    ] {
+        let (_guard, addr) = next_addr();
+        let body = serde_json::json!({"version": version}).to_string();
+        let (_rx, _trigger, server) =
+            build_test_server_generic(addr, move || http::Response::new(Body::from(body.clone())));
+        let server = tokio::spawn(server);
+        let config = ElasticsearchConfig {
+            endpoints: vec![HttpEndpoint::parse(&format!("http://{addr}")).unwrap()],
+            api_version: ElasticsearchApiVersion::Auto,
+            ..Default::default()
+        };
+        let common = ElasticsearchCommon::parse_single(&config).await.unwrap();
+        assert_eq!(
+            common.request_builder.encoder.suppress_type_name, suppress_type_name,
+            "unexpected type handling for {version}"
+        );
+        server.abort();
+    }
+}

src/sinks/mod.rs

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

src/sinks/opensearch.rs

@@ -0,0 +1,150 @@
+use vector_lib::configurable::configurable_component;
+
+use crate::{
+    config::{
+        AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext, ValidatedSink,
+    },
+    sinks::{Healthcheck, VectorSink, elasticsearch::ElasticsearchConfig},
+    template::ConfinementConfig,
+};
+
+/// Configuration for the `opensearch` sink.
+///
+/// Uses the Elasticsearch-compatible bulk API supported by OpenSearch. Configuration options
+/// and defaults are shared with the `elasticsearch` sink, including automatic API version
+/// detection and support for Amazon OpenSearch Service and OpenSearch Serverless.
+#[configurable_component(sink("opensearch", "Index observability events in OpenSearch."))]
+#[derive(Clone, Debug, Default)]
+pub struct OpenSearchConfig(ElasticsearchConfig);
+
+impl GenerateConfig for OpenSearchConfig {
+    fn generate_config() -> serde_json::Value {
+        ElasticsearchConfig::generate_config()
+    }
+}
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "opensearch")]
+impl SinkConfig for OpenSearchConfig {
+    fn confinement_config(&self) -> Option<&ConfinementConfig> {
+        self.0.confinement_config()
+    }
+
+    fn input(&self) -> Input {
+        self.0.input()
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        self.0.acknowledgements()
+    }
+}
+
+#[async_trait::async_trait]
+impl ValidatedSink for OpenSearchConfig {
+    type Validated = <ElasticsearchConfig as ValidatedSink>::Validated;
+
+    fn validate(&self) -> crate::Result<Self::Validated> {
+        self.0.validate()
+    }
+
+    async fn build(
+        &self,
+        validated: &Self::Validated,
+        cx: SinkContext,
+    ) -> crate::Result<(VectorSink, Healthcheck)> {
+        ValidatedSink::build(&self.0, validated, cx).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        config::BoxedSink,
+        sinks::elasticsearch::{ElasticsearchApiVersion, OpenSearchServiceType},
+    };
+
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<OpenSearchConfig>();
+    }
+
+    #[test]
+    fn config_roundtrip() {
+        let config: BoxedSink = serde_yaml::from_str(
+            r"
+            type: opensearch
+            endpoints: [http://localhost:9200]
+            bulk:
+              index: logs-%Y.%m.%d
+            acknowledgements:
+              enabled: true
+            ",
+        )
+        .unwrap();
+        assert!(config.acknowledgements().enabled());
+        assert!(config.confinement_config().is_some());
+        config.validate_dyn().unwrap();
+
+        let value = serde_json::to_value(&config).unwrap();
+        assert_eq!(value["type"], "opensearch");
+        assert_eq!(value["bulk"]["index"], "logs-%Y.%m.%d");
+        let roundtrip: BoxedSink = serde_json::from_value(value.clone()).unwrap();
+        assert_eq!(serde_json::to_value(roundtrip).unwrap(), value);
+    }
+
+    #[test]
+    fn defaults_to_managed_with_automatic_api_detection() {
+        let config: OpenSearchConfig =
+            serde_yaml::from_str("endpoints: [http://localhost:9200]").unwrap();
+        assert_eq!(config.0.api_version, ElasticsearchApiVersion::Auto);
+        assert_eq!(
+            config.0.opensearch_service_type,
+            OpenSearchServiceType::Managed
+        );
+    }
+
+    #[test]
+    fn rejects_invalid_config() {
+        for yaml in [
+            "type: opensearch",
+            "type: opensearch\nendpoints: [http://localhost:9200]\nbatch:\n  max_events: 0",
+            "type: opensearch\nendpoints: [http://localhost:9200]\nbulk:\n  index: '{{ index }}'",
+            "type: opensearch\nendpoints: [http://localhost:9200]\nopensearch_service_type: serverless",
+        ] {
+            let config: BoxedSink = serde_yaml::from_str(yaml).unwrap();
+            assert!(config.validate_dyn().is_err());
+        }
+    }
+
+    #[test]
+    fn rejects_unknown_options() {
+        assert!(
+            serde_yaml::from_str::<BoxedSink>(
+                "type: opensearch\nendpoints: [http://localhost:9200]\nunknown_option: true"
+            )
+            .is_err()
+        );
+    }
+
+    #[cfg(feature = "aws-core")]
+    #[test]
+    fn serverless_requires_aws_auth_and_automatic_api_detection() {
+        let yaml = r"
+            type: opensearch
+            endpoints: [https://collection.us-east-1.aoss.amazonaws.com]
+            opensearch_service_type: serverless
+            auth:
+              strategy: aws
+            aws:
+              region: us-east-1
+        ";
+        let config: BoxedSink = serde_yaml::from_str(yaml).unwrap();
+        config.validate_dyn().unwrap();
+
+        let mut value = serde_json::to_value(config).unwrap();
+        value["api_version"] = serde_json::json!("v7");
+        let config: BoxedSink = serde_json::from_value(value).unwrap();
+        assert!(config.validate_dyn().is_err());
+    }
+}

tests/integration/opensearch/config/test.yaml

@@ -1,9 +1,10 @@
 features:
-  - es-integration-tests
+  - opensearch-integration-tests
 
 test_filter: "::elasticsearch::integration_tests::"
 
 env:
+  ELASTICSEARCH_SINK_TYPE: opensearch
   AWS_ACCESS_KEY_ID: dummy
   AWS_SECRET_ACCESS_KEY: dummy
   ELASTICSEARCH_AWS_ADDRESS: http://opensearch:9200
@@ -20,6 +21,7 @@ matrix:
 # expressions are evaluated using https://github.com/micromatch/picomatch
 paths:
   - "src/sinks/elasticsearch/**"
+  - "src/sinks/opensearch.rs"
   - "src/sinks/util/**"
   - "tests/data/ca/**"
   - "tests/integration/opensearch/**"