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. The new sink selects an OpenSearch-compatible bulk
+API version by default, so an explicit `api_version: v7` is not needed.
+
+authors: pront

src/sinks/elasticsearch/common.rs

@@ -23,6 +23,22 @@ use crate::{
     transforms::metric_to_log::MetricToLog,
 };
 
+/// The backend selected by the sink constructor.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Distribution {
+    Elasticsearch,
+    OpenSearch,
+}
+
+impl Distribution {
+    pub(super) const fn component_name(self) -> &'static str {
+        match self {
+            Self::Elasticsearch => "elasticsearch",
+            Self::OpenSearch => "opensearch",
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct ElasticsearchCommon {
     pub base_url: String,
@@ -50,6 +66,7 @@ impl ElasticsearchCommon {
     #[allow(clippy::too_many_lines, reason = "Keep endpoint setup together.")]
     pub async fn parse_config(
         config: &ElasticsearchConfig,
+        distribution: Distribution,
         endpoint: &HttpEndpoint,
         proxy_config: &ProxyConfig,
         version: &mut Option<usize>,
@@ -69,7 +86,7 @@ impl ElasticsearchCommon {
 
         let base_url = uri.uri.to_string().trim_end_matches('/').to_owned();
 
-        let mode = config.common_mode()?;
+        let mode = config.common_mode(distribution.component_name())?;
 
         let tower_request = config.request.tower.into_settings();
 
@@ -157,6 +174,9 @@ impl ElasticsearchCommon {
                 ElasticsearchApiVersion::V6 => 6,
                 ElasticsearchApiVersion::V7 => 7,
                 ElasticsearchApiVersion::V8 => 8,
+                // OpenSearch's bulk API is compatible with Elasticsearch 7. Its own
+                // major version is unrelated, so no cluster-version probe is needed.
+                ElasticsearchApiVersion::Auto if distribution == Distribution::OpenSearch => 7,
                 ElasticsearchApiVersion::Auto => {
                     match get_version(
                         &base_url,
@@ -286,6 +306,7 @@ impl ElasticsearchCommon {
     /// or any endpoint fails [`Self::parse_config`].
     pub async fn parse_many(
         config: &ElasticsearchConfig,
+        distribution: Distribution,
         proxy_config: &ProxyConfig,
     ) -> crate::Result<Vec<Self>> {
         let mut version = None;
@@ -295,7 +316,8 @@ impl ElasticsearchCommon {
             );
             if config.endpoints.is_empty() {
                 Ok(vec![
-                    Self::parse_config(config, endpoint, proxy_config, &mut version).await?,
+                    Self::parse_config(config, distribution, endpoint, proxy_config, &mut version)
+                        .await?,
                 ])
             } else {
                 Err(ParseError::EndpointsExclusive.into())
@@ -305,8 +327,10 @@ impl ElasticsearchCommon {
         } else {
             let mut commons = Vec::new();
             for endpoint in &config.endpoints {
-                commons
-                    .push(Self::parse_config(config, endpoint, proxy_config, &mut version).await?);
+                commons.push(
+                    Self::parse_config(config, distribution, endpoint, proxy_config, &mut version)
+                        .await?,
+                );
             }
             Ok(commons)
         }
@@ -321,8 +345,12 @@ impl ElasticsearchCommon {
     /// Panics if the configuration contains more than one endpoint.
     #[cfg(test)]
     pub async fn parse_single(config: &ElasticsearchConfig) -> crate::Result<Self> {
-        let mut commons =
-            Self::parse_many(config, crate::config::SinkContext::default().proxy()).await?;
+        let mut commons = Self::parse_many(
+            config,
+            Distribution::Elasticsearch,
+            crate::config::SinkContext::default().proxy(),
+        )
+        .await?;
         assert_eq!(commons.len(), 1);
         Ok(commons.remove(0))
     }

src/sinks/elasticsearch/config.rs

@@ -21,7 +21,7 @@ use crate::{
     sinks::{
         Healthcheck, VectorSink,
         elasticsearch::{
-            ElasticsearchApiVersion, ElasticsearchAuthConfig, ElasticsearchCommon,
+            Distribution, ElasticsearchApiVersion, ElasticsearchAuthConfig, ElasticsearchCommon,
             ElasticsearchCommonMode, ElasticsearchMode, ParseError, VersionType,
             health::ElasticsearchHealthLogic,
             retry::ElasticsearchRetryLogic,
@@ -257,15 +257,18 @@ impl ElasticsearchConfig {
     ///
     /// # Errors
     /// Returns an error if an active routing template fails confinement validation.
-    pub fn common_mode(&self) -> crate::Result<ElasticsearchCommonMode> {
+    pub fn common_mode(
+        &self,
+        component_name: &'static str,
+    ) -> crate::Result<ElasticsearchCommonMode> {
         match self.mode {
             ElasticsearchMode::Bulk => Ok(ElasticsearchCommonMode::Bulk {
                 // Only `index` is a routing field, so it is the only bulk template that is confined.
                 // `action` and `version` are not routing values and render unconfined, hence their
                 // `UnconfinedTemplate` type.
                 index: self.bulk.index.clone().confine(
                     &self.confinement,
-                    Self::NAME,
+                    component_name,
                     "bulk.index",
                 )?,
                 template_fallback_index: self.bulk.template_fallback_index.clone(),
@@ -277,7 +280,7 @@ impl ElasticsearchConfig {
                 self.data_stream
                     .clone()
                     .unwrap_or_default()
-                    .confine(&self.confinement, Self::NAME)?,
+                    .confine(&self.confinement, component_name)?,
             )),
         }
     }
@@ -748,6 +751,24 @@ impl ValidatedSink for ElasticsearchConfig {
     type Validated = ValidatedElasticsearch;
 
     fn validate(&self) -> crate::Result<ValidatedElasticsearch> {
+        self.validate_with_distribution(Distribution::Elasticsearch)
+    }
+
+    async fn build(
+        &self,
+        validated: &ValidatedElasticsearch,
+        cx: SinkContext,
+    ) -> crate::Result<(VectorSink, Healthcheck)> {
+        self.build_with_distribution(validated, cx, Distribution::Elasticsearch)
+            .await
+    }
+}
+
+impl ElasticsearchConfig {
+    pub(crate) fn validate_with_distribution(
+        &self,
+        distribution: Distribution,
+    ) -> crate::Result<ValidatedElasticsearch> {
         // Mirror the pure endpoint-required/exclusive checks from
         // `ElasticsearchCommon::parse_many` so configs that deterministically
         // fail at build time are rejected here instead.
@@ -795,7 +816,7 @@ impl ValidatedSink for ElasticsearchConfig {
         // so unconfined `bulk.index` / `data_stream.*` templates are rejected
         // here. The confined mode itself is reconstructed during build (inside
         // `ElasticsearchCommon::parse_many`).
-        self.common_mode()?;
+        self.common_mode(distribution.component_name())?;
 
         // Mirror the pure batch-settings and versioning checks from
         // `ElasticsearchCommon::parse_config` so configs that deterministically
@@ -829,16 +850,17 @@ impl ValidatedSink for ElasticsearchConfig {
         })
     }
 
-    async fn build(
+    pub(crate) async fn build_with_distribution(
         &self,
         validated: &ValidatedElasticsearch,
         cx: SinkContext,
+        distribution: Distribution,
     ) -> crate::Result<(VectorSink, Healthcheck)> {
         // Confinement of the active mode's routing templates happens in
         // [`ElasticsearchConfig::common_mode`], which each `ElasticsearchCommon` calls while
         // parsing (which performs I/O: AWS credential resolution and API-version autodetection).
         // The pure request/health settings were resolved during `validate`.
-        let commons = ElasticsearchCommon::parse_many(self, cx.proxy()).await?;
+        let commons = ElasticsearchCommon::parse_many(self, distribution, cx.proxy()).await?;
         let common = commons[0].clone();
 
         let client = HttpClient::new(common.tls_settings.clone(), cx.proxy())?;

src/sinks/elasticsearch/integration_tests.rs

@@ -393,6 +393,89 @@ async fn insert_events_over_http() {
     .await;
 }
 
+#[cfg(feature = "opensearch-integration-tests")]
+mod opensearch {
+    use super::*;
+    use crate::config::BoxedSink;
+
+    #[tokio::test]
+    async fn insert_events_with_auto_api_version_over_http() {
+        insert_event(false).await;
+    }
+
+    #[tokio::test]
+    async fn insert_events_with_auto_api_version_over_https() {
+        insert_event(true).await;
+    }
+
+    async fn insert_event(secure: bool) {
+        trace_init();
+        let endpoint = if secure {
+            https_server()
+        } else {
+            http_server()
+        }
+        .to_string();
+        let endpoint = endpoint.trim_end_matches('/');
+        let index = gen_index().to_string();
+        let mut config = json!({
+            "type": "opensearch",
+            "endpoints": [endpoint],
+            "bulk": { "index": index },
+            "compression": "gzip",
+        });
+        if secure {
+            config["auth"] = json!({
+                "strategy": "basic",
+                "user": http_username(),
+                "password": http_password(),
+            });
+            config["tls"] = json!({ "ca_file": tls::TEST_PEM_CA_PATH });
+        }
+        // Omit api_version to exercise the constructor-selected OpenSearch API version.
+        let config: BoxedSink = serde_json::from_value(config).unwrap();
+        let (sink, healthcheck) = config.build(SinkContext::default()).await.unwrap();
+        healthcheck.await.unwrap();
+
+        let (batch, mut receiver) = BatchNotifier::new_with_receiver();
+        let event = LogEvent::from("OpenSearch integration test").with_batch_notifier(&batch);
+        drop(batch);
+        run_and_assert_sink_compliance(
+            sink,
+            stream::once(ready(Event::from(event))),
+            &HTTP_SINK_TAGS,
+        )
+        .await;
+        assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));
+
+        let client = create_http_client();
+        client
+            .post(format!("{endpoint}/{index}/_refresh"))
+            .basic_auth(http_username(), Some(http_password()))
+            .send()
+            .await
+            .unwrap()
+            .error_for_status()
+            .unwrap();
+        let response: Value = client
+            .get(format!("{endpoint}/{index}/_search"))
+            .basic_auth(http_username(), Some(http_password()))
+            .send()
+            .await
+            .unwrap()
+            .error_for_status()
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        assert_eq!(response["hits"]["total"]["value"], 1);
+        assert_eq!(
+            response["hits"]["hits"][0]["_source"]["message"],
+            "OpenSearch integration test"
+        );
+    }
+}
+
 #[tokio::test]
 async fn insert_events_with_data_volume() {
     trace_init();
@@ -823,7 +906,7 @@ async fn run_insert_tests_with_config(
 
 async fn run_insert_tests_with_multiple_endpoints(config: &ElasticsearchConfig) {
     let cx = SinkContext::default();
-    let commons = ElasticsearchCommon::parse_many(config, cx.proxy())
+    let commons = ElasticsearchCommon::parse_many(config, Distribution::Elasticsearch, cx.proxy())
         .await
         .expect("Config error");
     let index = match config.mode {

src/sinks/elasticsearch/mod.rs

@@ -311,7 +311,10 @@ impl ElasticsearchCommonMode {
 #[serde(deny_unknown_fields, rename_all = "snake_case")]
 #[derive(Default)]
 pub enum ElasticsearchApiVersion {
-    /// Auto-detect the API version.
+    /// Automatically select the API version.
+    ///
+    /// The `opensearch` sink selects an OpenSearch-compatible API version without querying the server.
+    /// The `elasticsearch` sink detects the version from the server.
     ///
     /// If the [cluster state version endpoint][es_version] isn't reachable, a warning is logged to
     /// stdout, and the version is assumed to be V6 if the `suppress_type_name` option is set to