slbotbm commented on code in PR #3873: URL: https://github.com/apache/iggy/pull/3873#discussion_r3839785971
########## core/connectors/sinks/opensearch_sink/src/lib.rs: ########## @@ -0,0 +1,2903 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use bytes::{BufMut, Bytes, BytesMut}; +use iggy_common::{HeaderKey, HeaderValue, IggyTimestamp, calculate_256}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, + convert::owned_value_to_serde_json, + retry::{exponential_backoff, is_transient_status, jitter, parse_duration}, + sink_connector, +}; +use opensearch::{ + BulkParts, OpenSearch, + auth::Credentials, + cluster::ClusterHealthParts, + http::{ + StatusCode, + transport::{SingleNodeConnectionPool, TransportBuilder}, + }, + indices::{IndicesCreateParts, IndicesExistsParts}, + params::Refresh, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use std::{ + collections::BTreeMap, + future::Future, + net::IpAddr, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; +use tokio::time::sleep; +use tracing::{debug, error, info, warn}; +use url::Url; + +sink_connector!(OpenSearchSink); + +const DEFAULT_CREATE_INDEX_IF_NOT_EXISTS: bool = true; +const DEFAULT_INCLUDE_METADATA: bool = true; +const DEFAULT_BATCH_SIZE: usize = 1000; +const DEFAULT_TIMEOUT: &str = "30s"; +const DEFAULT_RETRY_DELAY: &str = "500ms"; +const DEFAULT_MAX_RETRY_DELAY: &str = "5s"; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_MAX_OPEN_RETRIES: u32 = 5; +const ENCODING_BASE64: &str = "base64"; +const ENCODING_UTF8: &str = "utf8"; +const GENERATED_ID_PREFIX: &str = "iggy_"; +const INDEX_ALREADY_EXISTS_ERROR: &str = "resource_already_exists_exception"; + +/// OpenSearch rejects `_id` values longer than 512 bytes. Payload-supplied IDs +/// are checked before the batch is built because that rejection fails the whole +/// `_bulk` call with an `action_request_validation_exception` rather than the +/// one item: a single oversized ID would cost every document in its chunk. +const MAX_DOCUMENT_ID_BYTES: usize = 512; + +// No `Serialize`: nothing serializes this type, and the only in-tree helper for +// a `SecretString` field writes the credential in plaintext. +#[derive(Debug, Default, Deserialize)] +pub struct OpenSearchSinkConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub document_id_field: Option<String>, + pub create_index_if_not_exists: Option<bool>, + pub index_mapping: Option<Value>, + pub include_metadata: Option<bool>, + pub batch_size: Option<usize>, + pub timeout: Option<String>, + pub refresh: Option<Refresh>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub max_retry_delay: Option<String>, + pub max_open_retries: Option<u32>, + pub verbose_logging: Option<bool>, +} + +pub struct OpenSearchSink { + id: u32, + config: ResolvedOpenSearchSinkConfig, + client: Option<OpenSearch>, + invocations_count: AtomicU64, + documents_indexed: AtomicU64, + errors_count: AtomicU64, +} + +// `OpenSearch` derives `Debug` down through its `Transport`, and +// `opensearch::auth::Credentials::Basic` derives `Debug` on its raw +// `(String, String)` without redaction, so a derived `Debug` on this struct +// would print the Basic-auth password in plaintext once `client` is set. +impl std::fmt::Debug for OpenSearchSink { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OpenSearchSink") + .field("id", &self.id) + .field("config", &self.config) + .field("client", &self.client.is_some()) + .field("invocations_count", &self.invocations_count) + .field("documents_indexed", &self.documents_indexed) + .field("errors_count", &self.errors_count) + .finish() + } +} Review Comment: This hides the authenticated client and relies on `SecretString` for the dedicated password field, but it still prints `self.config.url` verbatim through `self.config`. Since embedded URL credentials are rejected only later in `open()`, formatting a newly constructed sink from `https://admin:hunter2@host` exposes `hunter2`. Let's redact or sanitize the URL in the resolved config's `Debug` output. ########## core/connectors/sinks/opensearch_sink/Cargo.toml: ########## @@ -0,0 +1,52 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "iggy_connector_opensearch_sink" +version = "0.5.0-edge.1" Review Comment: update to "0.5.0-edge.4" ########## core/connectors/sinks/opensearch_sink/src/lib.rs: ########## @@ -0,0 +1,2903 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use async_trait::async_trait; +use base64::{Engine as _, engine::general_purpose}; +use bytes::{BufMut, Bytes, BytesMut}; +use iggy_common::{HeaderKey, HeaderValue, IggyTimestamp, calculate_256}; +use iggy_connector_sdk::{ + ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata, + convert::owned_value_to_serde_json, + retry::{exponential_backoff, is_transient_status, jitter, parse_duration}, + sink_connector, +}; +use opensearch::{ + BulkParts, OpenSearch, + auth::Credentials, + cluster::ClusterHealthParts, + http::{ + StatusCode, + transport::{SingleNodeConnectionPool, TransportBuilder}, + }, + indices::{IndicesCreateParts, IndicesExistsParts}, + params::Refresh, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use std::{ + collections::BTreeMap, + future::Future, + net::IpAddr, + sync::atomic::{AtomicU64, Ordering}, + time::Duration, +}; +use tokio::time::sleep; +use tracing::{debug, error, info, warn}; +use url::Url; + +sink_connector!(OpenSearchSink); + +const DEFAULT_CREATE_INDEX_IF_NOT_EXISTS: bool = true; +const DEFAULT_INCLUDE_METADATA: bool = true; +const DEFAULT_BATCH_SIZE: usize = 1000; +const DEFAULT_TIMEOUT: &str = "30s"; +const DEFAULT_RETRY_DELAY: &str = "500ms"; +const DEFAULT_MAX_RETRY_DELAY: &str = "5s"; +const DEFAULT_MAX_RETRIES: u32 = 3; +const DEFAULT_MAX_OPEN_RETRIES: u32 = 5; +const ENCODING_BASE64: &str = "base64"; +const ENCODING_UTF8: &str = "utf8"; +const GENERATED_ID_PREFIX: &str = "iggy_"; +const INDEX_ALREADY_EXISTS_ERROR: &str = "resource_already_exists_exception"; + +/// OpenSearch rejects `_id` values longer than 512 bytes. Payload-supplied IDs +/// are checked before the batch is built because that rejection fails the whole +/// `_bulk` call with an `action_request_validation_exception` rather than the +/// one item: a single oversized ID would cost every document in its chunk. +const MAX_DOCUMENT_ID_BYTES: usize = 512; + +// No `Serialize`: nothing serializes this type, and the only in-tree helper for +// a `SecretString` field writes the credential in plaintext. +#[derive(Debug, Default, Deserialize)] +pub struct OpenSearchSinkConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub document_id_field: Option<String>, + pub create_index_if_not_exists: Option<bool>, + pub index_mapping: Option<Value>, + pub include_metadata: Option<bool>, + pub batch_size: Option<usize>, + pub timeout: Option<String>, + pub refresh: Option<Refresh>, + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub max_retry_delay: Option<String>, + pub max_open_retries: Option<u32>, + pub verbose_logging: Option<bool>, +} + +pub struct OpenSearchSink { + id: u32, + config: ResolvedOpenSearchSinkConfig, + client: Option<OpenSearch>, + invocations_count: AtomicU64, + documents_indexed: AtomicU64, + errors_count: AtomicU64, +} + +// `OpenSearch` derives `Debug` down through its `Transport`, and +// `opensearch::auth::Credentials::Basic` derives `Debug` on its raw +// `(String, String)` without redaction, so a derived `Debug` on this struct +// would print the Basic-auth password in plaintext once `client` is set. +impl std::fmt::Debug for OpenSearchSink { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OpenSearchSink") + .field("id", &self.id) + .field("config", &self.config) + .field("client", &self.client.is_some()) + .field("invocations_count", &self.invocations_count) + .field("documents_indexed", &self.documents_indexed) + .field("errors_count", &self.errors_count) + .finish() + } +} + +#[derive(Debug)] +struct ResolvedOpenSearchSinkConfig { + url: String, + index: String, + username: Option<String>, + password: Option<SecretString>, + document_id_field: Option<String>, + create_index_if_not_exists: bool, + index_mapping: Option<Value>, + include_metadata: bool, + batch_size: usize, + timeout: Duration, + refresh: Option<Refresh>, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + max_open_retries: u32, + verbose_logging: bool, +} + +impl From<OpenSearchSinkConfig> for ResolvedOpenSearchSinkConfig { + fn from(config: OpenSearchSinkConfig) -> Self { + let mut retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + let mut max_retry_delay = + parse_duration(config.max_retry_delay.as_deref(), DEFAULT_MAX_RETRY_DELAY); + if retry_delay > max_retry_delay { + warn!( + "OpenSearch sink retry_delay ({:?}) exceeds max_retry_delay ({:?}). Swapping values.", + retry_delay, max_retry_delay + ); + std::mem::swap(&mut retry_delay, &mut max_retry_delay); + } + + Self { + url: config.url, + index: config.index.trim().to_string(), + username: trimmed_non_empty(config.username), + password: config + .password + .filter(|password| !is_blank_secret(password)), + document_id_field: trimmed_non_empty(config.document_id_field), + create_index_if_not_exists: config + .create_index_if_not_exists + .unwrap_or(DEFAULT_CREATE_INDEX_IF_NOT_EXISTS), + index_mapping: config.index_mapping, + include_metadata: config.include_metadata.unwrap_or(DEFAULT_INCLUDE_METADATA), + batch_size: config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1), + timeout: parse_duration(config.timeout.as_deref(), DEFAULT_TIMEOUT), + refresh: config.refresh, + max_retries: config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + retry_delay, + max_retry_delay, + max_open_retries: config.max_open_retries.unwrap_or(DEFAULT_MAX_OPEN_RETRIES), + verbose_logging: config.verbose_logging.unwrap_or(false), + } + } +} + +impl OpenSearchSink { + pub fn new(id: u32, config: OpenSearchSinkConfig) -> Self { + Self { + id, + config: config.into(), + client: None, + invocations_count: AtomicU64::new(0), + documents_indexed: AtomicU64::new(0), + errors_count: AtomicU64::new(0), + } + } + + fn validate_config(&self) -> Result<(), Error> { + if self.config.index.is_empty() { + return Err(Error::InvalidConfigValue( + "OpenSearch index cannot be empty".to_string(), + )); + } + + match (&self.config.username, &self.config.password) { + (Some(_), None) => Err(Error::InvalidConfigValue( + "OpenSearch username is set without a password".to_string(), + )), + (None, Some(_)) => Err(Error::InvalidConfigValue( + "OpenSearch password is set without a username".to_string(), + )), + _ => Ok(()), + } + } + + /// Takes the normalized URL rather than normalizing again, so + /// `normalize_url`'s warnings are emitted once per `open()`. + fn create_client(&self, normalized_url: &str) -> Result<OpenSearch, Error> { + warn_if_credentials_use_insecure_http( + &self.config.url, + normalized_url, + self.config.password.is_some(), + ); + + let url = Url::parse(normalized_url) + .map_err(|error| Error::Connection(format!("Invalid OpenSearch URL: {error}")))?; + // The transport defaults to no timeout at all, and unlike the per-call + // `tokio::time::timeout` guards this also covers reading response bodies. + let mut builder = + TransportBuilder::new(SingleNodeConnectionPool::new(url)).timeout(self.config.timeout); + if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { + builder = builder.auth(Credentials::Basic( + username.to_owned(), + password.expose_secret().to_owned(), + )); + } + + let transport = builder.build().map_err(|error| { + Error::Connection(format!("Failed to build OpenSearch transport: {error}")) + })?; + Ok(OpenSearch::new(transport)) + } + + /// Retries a transiently failing `open()`-time call, so one blip does not + /// park the connector in `Error` until an operator restarts it. + async fn retry_on_open<T, F, Fut>(&self, operation: &str, call: F) -> Result<T, Error> + where + F: Fn() -> Fut, + Fut: Future<Output = Result<T, Error>>, + { + let mut retries = 0u32; + + loop { + let failure = match call().await { + Ok(value) => return Ok(value), + Err(error) if is_transient_error(&error) => error.to_string(), + Err(error) => return Err(error), + }; + + if retries >= self.config.max_open_retries { + return Err(Error::InitError(format!( + "OpenSearch {operation} failed after {} retries ({failure})", + self.config.max_open_retries + ))); + } + + retries += 1; + self.sleep_before_retry(operation, retries, self.config.max_open_retries, &failure) + .await; + } + } + + async fn check_connectivity(&self, client: &OpenSearch) -> Result<(), Error> { + self.retry_on_open("health check", || self.cluster_health(client)) + .await + } + + async fn cluster_health(&self, client: &OpenSearch) -> Result<(), Error> { + let response = tokio::time::timeout( + self.config.timeout, + client.cluster().health(ClusterHealthParts::None).send(), + ) + .await + .map_err(|_| { + Error::HttpRequestFailed(format!( + "OpenSearch health check timed out after {:?}", + self.config.timeout + )) + })? + .map_err(|error| map_client_error("health check", error))?; + + let status = response.status_code(); + if status.is_success() { + return Ok(()); + } + + // A 403 only means cluster:monitor/health is missing, not that the cluster is down. + if status == StatusCode::FORBIDDEN { + warn!( + "OpenSearch health check returned 403: the configured user lacks the cluster-scoped cluster:monitor/health privilege. Treating the cluster as reachable; grant that privilege to restore the check." + ); + return Ok(()); + } + + let body = response.text().await.unwrap_or_default(); + Err(map_status_error("health check", status, &body)) + } + + async fn ensure_index_exists(&self, client: &OpenSearch) -> Result<(), Error> { + if self + .retry_on_open("index existence check", || self.index_exists(client)) + .await? + { + info!("OpenSearch index '{}' already exists", self.config.index); + return Ok(()); + } + + if !self.config.create_index_if_not_exists { + return Err(Error::InitError(format!( + "OpenSearch index '{}' does not exist and create_index_if_not_exists=false", + self.config.index + ))); + } + + self.retry_on_open("index creation", || self.create_index(client)) + .await + } + + async fn index_exists(&self, client: &OpenSearch) -> Result<bool, Error> { + let response = client + .indices() + .exists(IndicesExistsParts::Index(&[&self.config.index])) + .send() + .await + .map_err(|error| map_client_error("index existence check", error))?; + + let status = response.status_code(); + if status.is_success() { + return Ok(true); + } + + // A missing index comes back as a 404 response, not a transport error, + // so anything else is a genuine failure worth surfacing. + if status == StatusCode::NOT_FOUND { + return Ok(false); + } + + let body = response.text().await.unwrap_or_default(); + Err(map_status_error("index existence check", status, &body)) + } + + async fn create_index(&self, client: &OpenSearch) -> Result<(), Error> { + info!("Creating OpenSearch index '{}'", self.config.index); + + let indices = client.indices(); + let request = indices.create(IndicesCreateParts::Index(&self.config.index)); + let response = if let Some(mapping) = &self.config.index_mapping { + request.body(mapping.clone()).send().await + } else { + request.send().await + } + .map_err(|error| map_client_error("index creation", error))?; + + let status = response.status_code(); + if status.is_success() { + info!("Created OpenSearch index '{}'", self.config.index); + return Ok(()); + } + + let body = response.text().await.unwrap_or_default(); + // Another runtime instance winning the create race is not an error. + if is_index_already_exists_error(&body) { + info!( + "OpenSearch index '{}' was created concurrently", + self.config.index + ); + return Ok(()); + } + + Err(map_status_error("index creation", status, &body)) + } + + fn prepare_document( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + mut message: ConsumedMessage, + ) -> Result<PreparedDocument, Error> { + let payload = std::mem::replace(&mut message.payload, Payload::Raw(Vec::new())); + + let mut document = match payload { + Payload::Json(value) => document_from_json(owned_value_to_serde_json(&value)), + Payload::Raw(bytes) => document_from_raw(bytes), + Payload::Text(text) => Map::from_iter([ + ("text".to_string(), Value::String(text)), + ("data_type".to_string(), Value::String("text".to_string())), + ]), + _ => { + return Err(Error::InvalidRecordValue(format!( + "Unsupported payload format for OpenSearch sink: {}", + messages_metadata.schema + ))); + } + }; + + let id = match self.document_id_from_field(&document)? { + Some(id) => id, + None => generated_document_id( + topic_metadata, + messages_metadata, + message.offset, + message.id, + )?, + }; + + if self.config.include_metadata { + inject_metadata(&mut document, topic_metadata, messages_metadata, &message); + } + + Ok(PreparedDocument { + id, + document: Value::Object(document), + }) + } + + fn document_id_from_field( + &self, + document: &Map<String, Value>, + ) -> Result<Option<String>, Error> { + let Some(field) = self.config.document_id_field.as_deref() else { + return Ok(None); + }; + let Some(value) = document.get(field) else { + return Ok(None); + }; + + let id = match value { + Value::String(text) => text.clone(), + Value::Number(number) => number.to_string(), + Value::Bool(flag) => flag.to_string(), + Value::Null | Value::Array(_) | Value::Object(_) => { + return Err(Error::InvalidRecordValue(format!( + "OpenSearch document_id_field '{field}' must be a string, number, or boolean" + ))); + } + }; + + if id.is_empty() { + return Err(Error::InvalidRecordValue(format!( + "OpenSearch document_id_field '{field}' is empty" + ))); + } + if id.len() > MAX_DOCUMENT_ID_BYTES { + return Err(Error::InvalidRecordValue(format!( + "OpenSearch document_id_field '{field}' exceeds the {MAX_DOCUMENT_ID_BYTES} byte limit" + ))); + } + + Ok(Some(id)) + } + + async fn index_documents( + &self, + client: &OpenSearch, + documents: Vec<PreparedDocument>, + ) -> Result<usize, PartialIndexError> { + let total = documents.len(); + let mut indexed = 0usize; + let mut attempted = 0usize; + // The runtime commits the offset regardless of this error, so returning + // early would drop the remaining chunks for good. + let mut last_error: Option<Error> = None; + + for chunk in documents.chunks(self.config.batch_size) { + attempted += chunk.len(); + match self.index_chunk(client, chunk).await { + Ok(outcome) => { + indexed += outcome.indexed; + if let Some(error) = outcome.into_error(&self.config.index) { + last_error = Some(error); + } + } + Err(error) => last_error = Some(error), + } + debug!( + "OpenSearch sink with ID: {} indexed {}/{} documents", + self.id, attempted, total + ); + } + + match last_error { + Some(error) => Err(PartialIndexError { + indexed, + failed: total - indexed, + error, + }), + None => Ok(indexed), + } + } + + async fn index_chunk( + &self, + client: &OpenSearch, + documents: &[PreparedDocument], + ) -> Result<BulkOutcome, Error> { + // Only the documents OpenSearch has not yet accepted. A per-item + // transient rejection (429 under load, most commonly) shrinks this to + // just those documents rather than resending the whole chunk. + let mut pending: Vec<&PreparedDocument> = documents.iter().collect(); + // Rebuilt only when `pending` shrinks; `Bytes` clones are a refcount + // bump, so a whole-request retry resends without re-serializing. + let mut body = build_bulk_body(&self.config.index, &pending)?; + let mut outcome = BulkOutcome::default(); + let mut retries = 0u32; + + loop { + let request = client.bulk(BulkParts::None).body(vec![body.clone()]); + let request = match self.config.refresh { + Some(refresh) => request.refresh(refresh), + None => request, + }; + + let failure = match tokio::time::timeout(self.config.timeout, request.send()).await { + Ok(Ok(response)) => { + let status = response.status_code(); + if !status.is_success() { + let response_body = response.text().await.unwrap_or_default(); + let error = map_status_error("bulk request", status, &response_body); + if !is_transient_status(status) { + // Documents an earlier attempt in this same retry loop + // already indexed must still count; only `pending` (not + // yet accepted) is failed by this non-transient status. + outcome.merge(BulkOutcome { + indexed: 0, + failed: pending.len(), + transient: false, + first_failure: Some(error.to_string()), + }); + return Ok(outcome); + } + error.to_string() + } else { + // A bulk call answers 200 even when individual documents + // fail, so the per-item results decide the outcome. + match response.json::<Value>().await { + Ok(payload) => match parse_bulk_response(&payload, pending.len()) { + Ok(attempt) => { + let retry_set = documents_at(&pending, &attempt.retryable); + + if retry_set.is_empty() || retries >= self.config.max_retries { + outcome.merge(attempt.into_outcome()); + return Ok(outcome); + } + + // Only the permanent half is final; the retryable half + // is settled by a later attempt. + outcome.merge(BulkOutcome { + indexed: attempt.indexed, + failed: attempt.permanent_failed, + transient: false, + first_failure: attempt.first_permanent_failure, + }); + + let rejected = retry_set.len(); + pending = retry_set; + body = build_bulk_body(&self.config.index, &pending)?; + format!( + "{rejected} document(s) rejected with a transient status" + ) + } + // A 200 whose `items` array can't be trusted to cover + // every pending document leaves the true per-item + // outcome unknown. Retrying is safe because indexing + // is idempotent by `_id`, so this is treated as + // transient rather than silently counting the chunk + // as handled. + Err(error) => { + format!("failed to parse bulk response items: {error}") + } + }, + // A 200 with an unparsable body leaves the true + // per-item outcome unknown. Retrying is safe + // because indexing is idempotent by `_id`, so this + // is treated as transient rather than hard-failing + // a chunk OpenSearch may already have accepted. + Err(error) => { + format!("failed to parse bulk response body: {error}") + } + } + } + } + Ok(Err(error)) => { + if !is_transient_client_error(&error) { + let error = map_client_error("bulk request", error); + outcome.merge(BulkOutcome { + indexed: 0, + failed: pending.len(), + transient: false, + first_failure: Some(error.to_string()), + }); + return Ok(outcome); + } + error.to_string() + } + Err(_) => format!("timed out after {:?}", self.config.timeout), + }; + + // Reported through the outcome rather than `Err` so documents + // accepted by earlier attempts still count as indexed. + if retries >= self.config.max_retries { + outcome.merge(BulkOutcome { + indexed: 0, + failed: pending.len(), + transient: true, + first_failure: Some(format!( + "OpenSearch bulk request failed after {} retries ({failure})", + self.config.max_retries + )), + }); + return Ok(outcome); + } + + retries += 1; + self.sleep_before_retry("bulk request", retries, self.config.max_retries, &failure) + .await; + } + } + + /// Waits out the backoff for an already-incremented `retries`. The first + /// retry (`retries == 1`) uses attempt `0` so it sleeps `retry_delay` + /// itself rather than `retry_delay * 2`. Jitter can push the raw backoff + /// up to 20% above `max_retry_delay`, so the jittered result is clamped + /// back down to it. + async fn sleep_before_retry( + &self, + operation: &str, + retries: u32, + max_retries: u32, + failure: &str, + ) { + let delay = jitter(exponential_backoff( + self.config.retry_delay, + retries - 1, + self.config.max_retry_delay, + )) + .min(self.config.max_retry_delay); + warn!( + "OpenSearch {} failed (retry {}/{}): {}. Retrying in {:?}...", + operation, retries, max_retries, failure, delay + ); + sleep(delay).await; + } +} + +#[async_trait] +impl Sink for OpenSearchSink { + async fn open(&mut self) -> Result<(), Error> { + self.validate_config()?; + let normalized_url = normalize_url(&self.config.url)?; + info!( + "Opening OpenSearch sink connector with ID: {} for URL: {}, index: {}", + self.id, + sanitize_url_for_log(&normalized_url), + self.config.index + ); + + let client = self.create_client(&normalized_url)?; + self.check_connectivity(&client).await?; + self.ensure_index_exists(&client).await?; + self.client = Some(client); + + info!( + "Successfully opened OpenSearch sink connector with ID: {}", + self.id + ); + Ok(()) + } + + async fn consume( + &self, + topic_metadata: &TopicMetadata, + messages_metadata: MessagesMetadata, + messages: Vec<ConsumedMessage>, + ) -> Result<(), Error> { + let invocation = self.invocations_count.fetch_add(1, Ordering::Relaxed) + 1; + + if self.config.verbose_logging { + info!( + "OpenSearch sink with ID: {} received: {} messages, schema: {}, stream: {}, topic: {}, partition: {}, offset: {}, invocation: {}", + self.id, + messages.len(), + messages_metadata.schema, + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset, + invocation + ); + } else { + debug!( + "OpenSearch sink with ID: {} received: {} messages, schema: {}, stream: {}, topic: {}, partition: {}, offset: {}, invocation: {}", + self.id, + messages.len(), + messages_metadata.schema, + topic_metadata.stream, + topic_metadata.topic, + messages_metadata.partition_id, + messages_metadata.current_offset, + invocation + ); + } + + let client = self + .client + .as_ref() + .ok_or_else(|| Error::Connection("OpenSearch client not initialized".to_string()))?; + + let messages_count = messages.len(); + let mut documents = Vec::with_capacity(messages_count); + let mut invalid_records = 0usize; + let mut preparation_errors = 0usize; + for message in messages { + match self.prepare_document(topic_metadata, &messages_metadata, message) { + Ok(document) => documents.push(document), + Err(Error::InvalidRecordValue(reason)) => { + invalid_records += 1; + warn!( + "Dropping invalid OpenSearch sink record for connector ID: {}, reason: {}", + self.id, reason + ); + } + // A single message's preparation failing must not discard the + // documents already built from earlier messages in this batch. + Err(error) => { + preparation_errors += 1; + error!( + "Failed to prepare OpenSearch sink document for connector ID: {}, error: {}", + self.id, error + ); + } + } + } + if invalid_records > 0 || preparation_errors > 0 { + self.errors_count.fetch_add( + (invalid_records + preparation_errors) as u64, + Ordering::Relaxed, + ); + } + + if documents.is_empty() { + return Ok(()); + } + + match self.index_documents(client, documents).await { + Ok(indexed) => { + self.documents_indexed + .fetch_add(indexed as u64, Ordering::Relaxed); + if self.config.verbose_logging { + info!( + "Indexed {} of {} messages into OpenSearch index '{}'", + indexed, messages_count, self.config.index + ); + } else { + debug!( + "Indexed {} of {} messages into OpenSearch index '{}'", + indexed, messages_count, self.config.index + ); + } + Ok(()) + } + Err(partial) => { + self.documents_indexed + .fetch_add(partial.indexed as u64, Ordering::Relaxed); + self.errors_count + .fetch_add(partial.failed as u64, Ordering::Relaxed); + error!( + "Failed to index OpenSearch sink batch for connector ID: {}, index: {}, indexed: {}, failed: {}, error: {}", + self.id, self.config.index, partial.indexed, partial.failed, partial.error + ); + Err(partial.error) + } + } + } + + async fn close(&mut self) -> Result<(), Error> { + info!( + "OpenSearch sink connector with ID: {} is closing. Stats: {} invocations, {} documents indexed, {} errors", + self.id, + self.invocations_count.load(Ordering::Relaxed), + self.documents_indexed.load(Ordering::Relaxed), + self.errors_count.load(Ordering::Relaxed) + ); + + self.client = None; + info!("OpenSearch sink connector with ID: {} is closed.", self.id); + Ok(()) + } +} + +#[derive(Debug)] +struct PreparedDocument { + id: String, + document: Value, +} + +#[derive(Debug)] +struct PartialIndexError { + indexed: usize, + failed: usize, + error: Error, +} + +/// Aggregated result of every `_bulk` attempt made for one chunk, including +/// the per-item retries. +#[derive(Debug, Default, PartialEq, Eq)] +struct BulkOutcome { + indexed: usize, + failed: usize, + transient: bool, + first_failure: Option<String>, +} + +impl BulkOutcome { + /// Folds one attempt's totals in. The first failure seen across every + /// attempt for the chunk is the one reported. + fn merge(&mut self, other: BulkOutcome) { + self.indexed += other.indexed; + self.failed += other.failed; + self.transient |= other.transient; + self.first_failure = self.first_failure.take().or(other.first_failure); + } + + fn into_error(self, index: &str) -> Option<Error> { + let failure = self.first_failure?; + let message = format!( + "OpenSearch bulk indexing into '{index}' failed for {} of {} documents: {failure}", + self.failed, + self.failed + self.indexed + ); + Some(if self.transient { + Error::HttpRequestFailed(message) + } else { + Error::PermanentHttpError(message) + }) + } +} + +/// Per-item breakdown of a single `_bulk` call. OpenSearch answers 200 with a +/// per-item list echoed back in request order, so `retryable` holds positions +/// into the slice that was sent. +#[derive(Debug, Default, PartialEq, Eq)] +struct BulkAttempt { + indexed: usize, + permanent_failed: usize, + first_permanent_failure: Option<String>, + retryable: Vec<usize>, + first_retryable_failure: Option<String>, +} + +impl BulkAttempt { + /// Collapses one attempt into the aggregate shape, counting every + /// still-retryable item as failed. This is what a caller out of retries + /// reports. + fn into_outcome(self) -> BulkOutcome { + BulkOutcome { + indexed: self.indexed, + failed: self.permanent_failed + self.retryable.len(), + transient: !self.retryable.is_empty(), + first_failure: self + .first_permanent_failure + .or(self.first_retryable_failure), + } + } +} + +/// Serializes the `_bulk` NDJSON payload (alternating `index` action line and +/// document line per document) into one `Bytes` buffer. Takes references so a +/// retry can serialize just the rejected subset without cloning documents. +fn build_bulk_body(index: &str, documents: &[&PreparedDocument]) -> Result<Bytes, Error> { + let mut buffer = BytesMut::new(); + for document in documents { + serde_json::to_writer( + (&mut buffer).writer(), + &json!({ "index": { "_index": index, "_id": document.id } }), + ) + .map_err(|error| { + Error::Serialization(format!( + "Failed to serialize OpenSearch bulk action: {error}" + )) + })?; + buffer.put_u8(b'\n'); + serde_json::to_writer((&mut buffer).writer(), &document.document).map_err(|error| { + Error::Serialization(format!( + "Failed to serialize OpenSearch bulk document: {error}" + )) + })?; + buffer.put_u8(b'\n'); + } + Ok(buffer.freeze()) +} + +/// Maps server-echoed item positions back to the documents that were sent. +/// Out-of-range positions are dropped: a response carrying more items than +/// were sent would otherwise panic, and a panic crossing the plugin's +/// `extern "C"` boundary aborts the whole connectors runtime process. +fn documents_at<'a>( + pending: &[&'a PreparedDocument], + positions: &[usize], +) -> Vec<&'a PreparedDocument> { + positions + .iter() + .filter_map(|&position| pending.get(position).copied()) + .collect() +} + +/// Why a `_bulk` response's `items` array could not be trusted to reflect +/// what OpenSearch actually did with every pending document. +#[derive(Debug)] +enum BulkResponseError { + MissingItems, + ItemCountMismatch { expected: usize, actual: usize }, + MalformedItem { position: usize }, +} + +impl std::fmt::Display for BulkResponseError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::MissingItems => write!(f, "response has no `items` array"), + Self::ItemCountMismatch { expected, actual } => write!( + f, + "response `items` array has {actual} entries, expected {expected}" + ), + Self::MalformedItem { position } => { + write!(f, "item at position {position} has no recognizable result") + } + } + } +} + +/// Requires exactly one item per pending document. A missing, short, or +/// malformed `items` array leaves the true per-item outcome unknown, so the +/// whole response is rejected rather than silently under-accounted: a chunk +/// that OpenSearch answered with fewer or unparsable items must not be +/// treated as cleanly handled just because nothing came back marked failed. +fn parse_bulk_response( + response: &Value, + expected: usize, +) -> Result<BulkAttempt, BulkResponseError> { + let items = response + .get("items") + .and_then(Value::as_array) + .ok_or(BulkResponseError::MissingItems)?; + + if items.len() != expected { + return Err(BulkResponseError::ItemCountMismatch { + expected, + actual: items.len(), + }); + } + + // The top-level flag lets a clean batch skip the per-item scan entirely. + if !response + .get("errors") + .and_then(Value::as_bool) + .unwrap_or(true) + { + return Ok(BulkAttempt { + indexed: items.len(), + ..BulkAttempt::default() Review Comment: This fast path still accepts malformed item entries without validating them. For example, `{"errors":false,"items":[{}]}` passes the length check and is reported as one indexed document, even though the response contains no recognizable `index` result or successful status. That leaves the malformed-response fix incomplete, and the runtime will commit the offset for a document that was never actually accounted for. Please validate that every item contains the expected operation result and a 2xx status before returning success, and add an `errors: false` regression case with a malformed item. ########## core/connectors/sinks/opensearch_sink/Cargo.toml: ########## @@ -0,0 +1,52 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +[package] +name = "iggy_connector_opensearch_sink" +version = "0.5.0-edge.1" +description = "Iggy OpenSearch sink connector" +edition = "2024" +license = "Apache-2.0" +keywords = ["iggy", "messaging", "streaming", "opensearch", "sink"] +categories = ["command-line-utilities", "database", "network-programming"] +homepage = "https://iggy.apache.org"; +documentation = "https://iggy.apache.org/docs"; +repository = "https://github.com/apache/iggy"; +readme = "../../README.md" +publish = false + +[lib] +crate-type = ["cdylib", "lib"] + +[dependencies] +async-trait = { workspace = true } +base64 = { workspace = true } +bytes = { workspace = true } +iggy_common = { workspace = true } +iggy_connector_sdk = { workspace = true } +opensearch = { workspace = true } +secrecy = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +simd-json = { workspace = true } Review Comment: `simd-json` is used only by test fixtures inside the `#[cfg(test)]` module. It should be under `[dev-dependencies]`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
