Copilot commented on code in PR #4730: URL: https://github.com/apache/iggy/pull/4730#discussion_r4463451453
########## core/connectors/sources/opensearch_source/src/lib.rs: ########## @@ -0,0 +1,1043 @@ +// 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 iggy_common::{DateTime, HeaderKey, HeaderValue, Utc}; +use iggy_connector_sdk::retry::{CircuitBreaker, RetryPolicy, parse_duration, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use opensearch::{ + OpenSearch, SearchParts, + auth::Credentials, + http::{StatusCode, Url, response::Response, transport::TransportBuilder}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; +use tokio::{sync::Mutex, time::sleep}; +use tracing::{debug, error, info, warn}; + +mod retry; +use crate::retry::{ + DEFAULT_CB_COOL_DOWN, DEFAULT_CB_THRESHOLD, DEFAULT_MAX_OPEN_RETRIES, DEFAULT_MAX_RETRIES, + DEFAULT_OPEN_RETRY_MAX_DELAY, DEFAULT_RETRY_DELAY, DEFAULT_RETRY_MAX_DELAY, RequestFailure, +}; + +source_connector!(OpenSearchSource); + +const CONNECTOR_NAME: &str = "OpenSearch source"; +const DEFAULT_POLLING_INTERVAL: &str = "10s"; +const DEFAULT_BATCH_SIZE: usize = 100; +// OpenSearch rejects `size` above the default `index.max_result_window` with HTTP 400. +const MAX_BATCH_SIZE: usize = 10_000; +const DOCUMENT_ID_HEADER: &str = "opensearch.document_id"; +const INDEX_HEADER: &str = "opensearch.index"; +const MAX_HEADER_VALUE_LENGTH: usize = 255; +const MAX_LAST_ERROR_CHARS: usize = 512; +// What OpenSearch reports for a document missing the sort field when it sorts last. The sort +// asks for missing values first, so seeing this means that request option was not honored. +const MISSING_LAST_SORT_VALUE: i64 = i64::MAX; + +static DOCUMENT_ID_HEADER_KEY: std::sync::LazyLock<HeaderKey> = std::sync::LazyLock::new(|| { + HeaderKey::try_from(DOCUMENT_ID_HEADER).expect("document id header key is a valid literal") +}); +static INDEX_HEADER_KEY: std::sync::LazyLock<HeaderKey> = std::sync::LazyLock::new(|| { + HeaderKey::try_from(INDEX_HEADER).expect("index header key is a valid literal") +}); + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + #[serde(default)] + last_poll_timestamp: Option<DateTime<Utc>>, + #[serde(default)] + total_documents_published: usize, + #[serde(default)] + poll_count: usize, + /// OpenSearch `search_after` tuple from the last hit in the previous batch. + #[serde(default)] + search_after: Option<Vec<Value>>, + #[serde(default)] + error_count: usize, + #[serde(default)] + last_error: Option<String>, + #[serde(default)] + processing_stats: ProcessingStats, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct ProcessingStats { + #[serde(default)] + total_bytes_processed: u64, + /// Running cumulative average over the connector's lifetime, persisted and accumulated + /// across restarts. Reflects long-term throughput baseline, not session-only average. + #[serde(default)] + avg_batch_processing_time_ms: f64, + #[serde(default)] + last_successful_poll: Option<DateTime<Utc>>, + #[serde(default)] + empty_polls_count: usize, + #[serde(default)] + successful_polls_count: usize, +} + +// `Deserialize` only: nothing re-serializes a plugin config, and omitting `Serialize` +// keeps `password` from ever being written out as plaintext. +#[derive(Debug, Clone, Deserialize)] +pub struct OpenSearchSourceConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub query: Option<Value>, + pub polling_interval: Option<String>, + pub batch_size: Option<usize>, + pub timestamp_field: Option<String>, + #[serde(default)] + pub verbose_logging: bool, + /// Total poll-phase attempt count (not retry count); 1 = no retries, 3 = 2 retries. + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub retry_max_delay: Option<String>, + /// Total open-phase attempt count (not retry count); 1 = no retries, 5 = 4 retries. + pub max_open_retries: Option<u32>, + pub open_retry_max_delay: Option<String>, + pub circuit_breaker_threshold: Option<u32>, + pub circuit_breaker_cool_down: Option<String>, +} + +#[derive(Debug)] +pub struct OpenSearchSource { + id: u32, + config: OpenSearchSourceConfig, + client: Option<OpenSearch>, + polling_interval: Duration, + /// Pre-built query body (query + size + sort). Set once in `open()` after config validation. + /// `None` before `open()` and after a failed `open()`. + search_body_base: Option<Value>, + verbose: bool, + poll_retry_policy: RetryPolicy, + open_retry_policy: RetryPolicy, + circuit_breaker: Arc<CircuitBreaker>, + state: Mutex<State>, + /// Candidate state of the batch in flight; committed on ACK, dropped on NACK. + pending_state: Mutex<Option<State>>, + /// `Some(cause)` when runtime state restore was rejected; `None` means restore succeeded. + state_restore_error: Option<String>, +} + +struct SearchOutcome { + messages: Vec<ProducedMessage>, + search_after: Option<Vec<Value>>, + last_poll_timestamp: Option<DateTime<Utc>>, + batch_bytes: u64, +} + +impl OpenSearchSource { + pub fn new(id: u32, config: OpenSearchSourceConfig, state: Option<ConnectorState>) -> Self { + let polling_interval = + parse_duration(config.polling_interval.as_deref(), DEFAULT_POLLING_INTERVAL); + let verbose = config.verbose_logging; + let (restored_state, state_restore_error) = restore_state(id, state); + + let cb_threshold = config + .circuit_breaker_threshold + .unwrap_or(DEFAULT_CB_THRESHOLD); + let cb_cool_down = parse_duration( + config.circuit_breaker_cool_down.as_deref(), + DEFAULT_CB_COOL_DOWN, + ); + let circuit_breaker = Arc::new(CircuitBreaker::new(cb_threshold, cb_cool_down)); + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + let poll_retry_policy = RetryPolicy { + max_attempts: config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + base_delay: retry_delay, + max_delay: parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY), + }; + // The open phase has no base delay of its own: it shares `retry_delay` and only + // `open_retry_max_delay` caps the growth independently. + let open_retry_policy = RetryPolicy { + max_attempts: config.max_open_retries.unwrap_or(DEFAULT_MAX_OPEN_RETRIES), + base_delay: retry_delay, + max_delay: parse_duration( + config.open_retry_max_delay.as_deref(), + DEFAULT_OPEN_RETRY_MAX_DELAY, + ), + }; + + OpenSearchSource { + id, + config, + poll_retry_policy, + open_retry_policy, + circuit_breaker, + client: None, + polling_interval, + search_body_base: None, + verbose, + state: Mutex::new(restored_state), + pending_state: Mutex::new(None), + state_restore_error, + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn batch_size(&self) -> usize { + self.config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE) + } + + fn timestamp_field(&self) -> Result<&str, Error> { + self.config + .timestamp_field + .as_deref() + .filter(|s| !s.is_empty()) + .ok_or_else(|| Error::InvalidConfigValue("timestamp_field is required".to_string())) + } + + async fn create_client(&self) -> Result<OpenSearch, Error> { + let url = Url::parse(&self.config.url).map_err(|error| { + Error::InvalidConfigValue(format!("Invalid OpenSearch URL: {error}")) + })?; + + let conn_pool = opensearch::http::transport::SingleNodeConnectionPool::new(url); + let mut transport_builder = TransportBuilder::new(conn_pool); + + if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { + let credentials = + Credentials::Basic(username.clone(), password.expose_secret().to_string()); + transport_builder = transport_builder.auth(credentials); + } + + let transport = transport_builder + .build() + .map_err(|error| Error::InitError(format!("Failed to build transport: {error}")))?; Review Comment: The OpenSearch transport defaults to no request timeout. If the cluster accepts a connection but never finishes the response, the index check or search can wait indefinitely, so neither the retry budget nor the circuit breaker gets a chance to run and ingestion stops progressing. Set a finite transport timeout so these requests eventually fail and can be retried. ########## core/connectors/sources/opensearch_source/src/lib.rs: ########## @@ -0,0 +1,1043 @@ +// 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 iggy_common::{DateTime, HeaderKey, HeaderValue, Utc}; +use iggy_connector_sdk::retry::{CircuitBreaker, RetryPolicy, parse_duration, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use opensearch::{ + OpenSearch, SearchParts, + auth::Credentials, + http::{StatusCode, Url, response::Response, transport::TransportBuilder}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; +use tokio::{sync::Mutex, time::sleep}; +use tracing::{debug, error, info, warn}; + +mod retry; +use crate::retry::{ + DEFAULT_CB_COOL_DOWN, DEFAULT_CB_THRESHOLD, DEFAULT_MAX_OPEN_RETRIES, DEFAULT_MAX_RETRIES, + DEFAULT_OPEN_RETRY_MAX_DELAY, DEFAULT_RETRY_DELAY, DEFAULT_RETRY_MAX_DELAY, RequestFailure, +}; + +source_connector!(OpenSearchSource); + +const CONNECTOR_NAME: &str = "OpenSearch source"; +const DEFAULT_POLLING_INTERVAL: &str = "10s"; +const DEFAULT_BATCH_SIZE: usize = 100; +// OpenSearch rejects `size` above the default `index.max_result_window` with HTTP 400. +const MAX_BATCH_SIZE: usize = 10_000; +const DOCUMENT_ID_HEADER: &str = "opensearch.document_id"; +const INDEX_HEADER: &str = "opensearch.index"; +const MAX_HEADER_VALUE_LENGTH: usize = 255; +const MAX_LAST_ERROR_CHARS: usize = 512; +// What OpenSearch reports for a document missing the sort field when it sorts last. The sort +// asks for missing values first, so seeing this means that request option was not honored. +const MISSING_LAST_SORT_VALUE: i64 = i64::MAX; + +static DOCUMENT_ID_HEADER_KEY: std::sync::LazyLock<HeaderKey> = std::sync::LazyLock::new(|| { + HeaderKey::try_from(DOCUMENT_ID_HEADER).expect("document id header key is a valid literal") +}); +static INDEX_HEADER_KEY: std::sync::LazyLock<HeaderKey> = std::sync::LazyLock::new(|| { + HeaderKey::try_from(INDEX_HEADER).expect("index header key is a valid literal") +}); + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + #[serde(default)] + last_poll_timestamp: Option<DateTime<Utc>>, + #[serde(default)] + total_documents_published: usize, + #[serde(default)] + poll_count: usize, + /// OpenSearch `search_after` tuple from the last hit in the previous batch. + #[serde(default)] + search_after: Option<Vec<Value>>, + #[serde(default)] + error_count: usize, + #[serde(default)] + last_error: Option<String>, + #[serde(default)] + processing_stats: ProcessingStats, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct ProcessingStats { + #[serde(default)] + total_bytes_processed: u64, + /// Running cumulative average over the connector's lifetime, persisted and accumulated + /// across restarts. Reflects long-term throughput baseline, not session-only average. + #[serde(default)] + avg_batch_processing_time_ms: f64, + #[serde(default)] + last_successful_poll: Option<DateTime<Utc>>, + #[serde(default)] + empty_polls_count: usize, + #[serde(default)] + successful_polls_count: usize, +} + +// `Deserialize` only: nothing re-serializes a plugin config, and omitting `Serialize` +// keeps `password` from ever being written out as plaintext. +#[derive(Debug, Clone, Deserialize)] +pub struct OpenSearchSourceConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub query: Option<Value>, + pub polling_interval: Option<String>, + pub batch_size: Option<usize>, + pub timestamp_field: Option<String>, + #[serde(default)] + pub verbose_logging: bool, + /// Total poll-phase attempt count (not retry count); 1 = no retries, 3 = 2 retries. + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub retry_max_delay: Option<String>, + /// Total open-phase attempt count (not retry count); 1 = no retries, 5 = 4 retries. + pub max_open_retries: Option<u32>, + pub open_retry_max_delay: Option<String>, + pub circuit_breaker_threshold: Option<u32>, + pub circuit_breaker_cool_down: Option<String>, +} + +#[derive(Debug)] +pub struct OpenSearchSource { + id: u32, + config: OpenSearchSourceConfig, + client: Option<OpenSearch>, + polling_interval: Duration, + /// Pre-built query body (query + size + sort). Set once in `open()` after config validation. + /// `None` before `open()` and after a failed `open()`. + search_body_base: Option<Value>, + verbose: bool, + poll_retry_policy: RetryPolicy, + open_retry_policy: RetryPolicy, + circuit_breaker: Arc<CircuitBreaker>, + state: Mutex<State>, + /// Candidate state of the batch in flight; committed on ACK, dropped on NACK. + pending_state: Mutex<Option<State>>, + /// `Some(cause)` when runtime state restore was rejected; `None` means restore succeeded. + state_restore_error: Option<String>, +} + +struct SearchOutcome { + messages: Vec<ProducedMessage>, + search_after: Option<Vec<Value>>, + last_poll_timestamp: Option<DateTime<Utc>>, + batch_bytes: u64, +} + +impl OpenSearchSource { + pub fn new(id: u32, config: OpenSearchSourceConfig, state: Option<ConnectorState>) -> Self { + let polling_interval = + parse_duration(config.polling_interval.as_deref(), DEFAULT_POLLING_INTERVAL); + let verbose = config.verbose_logging; + let (restored_state, state_restore_error) = restore_state(id, state); + + let cb_threshold = config + .circuit_breaker_threshold + .unwrap_or(DEFAULT_CB_THRESHOLD); + let cb_cool_down = parse_duration( + config.circuit_breaker_cool_down.as_deref(), + DEFAULT_CB_COOL_DOWN, + ); + let circuit_breaker = Arc::new(CircuitBreaker::new(cb_threshold, cb_cool_down)); + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + let poll_retry_policy = RetryPolicy { + max_attempts: config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + base_delay: retry_delay, + max_delay: parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY), + }; + // The open phase has no base delay of its own: it shares `retry_delay` and only + // `open_retry_max_delay` caps the growth independently. + let open_retry_policy = RetryPolicy { + max_attempts: config.max_open_retries.unwrap_or(DEFAULT_MAX_OPEN_RETRIES), + base_delay: retry_delay, + max_delay: parse_duration( + config.open_retry_max_delay.as_deref(), + DEFAULT_OPEN_RETRY_MAX_DELAY, + ), + }; + + OpenSearchSource { + id, + config, + poll_retry_policy, + open_retry_policy, + circuit_breaker, + client: None, + polling_interval, + search_body_base: None, + verbose, + state: Mutex::new(restored_state), + pending_state: Mutex::new(None), + state_restore_error, + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn batch_size(&self) -> usize { + self.config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE) + } + + fn timestamp_field(&self) -> Result<&str, Error> { + self.config + .timestamp_field + .as_deref() + .filter(|s| !s.is_empty()) + .ok_or_else(|| Error::InvalidConfigValue("timestamp_field is required".to_string())) + } + + async fn create_client(&self) -> Result<OpenSearch, Error> { + let url = Url::parse(&self.config.url).map_err(|error| { + Error::InvalidConfigValue(format!("Invalid OpenSearch URL: {error}")) + })?; + + let conn_pool = opensearch::http::transport::SingleNodeConnectionPool::new(url); + let mut transport_builder = TransportBuilder::new(conn_pool); + + if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { + let credentials = + Credentials::Basic(username.clone(), password.expose_secret().to_string()); + transport_builder = transport_builder.auth(credentials); + } + + let transport = transport_builder + .build() + .map_err(|error| Error::InitError(format!("Failed to build transport: {error}")))?; + + Ok(OpenSearch::new(transport)) + } + + async fn check_index_exists_with_retry(&self, client: &OpenSearch) -> Result<(), Error> { + let index = &self.config.index; + retry_async( + self.open_retry_policy, + &format!("{CONNECTOR_NAME} ID {} index exists check", self.id), + RequestFailure::is_transient, + || async { + let response = client + .indices() + .exists(opensearch::indices::IndicesExistsParts::Index(&[index])) + .send() + .await + .map_err(RequestFailure::transport)?; + RequestFailure::check(response).await.map(drop) + }, + ) + .await + .map_err(|failure| match failure.into_error() { + RequestFailure::Transport(message) => { + Error::InitError(format!("Failed to check index existence: {message}")) + } + RequestFailure::Status { status, .. } if status == StatusCode::NOT_FOUND => { + Error::InitError(format!( + "Index '{index}' does not exist or is not accessible" + )) + } + failure @ RequestFailure::Status { .. } => { + Error::InitError(format!("Index '{index}' check failed: {failure}")) + } + }) + } + + async fn send_search_with_retry( + &self, + client: &OpenSearch, + search_body: &Value, + ) -> Result<Response, Error> { + retry_async( + self.poll_retry_policy, + &format!("{CONNECTOR_NAME} ID {} search", self.id), + RequestFailure::is_transient, + || async { + let response = client + .search(SearchParts::Index(&[&self.config.index])) + .allow_partial_search_results(false) + .body(search_body) + .send() + .await + .map_err(RequestFailure::transport)?; + RequestFailure::check(response).await + }, + ) + .await + .map_err(|failure| failure.into_error().into_search_error()) + } + + async fn search_documents(&self, client: &OpenSearch) -> Result<SearchOutcome, Error> { + let state = self.state.lock().await; + let search_after = state.search_after.clone(); + drop(state); + + let timestamp_field = self.timestamp_field()?; + + let mut search_body = self + .search_body_base + .as_ref() + .ok_or_else(|| { + Error::Connection("connector not initialized; call open() first".to_string()) + })? + .clone(); + + if let Some(cursor) = search_after { + search_body["search_after"] = json!(cursor); + } + + let response = self.send_search_with_retry(client, &search_body).await?; + + let mut response_body: Value = response.json().await.map_err(|e| { + Error::PermanentHttpError(format!("Failed to parse search response: {e}")) + })?; + ensure_complete_response(&response_body)?; + + let hits: Vec<Value> = response_body + .get_mut("hits") + .and_then(|h| h.get_mut("hits")) + .and_then(|arr| arr.as_array_mut()) + .map(std::mem::take) + .unwrap_or_default(); + + let mut messages = Vec::with_capacity(hits.len()); + let mut batch_bytes = 0u64; + let mut last_sort = None; + let mut last_poll_timestamp = None; + + for hit in &hits { + let Some(sort) = hit + .get("sort") + .and_then(|s| s.as_array()) + .filter(|a| !a.is_empty()) + else { + return Err(Error::SchemaMismatch(format!( + "OpenSearch hit {:?} has no sort tuple; the cursor cannot move past it \ + without risking skipped documents", + hit.get("_id").and_then(Value::as_str) + ))); + }; + + if sort.first().and_then(Value::as_i64) == Some(MISSING_LAST_SORT_VALUE) { + return Err(Error::SchemaMismatch(format!( + "OpenSearch hit {:?} sorted on the missing-value sentinel; a cursor there \ + would skip every document indexed later", + hit.get("_id").and_then(Value::as_str) + ))); + } + + let Some(source) = hit.get("_source") else { + warn!( + connector_id = self.id, + hit_id = hit.get("_id").and_then(|v| v.as_str()), + hit_index = hit.get("_index").and_then(|v| v.as_str()), + "Skipping OpenSearch hit without _source; it has no body to publish" + ); + last_sort = Some(sort); + continue; + }; + + let document_timestamp = source + .get(timestamp_field) + .and_then(parse_document_timestamp); Review Comment: A normal OpenSearch field path such as `event.created` works in the sort, but `Value::get("event.created")` only looks for a literal top-level key. For a document containing `{"event":{"created":"2024-01-01T00:00:00Z"}}`, this silently leaves `origin_timestamp` unset and never updates `last_poll_timestamp`. Resolve dotted object paths as well as literal keys when reading the configured timestamp. ########## core/integration/tests/connectors/fixtures/opensearch/resilience.rs: ########## @@ -0,0 +1,204 @@ +// 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 super::container::{ + DEFAULT_TEST_STREAM, DEFAULT_TEST_TOPIC, ENV_SOURCE_BATCH_SIZE, + ENV_SOURCE_CIRCUIT_BREAKER_COOL_DOWN, ENV_SOURCE_CIRCUIT_BREAKER_THRESHOLD, ENV_SOURCE_INDEX, + ENV_SOURCE_MAX_OPEN_RETRIES, ENV_SOURCE_MAX_RETRIES, ENV_SOURCE_OPEN_RETRY_MAX_DELAY, + ENV_SOURCE_PATH, ENV_SOURCE_POLLING_INTERVAL, ENV_SOURCE_RETRY_DELAY, + ENV_SOURCE_RETRY_MAX_DELAY, ENV_SOURCE_STREAMS_0_SCHEMA, ENV_SOURCE_STREAMS_0_STREAM, + ENV_SOURCE_STREAMS_0_TOPIC, ENV_SOURCE_TIMESTAMP_FIELD, ENV_SOURCE_URL, +}; +use async_trait::async_trait; +use integration::harness::{TestBinaryError, TestFixture}; +use std::collections::HashMap; +use wiremock::matchers::{method, path_regex}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +pub const RESILIENCE_INDEX: &str = "resilience_test"; + +fn index_path_pattern() -> &'static str { + r"/resilience_test$" +} + +fn search_path_pattern() -> &'static str { + r"/resilience_test/_search" +} + +fn search_success_body() -> serde_json::Value { + serde_json::json!({ + "hits": { + "hits": [{ + "_id": "doc-retry-1", + "_source": { + "id": 42, + "name": "retry_doc", + "value": 100, + "timestamp": "2024-01-15T12:00:00.000Z" + }, + "sort": ["2024-01-15T12:00:00.000Z", "doc-retry-1"] + }] + } + }) +} + +fn resilience_base_envs(mock_uri: &str) -> HashMap<String, String> { + let mut envs = HashMap::new(); + envs.insert(ENV_SOURCE_URL.to_string(), mock_uri.to_string()); + envs.insert(ENV_SOURCE_INDEX.to_string(), RESILIENCE_INDEX.to_string()); + envs.insert(ENV_SOURCE_POLLING_INTERVAL.to_string(), "100ms".to_string()); + envs.insert(ENV_SOURCE_BATCH_SIZE.to_string(), "10".to_string()); + envs.insert( + ENV_SOURCE_TIMESTAMP_FIELD.to_string(), + "timestamp".to_string(), + ); + envs.insert( + ENV_SOURCE_STREAMS_0_STREAM.to_string(), + DEFAULT_TEST_STREAM.to_string(), + ); + envs.insert( + ENV_SOURCE_STREAMS_0_TOPIC.to_string(), + DEFAULT_TEST_TOPIC.to_string(), + ); + envs.insert(ENV_SOURCE_STREAMS_0_SCHEMA.to_string(), "json".to_string()); + envs.insert( + ENV_SOURCE_PATH.to_string(), + "../../target/debug/libiggy_connector_opensearch_source".to_string(), + ); + envs.insert(ENV_SOURCE_RETRY_DELAY.to_string(), "50ms".to_string()); + envs.insert(ENV_SOURCE_RETRY_MAX_DELAY.to_string(), "200ms".to_string()); + envs.insert( + ENV_SOURCE_OPEN_RETRY_MAX_DELAY.to_string(), + "200ms".to_string(), + ); + envs.insert(ENV_SOURCE_MAX_OPEN_RETRIES.to_string(), "2".to_string()); + envs +} + +async fn mount_index_exists(mock_server: &MockServer) { + Mock::given(method("HEAD")) + .and(path_regex(index_path_pattern())) + .respond_with(ResponseTemplate::new(200)) + .mount(mock_server) + .await; +} + +async fn mount_transient_search_mocks(mock_server: &MockServer) { + Mock::given(method("POST")) + .and(path_regex(search_path_pattern())) + .respond_with(ResponseTemplate::new(503).set_body_string("temporary")) + .up_to_n_times(2) + .with_priority(1) + .mount(mock_server) + .await; + + Mock::given(method("POST")) + .and(path_regex(search_path_pattern())) + .respond_with(ResponseTemplate::new(200).set_body_json(search_success_body())) Review Comment: This success mock returns the same hit even after the request supplies its committed `search_after` cursor, so subsequent polls keep publishing document 42. If the consumer starts late, multiple copies can accumulate and fail the test's `received.len() == 1` assertion despite correct connector behavior. Return an empty page after the cursor advances, while still returning the hit on a Nack replay. ########## core/integration/tests/connectors/fixtures/opensearch/container.rs: ########## @@ -0,0 +1,368 @@ +// 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 integration::harness::TestBinaryError; +use reqwest_middleware::ClientWithMiddleware as HttpClient; +use reqwest_retry::RetryTransientMiddleware; +use reqwest_retry::policies::ExponentialBackoff; +use serde::Deserialize; +use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy; +use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{ + ContainerAsync, GenericImage, ImageExt, ReuseDirective, +}; +use tracing::info; + +const OPENSEARCH_IMAGE: &str = "docker.io/opensearchproject/opensearch"; +const OPENSEARCH_TAG: &str = "2.19.1"; +const OPENSEARCH_PORT: u16 = 9200; +const OPENSEARCH_HEALTH_ENDPOINT: &str = "/_cluster/health"; +// Fixed name + ReuseDirective::Always shares one container across nextest's +// per-test processes. Isolation comes from a unique index per fixture (see +// source.rs); index creation also drops any leftover index of the same name. +const OPENSEARCH_CONTAINER_NAME: &str = "iggy-test-opensearch"; + +pub const DEFAULT_TEST_STREAM: &str = "test_stream"; +pub const DEFAULT_TEST_TOPIC: &str = "test_topic"; + +pub const ENV_SOURCE_URL: &str = "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_URL"; +pub const ENV_SOURCE_INDEX: &str = "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_INDEX"; +pub const ENV_SOURCE_POLLING_INTERVAL: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_POLLING_INTERVAL"; +pub const ENV_SOURCE_BATCH_SIZE: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_BATCH_SIZE"; +pub const ENV_SOURCE_TIMESTAMP_FIELD: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_TIMESTAMP_FIELD"; +pub const ENV_SOURCE_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SOURCE_OPENSEARCH_STREAMS_0_STREAM"; +pub const ENV_SOURCE_STREAMS_0_TOPIC: &str = "IGGY_CONNECTORS_SOURCE_OPENSEARCH_STREAMS_0_TOPIC"; +pub const ENV_SOURCE_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SOURCE_OPENSEARCH_STREAMS_0_SCHEMA"; +pub const ENV_SOURCE_PATH: &str = "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PATH"; +pub const ENV_SOURCE_MAX_RETRIES: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_MAX_RETRIES"; +pub const ENV_SOURCE_RETRY_DELAY: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_RETRY_DELAY"; +pub const ENV_SOURCE_RETRY_MAX_DELAY: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_RETRY_MAX_DELAY"; +pub const ENV_SOURCE_MAX_OPEN_RETRIES: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_MAX_OPEN_RETRIES"; +pub const ENV_SOURCE_OPEN_RETRY_MAX_DELAY: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_OPEN_RETRY_MAX_DELAY"; +pub const ENV_SOURCE_CIRCUIT_BREAKER_THRESHOLD: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_CIRCUIT_BREAKER_THRESHOLD"; +pub const ENV_SOURCE_CIRCUIT_BREAKER_COOL_DOWN: &str = + "IGGY_CONNECTORS_SOURCE_OPENSEARCH_PLUGIN_CONFIG_CIRCUIT_BREAKER_COOL_DOWN"; + +pub struct OpenSearchContainer { + _container: ContainerAsync<GenericImage>, + pub base_url: String, +} + +impl OpenSearchContainer { + pub async fn start() -> Result<Self, TestBinaryError> { + let container = GenericImage::new(OPENSEARCH_IMAGE, OPENSEARCH_TAG) + .with_exposed_port(OPENSEARCH_PORT.tcp()) + .with_wait_for(WaitFor::http( + HttpWaitStrategy::new(OPENSEARCH_HEALTH_ENDPOINT) + .with_port(OPENSEARCH_PORT.tcp()) + .with_expected_status_code(200u16), + )) + .with_startup_timeout(std::time::Duration::from_secs(120)) + .with_env_var("discovery.type", "single-node") + .with_env_var("plugins.security.disabled", "true") + .with_env_var("DISABLE_INSTALL_DEMO_CONFIG", "true") + .with_env_var("OPENSEARCH_INITIAL_ADMIN_PASSWORD", "iggy-test-password1!") + .with_env_var("OPENSEARCH_JAVA_OPTS", "-Xms512m -Xmx512m") + .with_mapped_port(0, OPENSEARCH_PORT.tcp()) + .with_container_name(OPENSEARCH_CONTAINER_NAME) + .with_reuse(ReuseDirective::Always) + .start() + .await Review Comment: The new nextest group does not serialize the documented `cargo test -p integration -- connectors::opensearch` command. On a cold Docker daemon, parallel fixtures can both observe that this fixed-name container is absent and then race to create it, causing one setup to fail with Docker 409. Please make creation runner-independent, for example with the bounded create-or-attach retry already used by `start_shared_container` in `fixtures/elasticsearch/container.rs`. ########## core/integration/tests/connectors/opensearch/opensearch_source_types.rs: ########## @@ -0,0 +1,215 @@ +// 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 super::{ + TEST_MESSAGE_COUNT, expected_ids, json_payloads, poll_next, read_from_start, + sorted_document_ids, +}; +use crate::connectors::fixtures::{ + OpenSearchSourcePreCreatedFixture, OpenSearchSourceTypedFieldsFixture, +}; +use iggy_common::{HeaderKey, IggyTimestamp}; +use integration::harness::seeds; +use integration::iggy_harness; + +const DOCUMENT_ID_HEADER: &str = "opensearch.document_id"; +const INDEX_HEADER: &str = "opensearch.index"; +const ORIGIN_TIMESTAMP_TOLERANCE_MICROS: u64 = 3_600_000_000; + +// Mirrors the connector's id derivation on purpose: the id is a contract consumers dedupe on. +fn expected_message_id(index: &str, document_id: &str) -> u128 { + let mut hasher = blake3::Hasher::new(); + for part in [index, document_id] { + hasher.update(&(part.len() as u64).to_le_bytes()); + hasher.update(part.as_bytes()); + } + let mut id = [0u8; 16]; + hasher.finalize_xof().fill(&mut id); + u128::from_le_bytes(id) +} + +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/opensearch/source.toml")), + seed = seeds::connector_stream +)] +async fn given_document_in_index_when_connector_polls_should_publish_payload_id_headers_and_timestamp( + harness: &TestHarness, + fixture: OpenSearchSourcePreCreatedFixture, +) { + fixture + .insert_document(1, "identity_doc", 42) + .await + .expect("insert document"); + fixture.refresh_index().await.expect("refresh index"); + + let client = harness.root_client().await.unwrap(); + let messages = poll_next(&client, "single_document_consumer", 1).await; + + assert_eq!(messages.len(), 1, "expected one message"); + let message = &messages[0]; + + let payload = &json_payloads(&messages)[0]; + assert_eq!(payload["id"], 1); + assert_eq!(payload["name"], "identity_doc"); + assert_eq!(payload["value"], 42); + assert!( + payload.get("timestamp").is_some(), + "missing timestamp field" + ); + + assert_eq!( + message.header.id, + expected_message_id(fixture.index_name(), "1") + ); + let now_micros = IggyTimestamp::now().as_micros(); + assert!( + message.header.origin_timestamp.abs_diff(now_micros) < ORIGIN_TIMESTAMP_TOLERANCE_MICROS, + "origin_timestamp {} must be the document timestamp in microseconds (now: {now_micros})", + message.header.origin_timestamp Review Comment: This one-hour recency check passes even when the document timestamp is discarded: the runtime's `process_messages` does not pass `origin_timestamp` to `build_iggy_message`. Seed a known historical timestamp and assert its exact microsecond value instead. Forward that timestamp through the runtime message builder so the end-to-end test verifies preservation rather than recency. ########## core/connectors/sources/opensearch_source/src/lib.rs: ########## @@ -0,0 +1,1043 @@ +// 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 iggy_common::{DateTime, HeaderKey, HeaderValue, Utc}; +use iggy_connector_sdk::retry::{CircuitBreaker, RetryPolicy, parse_duration, retry_async}; +use iggy_connector_sdk::{ + ConnectorState, Error, ProducedMessage, ProducedMessages, Schema, Source, + source::SourceBatchResult, source_connector, +}; +use opensearch::{ + OpenSearch, SearchParts, + auth::Credentials, + http::{StatusCode, Url, response::Response, transport::TransportBuilder}, +}; +use secrecy::{ExposeSecret, SecretString}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; +use tokio::{sync::Mutex, time::sleep}; +use tracing::{debug, error, info, warn}; + +mod retry; +use crate::retry::{ + DEFAULT_CB_COOL_DOWN, DEFAULT_CB_THRESHOLD, DEFAULT_MAX_OPEN_RETRIES, DEFAULT_MAX_RETRIES, + DEFAULT_OPEN_RETRY_MAX_DELAY, DEFAULT_RETRY_DELAY, DEFAULT_RETRY_MAX_DELAY, RequestFailure, +}; + +source_connector!(OpenSearchSource); + +const CONNECTOR_NAME: &str = "OpenSearch source"; +const DEFAULT_POLLING_INTERVAL: &str = "10s"; +const DEFAULT_BATCH_SIZE: usize = 100; +// OpenSearch rejects `size` above the default `index.max_result_window` with HTTP 400. +const MAX_BATCH_SIZE: usize = 10_000; +const DOCUMENT_ID_HEADER: &str = "opensearch.document_id"; +const INDEX_HEADER: &str = "opensearch.index"; +const MAX_HEADER_VALUE_LENGTH: usize = 255; +const MAX_LAST_ERROR_CHARS: usize = 512; +// What OpenSearch reports for a document missing the sort field when it sorts last. The sort +// asks for missing values first, so seeing this means that request option was not honored. +const MISSING_LAST_SORT_VALUE: i64 = i64::MAX; + +static DOCUMENT_ID_HEADER_KEY: std::sync::LazyLock<HeaderKey> = std::sync::LazyLock::new(|| { + HeaderKey::try_from(DOCUMENT_ID_HEADER).expect("document id header key is a valid literal") +}); +static INDEX_HEADER_KEY: std::sync::LazyLock<HeaderKey> = std::sync::LazyLock::new(|| { + HeaderKey::try_from(INDEX_HEADER).expect("index header key is a valid literal") +}); + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct State { + #[serde(default)] + last_poll_timestamp: Option<DateTime<Utc>>, + #[serde(default)] + total_documents_published: usize, + #[serde(default)] + poll_count: usize, + /// OpenSearch `search_after` tuple from the last hit in the previous batch. + #[serde(default)] + search_after: Option<Vec<Value>>, + #[serde(default)] + error_count: usize, + #[serde(default)] + last_error: Option<String>, + #[serde(default)] + processing_stats: ProcessingStats, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +struct ProcessingStats { + #[serde(default)] + total_bytes_processed: u64, + /// Running cumulative average over the connector's lifetime, persisted and accumulated + /// across restarts. Reflects long-term throughput baseline, not session-only average. + #[serde(default)] + avg_batch_processing_time_ms: f64, + #[serde(default)] + last_successful_poll: Option<DateTime<Utc>>, + #[serde(default)] + empty_polls_count: usize, + #[serde(default)] + successful_polls_count: usize, +} + +// `Deserialize` only: nothing re-serializes a plugin config, and omitting `Serialize` +// keeps `password` from ever being written out as plaintext. +#[derive(Debug, Clone, Deserialize)] +pub struct OpenSearchSourceConfig { + pub url: String, + pub index: String, + pub username: Option<String>, + pub password: Option<SecretString>, + pub query: Option<Value>, + pub polling_interval: Option<String>, + pub batch_size: Option<usize>, + pub timestamp_field: Option<String>, + #[serde(default)] + pub verbose_logging: bool, + /// Total poll-phase attempt count (not retry count); 1 = no retries, 3 = 2 retries. + pub max_retries: Option<u32>, + pub retry_delay: Option<String>, + pub retry_max_delay: Option<String>, + /// Total open-phase attempt count (not retry count); 1 = no retries, 5 = 4 retries. + pub max_open_retries: Option<u32>, + pub open_retry_max_delay: Option<String>, + pub circuit_breaker_threshold: Option<u32>, + pub circuit_breaker_cool_down: Option<String>, +} + +#[derive(Debug)] +pub struct OpenSearchSource { + id: u32, + config: OpenSearchSourceConfig, + client: Option<OpenSearch>, + polling_interval: Duration, + /// Pre-built query body (query + size + sort). Set once in `open()` after config validation. + /// `None` before `open()` and after a failed `open()`. + search_body_base: Option<Value>, + verbose: bool, + poll_retry_policy: RetryPolicy, + open_retry_policy: RetryPolicy, + circuit_breaker: Arc<CircuitBreaker>, + state: Mutex<State>, + /// Candidate state of the batch in flight; committed on ACK, dropped on NACK. + pending_state: Mutex<Option<State>>, + /// `Some(cause)` when runtime state restore was rejected; `None` means restore succeeded. + state_restore_error: Option<String>, +} + +struct SearchOutcome { + messages: Vec<ProducedMessage>, + search_after: Option<Vec<Value>>, + last_poll_timestamp: Option<DateTime<Utc>>, + batch_bytes: u64, +} + +impl OpenSearchSource { + pub fn new(id: u32, config: OpenSearchSourceConfig, state: Option<ConnectorState>) -> Self { + let polling_interval = + parse_duration(config.polling_interval.as_deref(), DEFAULT_POLLING_INTERVAL); + let verbose = config.verbose_logging; + let (restored_state, state_restore_error) = restore_state(id, state); + + let cb_threshold = config + .circuit_breaker_threshold + .unwrap_or(DEFAULT_CB_THRESHOLD); + let cb_cool_down = parse_duration( + config.circuit_breaker_cool_down.as_deref(), + DEFAULT_CB_COOL_DOWN, + ); + let circuit_breaker = Arc::new(CircuitBreaker::new(cb_threshold, cb_cool_down)); + let retry_delay = parse_duration(config.retry_delay.as_deref(), DEFAULT_RETRY_DELAY); + let poll_retry_policy = RetryPolicy { + max_attempts: config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES), + base_delay: retry_delay, + max_delay: parse_duration(config.retry_max_delay.as_deref(), DEFAULT_RETRY_MAX_DELAY), + }; + // The open phase has no base delay of its own: it shares `retry_delay` and only + // `open_retry_max_delay` caps the growth independently. + let open_retry_policy = RetryPolicy { + max_attempts: config.max_open_retries.unwrap_or(DEFAULT_MAX_OPEN_RETRIES), + base_delay: retry_delay, + max_delay: parse_duration( + config.open_retry_max_delay.as_deref(), + DEFAULT_OPEN_RETRY_MAX_DELAY, + ), + }; + + OpenSearchSource { + id, + config, + poll_retry_policy, + open_retry_policy, + circuit_breaker, + client: None, + polling_interval, + search_body_base: None, + verbose, + state: Mutex::new(restored_state), + pending_state: Mutex::new(None), + state_restore_error, + } + } + + fn serialize_state(&self, state: &State) -> Option<ConnectorState> { + ConnectorState::serialize(state, CONNECTOR_NAME, self.id) + } + + fn batch_size(&self) -> usize { + self.config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE) + } + + fn timestamp_field(&self) -> Result<&str, Error> { + self.config + .timestamp_field + .as_deref() + .filter(|s| !s.is_empty()) + .ok_or_else(|| Error::InvalidConfigValue("timestamp_field is required".to_string())) + } + + async fn create_client(&self) -> Result<OpenSearch, Error> { + let url = Url::parse(&self.config.url).map_err(|error| { + Error::InvalidConfigValue(format!("Invalid OpenSearch URL: {error}")) + })?; + + let conn_pool = opensearch::http::transport::SingleNodeConnectionPool::new(url); + let mut transport_builder = TransportBuilder::new(conn_pool); + + if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { + let credentials = + Credentials::Basic(username.clone(), password.expose_secret().to_string()); + transport_builder = transport_builder.auth(credentials); + } + + let transport = transport_builder + .build() + .map_err(|error| Error::InitError(format!("Failed to build transport: {error}")))?; + + Ok(OpenSearch::new(transport)) + } + + async fn check_index_exists_with_retry(&self, client: &OpenSearch) -> Result<(), Error> { + let index = &self.config.index; + retry_async( + self.open_retry_policy, + &format!("{CONNECTOR_NAME} ID {} index exists check", self.id), + RequestFailure::is_transient, + || async { + let response = client + .indices() + .exists(opensearch::indices::IndicesExistsParts::Index(&[index])) + .send() + .await + .map_err(RequestFailure::transport)?; + RequestFailure::check(response).await.map(drop) + }, + ) + .await + .map_err(|failure| match failure.into_error() { + RequestFailure::Transport(message) => { + Error::InitError(format!("Failed to check index existence: {message}")) + } + RequestFailure::Status { status, .. } if status == StatusCode::NOT_FOUND => { + Error::InitError(format!( + "Index '{index}' does not exist or is not accessible" + )) + } + failure @ RequestFailure::Status { .. } => { + Error::InitError(format!("Index '{index}' check failed: {failure}")) + } + }) + } + + async fn send_search_with_retry( + &self, + client: &OpenSearch, + search_body: &Value, + ) -> Result<Response, Error> { + retry_async( + self.poll_retry_policy, + &format!("{CONNECTOR_NAME} ID {} search", self.id), + RequestFailure::is_transient, + || async { + let response = client + .search(SearchParts::Index(&[&self.config.index])) + .allow_partial_search_results(false) + .body(search_body) + .send() + .await + .map_err(RequestFailure::transport)?; + RequestFailure::check(response).await + }, + ) + .await + .map_err(|failure| failure.into_error().into_search_error()) + } + + async fn search_documents(&self, client: &OpenSearch) -> Result<SearchOutcome, Error> { + let state = self.state.lock().await; + let search_after = state.search_after.clone(); + drop(state); + + let timestamp_field = self.timestamp_field()?; + + let mut search_body = self + .search_body_base + .as_ref() + .ok_or_else(|| { + Error::Connection("connector not initialized; call open() first".to_string()) + })? + .clone(); + + if let Some(cursor) = search_after { + search_body["search_after"] = json!(cursor); + } + + let response = self.send_search_with_retry(client, &search_body).await?; + + let mut response_body: Value = response.json().await.map_err(|e| { + Error::PermanentHttpError(format!("Failed to parse search response: {e}")) + })?; + ensure_complete_response(&response_body)?; + + let hits: Vec<Value> = response_body + .get_mut("hits") + .and_then(|h| h.get_mut("hits")) + .and_then(|arr| arr.as_array_mut()) + .map(std::mem::take) + .unwrap_or_default(); + + let mut messages = Vec::with_capacity(hits.len()); + let mut batch_bytes = 0u64; + let mut last_sort = None; + let mut last_poll_timestamp = None; + + for hit in &hits { + let Some(sort) = hit + .get("sort") + .and_then(|s| s.as_array()) + .filter(|a| !a.is_empty()) + else { + return Err(Error::SchemaMismatch(format!( + "OpenSearch hit {:?} has no sort tuple; the cursor cannot move past it \ + without risking skipped documents", + hit.get("_id").and_then(Value::as_str) + ))); + }; + + if sort.first().and_then(Value::as_i64) == Some(MISSING_LAST_SORT_VALUE) { + return Err(Error::SchemaMismatch(format!( + "OpenSearch hit {:?} sorted on the missing-value sentinel; a cursor there \ + would skip every document indexed later", + hit.get("_id").and_then(Value::as_str) + ))); + } + + let Some(source) = hit.get("_source") else { + warn!( + connector_id = self.id, + hit_id = hit.get("_id").and_then(|v| v.as_str()), + hit_index = hit.get("_index").and_then(|v| v.as_str()), + "Skipping OpenSearch hit without _source; it has no body to publish" + ); + last_sort = Some(sort); + continue; + }; + + let document_timestamp = source + .get(timestamp_field) + .and_then(parse_document_timestamp); + if let Some(timestamp_utc) = document_timestamp { + last_poll_timestamp = Some(timestamp_utc); + } + + let payload = serde_json::to_vec(source) + .map_err(|e| Error::Serialization(format!("Failed to serialize document: {e}")))?; + batch_bytes += payload.len() as u64; + + let document_id = hit.get("_id").and_then(Value::as_str); + // `index` may be a pattern, so the concrete index of each hit keeps equal `_id`s + // in different indices from sharing a message id. + let reported_index = hit.get("_index").and_then(Value::as_str); + let hit_index = reported_index.unwrap_or(&self.config.index); + messages.push(ProducedMessage { + id: document_id.map(|document_id| message_id(hit_index, document_id)), + headers: document_headers(self.id, reported_index, document_id), + checksum: None, + timestamp: None, + origin_timestamp: document_timestamp + .and_then(|timestamp| u64::try_from(timestamp.timestamp_micros()).ok()), + payload, + }); + // Cursor advances only after the document is queued for publish. + last_sort = Some(sort); + } + + Ok(SearchOutcome { + messages, + search_after: last_sort.map(ToOwned::to_owned), + last_poll_timestamp, + batch_bytes, + }) + } + + async fn finalize_poll( + &self, + outcome: SearchOutcome, + processing_time_ms: f64, + ) -> Result<(Vec<ProducedMessage>, Option<ConnectorState>), Error> { + let mut state = self.state.lock().await.clone(); + state.total_documents_published += outcome.messages.len(); + state.poll_count += 1; + let cursor_moved = outcome.search_after.is_some(); + if let Some(cursor) = outcome.search_after { + state.search_after = Some(cursor); + } + if let Some(timestamp) = outcome.last_poll_timestamp { + state.last_poll_timestamp = Some(timestamp); + } + state.processing_stats.total_bytes_processed += outcome.batch_bytes; + + if outcome.messages.is_empty() { + state.processing_stats.empty_polls_count += 1; + } else { + state.processing_stats.successful_polls_count += 1; + state.processing_stats.last_successful_poll = Some(Utc::now()); + } + + let total_polls = state.processing_stats.successful_polls_count + + state.processing_stats.empty_polls_count; + state.processing_stats.avg_batch_processing_time_ms += (processing_time_ms + - state.processing_stats.avg_batch_processing_time_ms) + / total_polls as f64; + + let produced_count = outcome.messages.len(); + let total_documents_published = state.total_documents_published; + let messages = outcome.messages; + // A moved cursor waits for the batch result so a Nack can undo it, including a page + // whose hits were all skipped for lacking `_source`. A poll that saw no hits has no + // cursor to protect, so its counters commit directly. + let persisted_state = if cursor_moved { + let persisted = self.serialize_state(&state).ok_or_else(|| { + Error::Serialization("failed to serialize source state".to_string()) + })?; + *self.pending_state.lock().await = Some(state); + Some(persisted) + } else { + *self.state.lock().await = state; + None + }; + + if self.verbose { + info!( + "OpenSearch source connector ID: {} produced {produced_count} messages. \ + Total published: {total_documents_published}", + self.id + ); + } else { + debug!( + "OpenSearch source connector ID: {} produced {produced_count} messages. \ + Total published: {total_documents_published}", + self.id + ); + } + + Ok((messages, persisted_state)) + } + + #[cfg(test)] + fn client_initialized(&self) -> bool { + self.client.is_some() + } + + #[cfg(test)] + async fn test_metrics(&self) -> (usize, usize, usize, usize) { + let state = self.state.lock().await; + ( + state.total_documents_published, + state.poll_count, + state.error_count, + state.processing_stats.empty_polls_count, + ) + } + + #[cfg(test)] + async fn test_search_after(&self) -> Option<Vec<Value>> { + self.state.lock().await.search_after.clone() + } + + #[cfg(test)] + async fn test_last_poll_timestamp(&self) -> Option<DateTime<Utc>> { + self.state.lock().await.last_poll_timestamp + } + + async fn handle_poll_error(&self, error: Error) -> Result<ProducedMessages, Error> { + // Only cluster-side trouble opens the breaker. A permanent error (bad query, missing + // sort tuple) is a configuration or mapping problem that skipping polls cannot fix, + // and an open breaker would hide it behind a quiet empty batch. + if matches!(error, Error::Connection(_) | Error::HttpRequestFailed(_)) { + self.circuit_breaker.record_failure().await; + } + let mut state = self.state.lock().await; + // error_count and last_error accumulate in State and are captured by the next + // finalize_poll call (success path). Both runtime ConnectorState and file state + // preserve these values across restarts. + state.error_count += 1; + state.last_error = Some(truncated(error.to_string())); + drop(state); + error!( + "{CONNECTOR_NAME} connector ID: {} poll failed: {error}", + self.id + ); + Err(error) + } +} + +// `last_error` is persisted with every batch, and an OpenSearch error body can be large. +fn truncated(message: String) -> String { + match message.char_indices().nth(MAX_LAST_ERROR_CHARS) { + Some((end, _)) => format!("{}...", &message[..end]), + None => message, + } +} + +// A partial page can omit hits sorted before the last returned one. The cursor would then move +// past them for good, so such a response must fail the poll before any cursor update. +fn ensure_complete_response(body: &Value) -> Result<(), Error> { + if body.get("timed_out").and_then(Value::as_bool) == Some(true) { + return Err(Error::HttpRequestFailed( + "OpenSearch search timed out; results may be partial".to_string(), + )); + } + let failed_shards = body + .get("_shards") + .and_then(|shards| shards.get("failed")) + .and_then(Value::as_u64) + .unwrap_or(0); + if failed_shards > 0 { + return Err(Error::HttpRequestFailed(format!( + "OpenSearch search failed on {failed_shards} shards; results may be partial" + ))); + } + Ok(()) +} + +fn restore_state(id: u32, state: Option<ConnectorState>) -> (State, Option<String>) { + let Some(connector_state) = state else { + return (State::default(), None); + }; + + match connector_state.deserialize::<State>(CONNECTOR_NAME, id) { + Some(restored) => { + info!( + "Restored state for {CONNECTOR_NAME} connector with ID: {id}. \ + Documents published: {}, poll count: {}", + restored.total_documents_published, restored.poll_count + ); + (restored, None) + } + None => { + // Starting fresh would re-publish the whole index, so refuse to open and let the + // operator repair or delete the state instead of silently resetting the cursor. + let cause = "persisted state exists but could not be deserialized. \ + Refusing to start to prevent silent cursor reset." + .to_string(); + error!("{CONNECTOR_NAME} ID {id}: {cause}"); + (State::default(), Some(cause)) + } + } +} + +fn validate_open_config(config: &OpenSearchSourceConfig) -> Result<(), Error> { + if config.timestamp_field.as_deref().is_none_or(str::is_empty) { + return Err(Error::InvalidConfigValue( + "timestamp_field is required for incremental OpenSearch polling".to_string(), + )); + } + + if let Some(batch_size) = config.batch_size + && !(1..=MAX_BATCH_SIZE).contains(&batch_size) + { + return Err(Error::InvalidConfigValue(format!( + "batch_size must be between 1 and {MAX_BATCH_SIZE}, got {batch_size}" + ))); + } + + if config + .query + .as_ref() + .is_some_and(|query| !query.is_object()) + { + return Err(Error::InvalidConfigValue( + "query must be a JSON object".to_string(), + )); + } + + Ok(()) +} + +/// Strips credentials and query from a URL before it reaches a log line, since a URL can +/// carry `user:password@` or an access token. +fn redacted_url(url: &str) -> String { + match Url::parse(url) { + Ok(mut parsed) => { + let _ = parsed.set_username(""); + let _ = parsed.set_password(None); + parsed.set_query(None); + parsed.to_string() + } + Err(_) => "<invalid url>".to_string(), + } +} + +/// Stable 128-bit message id so a consumer can drop a document replayed after a NACK or a +/// lost state save. Length-prefixed parts keep distinct (index, id) pairs from aliasing. +fn message_id(index: &str, document_id: &str) -> u128 { + let mut hasher = blake3::Hasher::new(); + for part in [index, document_id] { + hasher.update(&(part.len() as u64).to_le_bytes()); + hasher.update(part.as_bytes()); + } + let mut id = [0u8; 16]; + hasher.finalize_xof().fill(&mut id); + u128::from_le_bytes(id) +} + +/// The hashed message id cannot be mapped back to the document, so the original `_id` and +/// concrete `_index` travel as headers. A value over the header limit is omitted, the message +/// id remains. +fn document_headers( + connector_id: u32, + index: Option<&str>, + document_id: Option<&str>, +) -> Option<BTreeMap<HeaderKey, HeaderValue>> { + let mut headers = BTreeMap::new(); + for (key, name, value) in [ + (&*INDEX_HEADER_KEY, INDEX_HEADER, index), + (&*DOCUMENT_ID_HEADER_KEY, DOCUMENT_ID_HEADER, document_id), + ] { + let Some(value) = value else { continue }; + if value.len() > MAX_HEADER_VALUE_LENGTH { + warn!( + "{CONNECTOR_NAME} connector ID: {connector_id} omitted the {name} header: \ + value is {} bytes, limit is {MAX_HEADER_VALUE_LENGTH}", + value.len() + ); + continue; + } + if let Ok(header_value) = HeaderValue::try_from(value) { + headers.insert(key.clone(), header_value); + } + } + (!headers.is_empty()).then_some(headers) +} + +fn parse_document_timestamp(value: &Value) -> Option<DateTime<Utc>> { + match value { + Value::String(text) => DateTime::parse_from_rfc3339(text) + .ok() + .map(|timestamp| timestamp.with_timezone(&Utc)), + Value::Number(number) => { + let raw = number.as_i64()?; + // Values above 1e12 are already milliseconds (Unix epoch seconds won't reach + // 1e12 until year 33658). Values at or below are treated as seconds and + // multiplied by 1000. + let millis = if !(-1_000_000_000_000..=1_000_000_000_000).contains(&raw) { + raw + } else { + raw.saturating_mul(1_000) + }; + DateTime::from_timestamp_millis(millis).map(|timestamp| timestamp.with_timezone(&Utc)) + } + _ => None, + } +} + +#[async_trait] +impl Source for OpenSearchSource { + async fn open(&mut self) -> Result<(), Error> { + if let Some(ref cause) = self.state_restore_error { + return Err(Error::InitError(format!("state restore failed: {cause}"))); + } + + validate_open_config(&self.config)?; + + info!( + "Opening OpenSearch source connector with ID: {} for URL: {}, index: {}", + self.id, + redacted_url(&self.config.url), + self.config.index + ); + + let client = self.create_client().await?; + + self.check_index_exists_with_retry(&client).await?; + + let timestamp_field = self.timestamp_field()?; + let batch_size = self.batch_size(); + self.search_body_base = Some(json!({ + "query": self.config.query.clone().unwrap_or_else(|| json!({ "match_all": {} })), + "size": batch_size, + "track_total_hits": false, + "sort": [ + { timestamp_field: { "order": "asc", "missing": "_first" } }, + { "_id": { "order": "asc" } }, Review Comment: Sorting on `_id` loads all document IDs in each searched shard into heap-backed fielddata because `_id` has no doc values. Thus even a small `batch_size` can exhaust OpenSearch heap on a large index; clusters with `indices.id_field_data.enabled=false` reject every poll outright. Please support a unique, mapped `keyword` tie-breaker with doc values instead of unconditionally requiring `_id` sorting. OpenSearch documents this restriction at https://docs.opensearch.org/latest/mappings/metadata-fields/id/. -- 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]
