diff --git a/.config/nextest.toml b/.config/nextest.toml index 3070df8bdc..568c56e5da 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -48,6 +48,20 @@ max-threads = 1 filter = 'package(integration) and test(/connectors::elasticsearch::/)' test-group = "elasticsearch" +# OpenSearch tests share one reusable container (fixed name +# `iggy-test-opensearch`, ReuseDirective::Always), same reasoning as +# elasticsearch above. Serializing the group lets the first test create it and +# the rest attach by name, instead of racing to create the same name +# concurrently (Docker 409 Conflict) or hammering a single freshly-started +# instance with concurrent requests before it has stabilized. Per-test +# isolation comes from a unique index per fixture, not a fresh container. +[test-groups.opensearch] +max-threads = 1 + +[[profile.default.overrides]] +filter = 'package(integration) and test(/connectors::opensearch::/)' +test-group = "opensearch" + [profile.default] slow-timeout = { period = "60s", terminate-after = 5 } diff --git a/Cargo.lock b/Cargo.lock index 7e44d56d6f..21910fef1d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7094,6 +7094,27 @@ dependencies = [ "tracing", ] +[[package]] +name = "iggy_connector_opensearch_sink" +version = "0.5.0-edge.1" +dependencies = [ + "async-trait", + "base64", + "bytes", + "iggy_common", + "iggy_connector_sdk", + "opensearch", + "secrecy", + "serde", + "serde_json", + "simd-json", + "tokio", + "toml 1.1.3+spec-1.1.0", + "tracing", + "url", + "wiremock", +] + [[package]] name = "iggy_connector_postgres_sink" version = "0.5.0-edge.2" @@ -9227,6 +9248,26 @@ dependencies = [ "uuid", ] +[[package]] +name = "opensearch" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2af6815a23449a0860c8fe049a828c3589d3ad56d3b5875d0d1f340d1291871e" +dependencies = [ + "base64", + "bytes", + "dyn-clone", + "lazy_static", + "percent-encoding", + "reqwest 0.13.4", + "rustc_version", + "serde", + "serde_json", + "serde_with", + "url", + "void", +] + [[package]] name = "openssl-probe" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 31e37bacf1..1c9ecf760d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -42,6 +42,7 @@ members = [ "core/connectors/sinks/influxdb_sink", "core/connectors/sinks/meilisearch_sink", "core/connectors/sinks/mongodb_sink", + "core/connectors/sinks/opensearch_sink", "core/connectors/sinks/postgres_sink", "core/connectors/sinks/quickwit_sink", "core/connectors/sinks/s3_sink", @@ -233,6 +234,7 @@ nix = { version = "0.31.3", features = ["feature", "fs", "resource", "sched"] } nonzero_lit = "0.1.2" notify = "8.2.0" octocrab = "0.54.0" +opensearch = { version = "2.4.0", default-features = false, features = ["rustls-tls"] } opentelemetry = { version = "0.32.0", features = ["trace", "logs"] } opentelemetry-appender-tracing = { version = "0.32.0", features = ["log"] } opentelemetry-otlp = { version = "0.32.0", features = [ diff --git a/core/connectors/README.md b/core/connectors/README.md index 2690a97756..0cac67bb61 100644 --- a/core/connectors/README.md +++ b/core/connectors/README.md @@ -84,6 +84,7 @@ Each sink should have its own, custom configuration, which is passed along with - **Elasticsearch Sink** - sends messages to Elasticsearch indices - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog - **Meilisearch Sink** - indexes messages in Meilisearch +- **OpenSearch Sink** - indexes messages in OpenSearch for full-text search - **PostgreSQL Sink** - stores messages in PostgreSQL database tables - **Quickwit Sink** - indexes messages in Quickwit search engine - **S3 Sink** - writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces) diff --git a/core/connectors/runtime/example_config/connectors/opensearch_sink.toml b/core/connectors/runtime/example_config/connectors/opensearch_sink.toml new file mode 100644 index 0000000000..16d71a2701 --- /dev/null +++ b/core/connectors/runtime/example_config/connectors/opensearch_sink.toml @@ -0,0 +1,49 @@ +# 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. + +type = "sink" +key = "opensearch" +enabled = true +version = 0 +name = "OpenSearch sink" +path = "target/release/libiggy_connector_opensearch_sink" +verbose = false + +[[streams]] +stream = "example_stream" +topics = ["example_topic"] +schema = "json" +batch_length = 1000 +poll_interval = "5ms" +consumer_group = "opensearch_sink_connector" + +[plugin_config] +url = "http://localhost:9200" +index = "iggy_messages" +# username = "admin" +# password = "..." +# document_id_field = "order_id" +create_index_if_not_exists = true +include_metadata = true +batch_size = 1000 +timeout = "30s" +refresh = "false" +max_retries = 3 +retry_delay = "500ms" +max_retry_delay = "5s" +max_open_retries = 5 +verbose_logging = false diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md index e23e1ace9c..f9ef71450c 100644 --- a/core/connectors/sinks/README.md +++ b/core/connectors/sinks/README.md @@ -13,6 +13,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s | **iceberg_sink** | Writes data to Apache Iceberg tables via REST catalog with S3/GCS/Azure storage | | **influxdb_sink** | Writes messages to InfluxDB as line-protocol points; supports both V2 (org/bucket, Flux) and V3 (db, SQL) | | **meilisearch_sink** | Indexes messages in Meilisearch for full-text search | +| **opensearch_sink** | Indexes messages in OpenSearch for full-text search and retrieval | | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas | | **quickwit_sink** | Indexes messages in Quickwit search engine for log analytics | | **s3_sink** | Writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces) | diff --git a/core/connectors/sinks/opensearch_sink/Cargo.toml b/core/connectors/sinks/opensearch_sink/Cargo.toml new file mode 100644 index 0000000000..260fea347d --- /dev/null +++ b/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 } +tokio = { workspace = true } +tracing = { workspace = true } +url = { workspace = true } + +[dev-dependencies] +toml = { workspace = true } +wiremock = { workspace = true } diff --git a/core/connectors/sinks/opensearch_sink/README.md b/core/connectors/sinks/opensearch_sink/README.md new file mode 100644 index 0000000000..0eb58ac48d --- /dev/null +++ b/core/connectors/sinks/opensearch_sink/README.md @@ -0,0 +1,202 @@ +# OpenSearch Sink Connector + +A sink connector that consumes messages from Iggy streams and indexes them +into an OpenSearch index through the official Rust SDK. + +## Configuration + +```toml +[plugin_config] +url = "https://opensearch.example.com:9200" +index = "iggy_messages" +# username = "admin" +# password = "..." +# document_id_field = "order_id" +create_index_if_not_exists = true +include_metadata = true +batch_size = 1000 +timeout = "30s" +refresh = "false" +max_retries = 3 +retry_delay = "500ms" +max_retry_delay = "5s" +max_open_retries = 5 +verbose_logging = false +``` + +- `url`: OpenSearch base URL. A path is kept and used as the base every + request is joined onto, so OpenSearch behind a reverse-proxy subpath + (`https://proxy.example.com/opensearch`) works. Query strings and + fragments are ignored. Must not embed credentials + (`https://user:pass@host`); use `username`/`password` instead, or `open()` + fails config validation. +- `index`: Target index name. +- `username` / `password`: Optional HTTP Basic authentication. Both must be + set together, or neither; setting only one fails config validation. + `password` is a `SecretString` and is never logged. AWS SigV4 (for + AWS-managed OpenSearch / OpenSearch Serverless) is not supported; basic + auth only. +- `document_id_field`: Optional top-level payload field supplying the + document `_id`. Nested paths (for example `"order.id"`) are not supported; + only a direct top-level key is looked up. The value must be a string, + number, or boolean; must not be empty or exceed 512 bytes (OpenSearch's + `_id` limit). When absent from a message (or unconfigured), the connector + falls back to a generated, deterministic `_id`. See + [Behavior](#behavior) below for what each mode buys you. +- `create_index_if_not_exists`: Create the index during `open()` when + missing. Defaults to `true`. When `false` and the index does not exist, + `open()` fails. +- `index_mapping`: Optional OpenSearch index mapping body, applied when the + index is created. For example: + + ```toml + [plugin_config.index_mapping.mappings.properties.count] + type = "integer" + ``` + +- `include_metadata`: Add `iggy_*` provenance fields to each document. + Defaults to `true`. +- `batch_size`: Maximum documents per OpenSearch `_bulk` request. Defaults + to `1000`. +- `timeout`: Per-request timeout as a humantime string, for example `30s`. + Defaults to `30s`. Applies to every request the connector makes, including + reading the response body. +- `refresh`: OpenSearch bulk `refresh` parameter: `"false"` (default), + `"true"`, or `"wait_for"`. `"wait_for"` blocks the bulk response until the + write is visible to search, but only at the *next* scheduled OpenSearch + refresh cycle (default interval ~1s), not immediately. Do not assume a + fixed short delay is enough to observe a write via `_search` afterward. +- `max_retries`: Maximum transient retries for a bulk request after the + initial attempt. Defaults to `3`. +- `retry_delay` / `max_retry_delay`: Exponential backoff bounds for + transient retries. Defaults to `500ms` / `5s`. If configured with + `retry_delay > max_retry_delay`, the values are swapped and a warning is + logged. +- `max_open_retries`: Maximum transient retries for each `open()`-time + request: the cluster health check, the index existence check, and index + creation. Defaults to `5`. A permanent failure (for example a rejected + index mapping) is not retried and fails `open()` immediately. +- `verbose_logging`: Log per-batch receive and index counts at `info` instead + of `debug`. Defaults to `false`. + +## Required privileges + +With the security plugin enabled, the configured user needs +`cluster_composite_ops` (for `_bulk`) at cluster scope, plus `crud`, +`create_index`, and `indices:admin/get` on the target index pattern. + +`open()` also probes `GET /_cluster/health`, which needs the cluster-scoped +`cluster:monitor/health` privilege. That privilege is deliberately *not* +required: a `403` on the probe still proves the cluster answered and the +credentials authenticated, so the connector logs a warning and continues. A +`401` (credentials rejected) still fails `open()`. + +## Behavior + +`Payload::Json` object values are indexed as documents; non-object JSON +(arrays, scalars) is wrapped under a `value` field, since OpenSearch +documents must be objects. `Payload::Raw` bytes are parsed as JSON when +possible, otherwise indexed as `{data: , data_type: "raw", +data_encoding: "base64"}`. `Payload::Text` is indexed as `{text, data_type: +"text"}`. Unsupported payload schemas (Protobuf, FlatBuffer, Avro) are +dropped with a warning and counted as sink errors. This matches the +connector runtime's per-record drop convention, and because the sink returns +success after dropping such a record, the runtime commits the consumer +offset for it. There is no dead-letter queue for these drops. + +### Document ID + +When `document_id_field` names a field present in the payload, that value +(stringified) becomes the document `_id`. Otherwise the connector generates +a deterministic `_id` by hashing the exact Iggy stream, topic, partition, +offset, and message ID (blake3, hex-encoded), prefixed `iggy_`. Hashing keeps +the ID a fixed length regardless of how long the stream and topic names are; +encoding them verbatim could otherwise exceed OpenSearch's 512-byte `_id` +limit for long names. Because bulk `index` upserts on a repeated `_id`, both +modes make replaying the same message idempotent rather than duplicating it: +the generated ID is stable for a given stream, topic, partition, offset, and +message ID, and the natural-key ID is stable for a given `document_id_field` +value. The natural-key path is covered end to end against a live server by +`connectors::opensearch::opensearch_sink` in the integration suite, which +resends a payload under an existing `order_id` at a different Iggy offset and +asserts the document count is unchanged. + +If a user-provided `document_id_field` value collides across otherwise +distinct messages, those messages will collapse into a single document. +Operators choosing this field are responsible for its uniqueness. + +### Metadata + +When `include_metadata` is enabled, the connector writes reserved `iggy_*` +fields after payload parsing, overwriting any same-named payload fields so +provenance reflects the true stream, topic, partition, offset, checksum, and +timestamps: `iggy_message_id`, `iggy_offset`, `iggy_stream`, `iggy_topic`, +`iggy_partition`, `iggy_checksum` (stored as a string, since checksums +exceed the precision JSON numbers can represent exactly), `iggy_timestamp`, +`iggy_origin_timestamp`, `iggy_ingested_at`, and `iggy_headers` when the +message carries headers. + +Each header becomes a field under `iggy_headers`, named after the header key. +Every value uses the same shape, `{data: , data_encoding: "utf8" | +"base64"}`: raw binary values are base64-encoded, every other header kind is +stringified and marked `utf8`. The shape is uniform on purpose. OpenSearch +pins `iggy_headers.` to whatever type the first indexed document uses, so +a per-kind shape would make every later message using the other kind fail with +a `mapper_parsing_exception` and be dropped. + +Two hazards remain, both inherent to using header keys as field names: + +- A header key containing a `.` is read as a path. A key `a.b` alongside a key + `a` produces `object mapping for [iggy_headers.a] tried to parse field [a] as + object, but found a concrete value`, and the second document is dropped. +- Header key cardinality is unbounded, but an index has a + `index.mapping.total_fields.limit` (1000 by default). Past that limit, + documents introducing a new header key fail with `illegal_argument_exception` + and are dropped. Set `include_metadata = false`, or raise the limit through + `index_mapping`, if messages carry high-cardinality header keys. + +Both are permanent per-item failures, so they are subject to the visibility +caveat in [Delivery Semantics](#delivery-semantics). + +**The timestamp fields do not share a unit.** `iggy_timestamp` and +`iggy_origin_timestamp` come from the Iggy message header and are +**microseconds** since the Unix epoch; `iggy_ingested_at` is stamped by this +connector and is **milliseconds**, matching the `elasticsearch_sink` and +`meilisearch_sink` convention. All three are indexed as `long`, so subtracting +one from another without converting first is off by a factor of 1000. + +## Delivery Semantics + +**A batch that fails at `open()` time is visible.** A missing index with +`create_index_if_not_exists = false`, a persistently unreachable cluster, +and similar setup failures correctly flip the connector to +`ConnectorStatus::Error`, reported via the runtime's `/sinks` endpoint. + +**A batch that fails at `consume()` time is not currently visible anywhere +except this connector's own logs.** This is not specific to this connector; +it is how the shared connectors runtime invokes every sink's `consume()` +over FFI today: the plugin's returned status is not propagated to +`ConnectorStatus`, `last_error`, or the `/stats` `errors` counter, and the +runtime does not hold back or redeliver the failed batch. Verified against a +live server: a batch containing a real OpenSearch `mapper_parsing_exception` +was correctly classified and logged by this connector as a +`PermanentHttpError`, and the connector continued consuming and indexing +later messages normally, with `ConnectorStatus` staying `Running` +throughout. **Operators must monitor this connector's own `tracing` output +(`error!` at target `iggy_connector_opensearch_sink`) to detect indexing +failures; the runtime's own status and stats APIs will not show them.** + +Within a single `consume()` call, a `_bulk` request can return HTTP 200 while +individual documents fail. This connector parses the per-item `items[]` +results rather than trusting the top-level status, so a batch with one bad +document among many still indexes the valid ones. Item failures are +classified as retryable (HTTP 429/5xx) or permanent (everything else, +including OpenSearch mapping/parsing errors). Retryable items are resent on +their own, under the same `max_retries` and backoff bounds that govern a +whole-request failure, so a `429` from a saturated indexing queue does not +cost those documents. Permanent item failures are never resent. Anything +still failing once retries are exhausted is reported as a failed document for +the batch, subject to the visibility caveat above. The close-time +`documents_indexed` counter counts successful bulk index operations, not +distinct documents: replaying the same batch twice counts twice even though +the document count in OpenSearch does not change. diff --git a/core/connectors/sinks/opensearch_sink/config.toml b/core/connectors/sinks/opensearch_sink/config.toml new file mode 100644 index 0000000000..1a37d3e12e --- /dev/null +++ b/core/connectors/sinks/opensearch_sink/config.toml @@ -0,0 +1,46 @@ +# 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. + +type = "sink" +key = "opensearch" +enabled = true +version = 0 +name = "OpenSearch sink" +path = "../../target/release/libiggy_connector_opensearch_sink" +verbose = false + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "opensearch_sink" + +[plugin_config] +url = "http://localhost:9200" +index = "iggy_messages" +create_index_if_not_exists = true +include_metadata = true +batch_size = 1000 +timeout = "30s" +refresh = "false" +max_retries = 3 +retry_delay = "500ms" +max_retry_delay = "5s" +max_open_retries = 5 +verbose_logging = false diff --git a/core/connectors/sinks/opensearch_sink/src/lib.rs b/core/connectors/sinks/opensearch_sink/src/lib.rs new file mode 100644 index 0000000000..6d4818284c --- /dev/null +++ b/core/connectors/sinks/opensearch_sink/src/lib.rs @@ -0,0 +1,2691 @@ +// 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, + pub password: Option, + pub document_id_field: Option, + pub create_index_if_not_exists: Option, + pub index_mapping: Option, + pub include_metadata: Option, + pub batch_size: Option, + pub timeout: Option, + pub refresh: Option, + pub max_retries: Option, + pub retry_delay: Option, + pub max_retry_delay: Option, + pub max_open_retries: Option, + pub verbose_logging: Option, +} + +#[derive(Debug)] +pub struct OpenSearchSink { + id: u32, + config: ResolvedOpenSearchSinkConfig, + client: Option, + invocations_count: AtomicU64, + documents_indexed: AtomicU64, + errors_count: AtomicU64, +} + +#[derive(Debug)] +struct ResolvedOpenSearchSinkConfig { + url: String, + index: String, + username: Option, + password: Option, + document_id_field: Option, + create_index_if_not_exists: bool, + index_mapping: Option, + include_metadata: bool, + batch_size: usize, + timeout: Duration, + refresh: Option, + max_retries: u32, + retry_delay: Duration, + max_retry_delay: Duration, + max_open_retries: u32, + verbose_logging: bool, +} + +impl From 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 { + 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(&self, operation: &str, call: F) -> Result + where + F: Fn() -> Fut, + Fut: Future>, + { + 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 { + 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 { + 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, + ) -> Result, 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, + ) -> Result { + 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 = 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 { + // 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::().await { + Ok(payload) => { + let attempt = parse_bulk_response(&payload); + 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 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`. + 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, + 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, + ) -> 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, +} + +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 { + 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, + retryable: Vec, + first_retryable_failure: Option, +} + +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 { + 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() +} + +fn parse_bulk_response(response: &Value) -> BulkAttempt { + let Some(items) = response.get("items").and_then(Value::as_array) else { + return BulkAttempt::default(); + }; + + // 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 BulkAttempt { + indexed: items.len(), + ..BulkAttempt::default() + }; + } + + let mut attempt = BulkAttempt::default(); + for (position, item) in items.iter().enumerate() { + let Some(result) = item.as_object().and_then(|item| item.values().next()) else { + continue; + }; + let status = result.get("status").and_then(Value::as_u64).unwrap_or(0) as u16; + if result.get("error").is_none() && (200..300).contains(&status) { + attempt.indexed += 1; + continue; + } + + let reason = result + .get("error") + .and_then(|error| error.get("reason")) + .and_then(Value::as_str) + .unwrap_or("unknown error"); + let failure = format!("status {status}: {reason}"); + + if StatusCode::from_u16(status).is_ok_and(is_transient_status) { + attempt.retryable.push(position); + if attempt.first_retryable_failure.is_none() { + attempt.first_retryable_failure = Some(failure); + } + continue; + } + + attempt.permanent_failed += 1; + if attempt.first_permanent_failure.is_none() { + attempt.first_permanent_failure = Some(failure); + } + } + + attempt +} + +fn document_from_json(value: Value) -> Map { + match value { + Value::Object(object) => object, + other => Map::from_iter([("value".to_string(), other)]), + } +} + +fn document_from_raw(bytes: Vec) -> Map { + // simd_json parses destructively, so the base64 fallback needs its own copy. + let mut parse_buffer = bytes.clone(); + match simd_json::to_owned_value(&mut parse_buffer) { + Ok(value) => document_from_json(owned_value_to_serde_json(&value)), + Err(_) => Map::from_iter([ + ( + "data".to_string(), + Value::String(general_purpose::STANDARD.encode(&bytes)), + ), + ("data_type".to_string(), Value::String("raw".to_string())), + ( + "data_encoding".to_string(), + Value::String(ENCODING_BASE64.to_string()), + ), + ]), + } +} + +/// Writes the reserved `iggy_*` provenance fields, overwriting any same-named +/// payload fields so provenance always reflects the true message coordinates. +fn inject_metadata( + document: &mut Map, + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + message: &ConsumedMessage, +) { + let fields = [ + ("iggy_message_id", Value::String(message.id.to_string())), + ("iggy_offset", Value::from(message.offset)), + ("iggy_stream", Value::from(topic_metadata.stream.as_str())), + ("iggy_topic", Value::from(topic_metadata.topic.as_str())), + ( + "iggy_partition", + Value::from(messages_metadata.partition_id), + ), + // A string: checksums exceed the 2^53 range JSON represents exactly. + ("iggy_checksum", Value::String(message.checksum.to_string())), + ("iggy_timestamp", Value::from(message.timestamp)), + ( + "iggy_origin_timestamp", + Value::from(message.origin_timestamp), + ), + ( + "iggy_ingested_at", + Value::from(IggyTimestamp::now().as_millis() as i64), + ), + ]; + + for (field, value) in fields { + upsert_metadata_field(document, field, value); + } + + if let Some(headers) = &message.headers + && !headers.is_empty() + { + upsert_metadata_field(document, "iggy_headers", headers_to_json(headers)); + } +} + +/// `HeaderKey`/`HeaderValue` are structs, not strings, so `serde_json::to_value` +/// on the map directly fails with "key must be a string", so the keys and raw +/// binary values are converted explicitly instead. +fn headers_to_json(headers: &BTreeMap) -> Value { + let map: Map = headers + .iter() + .map(|(key, value)| { + // A per-kind shape would pin iggy_headers. to text or object on + // first use, so the other kind then fails to index under that key. + let (data, encoding) = match value.as_raw() { + Ok(raw) => (general_purpose::STANDARD.encode(raw), ENCODING_BASE64), + Err(_) => (value.to_string_value(), ENCODING_UTF8), + }; + ( + key.to_string_value(), + json!({ "data": data, "data_encoding": encoding }), + ) + }) + .collect(); + Value::Object(map) +} + +fn upsert_metadata_field(document: &mut Map, field: &str, value: Value) { + if document.insert(field.to_string(), value).is_some() { + debug!("Overwriting payload field '{field}' with OpenSearch connector provenance"); + } +} + +/// Stream and topic names can each reach `iggy_common::MAX_NAME_LENGTH` +/// (255), so encoding them into the ID verbatim can exceed +/// `MAX_DOCUMENT_ID_BYTES`. Hashing keeps the ID a fixed 69 bytes +/// (`GENERATED_ID_PREFIX` + 64 hex chars) regardless of input length, while +/// staying deterministic: same stream, topic, partition, offset, and message +/// ID always produce the same ID, preserving the upsert-based idempotency +/// this function exists for. +fn generated_document_id( + topic_metadata: &TopicMetadata, + messages_metadata: &MessagesMetadata, + offset: u64, + message_id: u128, +) -> Result { + let components = json!([ + topic_metadata.stream.as_str(), + topic_metadata.topic.as_str(), + messages_metadata.partition_id, + offset, + message_id.to_string() + ]); + let bytes = serde_json::to_vec(&components).map_err(|error| { + Error::Serialization(format!( + "Failed to serialize generated document ID: {error}" + )) + })?; + Ok(format!("{GENERATED_ID_PREFIX}{}", calculate_256(&bytes))) +} + +fn is_transient_error(error: &Error) -> bool { + matches!(error, Error::HttpRequestFailed(_)) +} + +fn is_transient_client_error(error: &opensearch::Error) -> bool { + if error.is_timeout() { + return true; + } + match error.status_code() { + Some(status) => is_transient_status(status), + None => true, + } +} + +fn map_client_error(operation: &str, error: opensearch::Error) -> Error { + if is_transient_client_error(&error) { + Error::HttpRequestFailed(format!("OpenSearch {operation} failed: {error}")) + } else { + Error::PermanentHttpError(format!("OpenSearch {operation} failed: {error}")) + } +} + +fn map_status_error(operation: &str, status: StatusCode, body: &str) -> Error { + let message = format!("OpenSearch {operation} failed with status {status}: {body}"); + if is_transient_status(status) { + Error::HttpRequestFailed(message) + } else { + Error::PermanentHttpError(message) + } +} + +fn is_index_already_exists_error(body: &str) -> bool { + let Ok(value) = serde_json::from_str::(body) else { + return false; + }; + value + .get("error") + .and_then(|error| error.get("type")) + .and_then(Value::as_str) + == Some(INDEX_ALREADY_EXISTS_ERROR) +} + +fn trimmed_non_empty(value: Option) -> Option { + value + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()) +} + +// Only checks blankness, never trims: unlike a username, trimming a +// password would silently alter a secret that may legitimately contain +// leading/trailing whitespace. +fn is_blank_secret(value: &SecretString) -> bool { + value.expose_secret().trim().is_empty() +} + +fn normalize_url(raw: &str) -> Result { + let trimmed = raw.trim(); + if trimmed.is_empty() { + return Err(Error::Connection( + "Invalid OpenSearch URL: host cannot be empty".to_string(), + )); + } + + let with_scheme = if trimmed.starts_with("http://") || trimmed.starts_with("https://") { + trimmed.to_string() + } else { + format!("http://{trimmed}") + }; + let mut url = Url::parse(&with_scheme) + .map_err(|error| Error::Connection(format!("Invalid OpenSearch URL: {error}")))?; + // Rejected rather than silently honored: reqwest promotes URL userinfo + // into a real `Authorization: Basic` header on every request, which + // would bypass both `warn_if_credentials_use_insecure_http` (gated on + // the dedicated `password` field) and the `SecretString` redaction the + // `username`/`password` config fields get. + if !url.username().is_empty() || url.password().is_some() { + return Err(Error::InvalidConfigValue( + "OpenSearch URL must not embed credentials; use the username/password config fields instead".to_string(), + )); + } + if url.query().is_some() || url.fragment().is_some() { + warn!("Ignoring the query string and fragment on the OpenSearch URL"); + } + url.set_query(None); + url.set_fragment(None); + + // Kept, not stripped: the transport joins request paths onto it, so a proxy subpath works. + if url.path() != "/" { + warn!( + "Using '{}' as the OpenSearch base path, so requests are sent to paths like '{}/_bulk'", + url.path(), + url.path().trim_end_matches('/') + ); + } + Ok(url.as_str().trim_end_matches('/').to_string()) +} + +/// The stripping below is unreachable from `open()`'s only call site: `normalize_url` +/// already rejects embedded credentials before a URL gets here. Kept as defense in +/// depth for any future caller that doesn't route through `normalize_url` first. +fn sanitize_url_for_log(normalized: &str) -> String { + let Ok(mut url) = Url::parse(normalized) else { + return "".to_string(); + }; + + if !url.username().is_empty() { + let _ = url.set_username(""); + } + if url.password().is_some() { + let _ = url.set_password(None); + } + url.to_string().trim_end_matches('/').to_string() +} + +fn warn_if_credentials_use_insecure_http(raw: &str, normalized: &str, has_password: bool) { + if !has_password { + return; + } + + let Ok(url) = Url::parse(normalized) else { + return; + }; + if url.scheme() != "http" { + return; + } + let Some(host) = url.host_str() else { + return; + }; + if is_loopback_host(host) { + return; + } + + let scheme_hint = if raw.trim().starts_with("http://") { + "explicit http://" + } else { + "implicit http://" + }; + warn!( + "OpenSearch credentials are configured with {scheme_hint} for non-loopback host '{host}'. They will be sent without TLS; use https:// unless this is intentional." + ); +} + +fn is_loopback_host(host: &str) -> bool { + host.eq_ignore_ascii_case("localhost") + || host + .parse::() + .map(|address| address.is_loopback()) + .unwrap_or(false) +} + +#[cfg(test)] +mod tests { + use super::*; + use iggy_connector_sdk::Schema; + use std::sync::atomic::AtomicU32; + use wiremock::matchers::{body_bytes, method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + fn topic_metadata() -> TopicMetadata { + TopicMetadata { + stream: "orders.stream".to_string(), + topic: "created/topic".to_string(), + } + } + + fn messages_metadata() -> MessagesMetadata { + MessagesMetadata { + partition_id: 7, + current_offset: 10, + schema: Schema::Json, + } + } + + fn message(payload: Payload) -> ConsumedMessage { + ConsumedMessage { + id: 42, + offset: 11, + checksum: 12, + timestamp: 13, + origin_timestamp: 14, + headers: None, + payload, + } + } + + fn base_config() -> OpenSearchSinkConfig { + OpenSearchSinkConfig { + url: "http://localhost:9200".to_string(), + index: "iggy_messages".to_string(), + ..Default::default() + } + } + + fn sink_with_config(config: OpenSearchSinkConfig) -> OpenSearchSink { + OpenSearchSink::new(1, config) + } + + fn expected_generated_id(message: &ConsumedMessage) -> String { + generated_document_id( + &topic_metadata(), + &messages_metadata(), + message.offset, + message.id, + ) + .expect("generated ID components should serialize") + } + + #[test] + fn given_empty_config_should_apply_documented_defaults() { + let sink = sink_with_config(base_config()); + + assert!(sink.config.create_index_if_not_exists); + assert!(sink.config.include_metadata); + assert_eq!(sink.config.batch_size, DEFAULT_BATCH_SIZE); + assert_eq!(sink.config.timeout, Duration::from_secs(30)); + assert_eq!(sink.config.max_retries, DEFAULT_MAX_RETRIES); + assert_eq!(sink.config.retry_delay, Duration::from_millis(500)); + assert_eq!(sink.config.max_retry_delay, Duration::from_secs(5)); + assert_eq!(sink.config.max_open_retries, DEFAULT_MAX_OPEN_RETRIES); + assert!(sink.config.refresh.is_none()); + assert!(sink.config.document_id_field.is_none()); + } + + #[test] + fn given_zero_batch_size_should_clamp_to_one() { + let mut config = base_config(); + config.batch_size = Some(0); + + assert_eq!(sink_with_config(config).config.batch_size, 1); + } + + #[test] + fn given_reversed_retry_delays_should_swap_them() { + let mut config = base_config(); + config.retry_delay = Some("10s".to_string()); + config.max_retry_delay = Some("1s".to_string()); + + let sink = sink_with_config(config); + + assert_eq!(sink.config.retry_delay, Duration::from_secs(1)); + assert_eq!(sink.config.max_retry_delay, Duration::from_secs(10)); + } + + #[test] + fn given_blank_index_should_fail_validation() { + let mut config = base_config(); + config.index = " ".to_string(); + + let error = sink_with_config(config) + .validate_config() + .expect_err("blank index should be rejected"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_username_without_password_should_fail_validation() { + let mut config = base_config(); + config.username = Some("admin".to_string()); + + let error = sink_with_config(config) + .validate_config() + .expect_err("username without password should be rejected"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_blank_password_with_username_should_fail_validation() { + let mut config = base_config(); + config.username = Some("admin".to_string()); + config.password = Some(SecretString::from(" ")); + + let error = sink_with_config(config) + .validate_config() + .expect_err("a whitespace-only password should be treated as unset"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_password_without_username_should_fail_validation() { + let mut config = base_config(); + config.password = Some(SecretString::from("secret")); + + let error = sink_with_config(config) + .validate_config() + .expect_err("password without username should be rejected"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_json_payload_should_inject_metadata_and_generated_id() { + let sink = sink_with_config(base_config()); + let message = message(Payload::Json(simd_json::json!({ "name": "Alice" }))); + let expected_id = expected_generated_id(&message); + + let prepared = sink + .prepare_document(&topic_metadata(), &messages_metadata(), message) + .expect("prepare document"); + + assert_eq!(prepared.id, expected_id); + assert_eq!(prepared.document["name"], "Alice"); + assert_eq!(prepared.document["iggy_offset"], 11); + assert_eq!(prepared.document["iggy_stream"], "orders.stream"); + assert_eq!(prepared.document["iggy_topic"], "created/topic"); + assert_eq!(prepared.document["iggy_partition"], 7); + assert_eq!(prepared.document["iggy_checksum"], "12"); + assert_eq!(prepared.document["iggy_message_id"], "42"); + } + + #[test] + fn given_include_metadata_disabled_should_omit_provenance_fields() { + let mut config = base_config(); + config.include_metadata = Some(false); + let sink = sink_with_config(config); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "name": "Alice" }))), + ) + .expect("prepare document"); + + assert_eq!(prepared.document["name"], "Alice"); + assert!(prepared.document.get("iggy_offset").is_none()); + assert!(prepared.document.get("iggy_stream").is_none()); + assert!(prepared.id.starts_with(GENERATED_ID_PREFIX)); + } + + #[test] + fn given_payload_reusing_metadata_names_should_overwrite_with_provenance() { + let sink = sink_with_config(base_config()); + let payload = Payload::Json(simd_json::json!({ + "iggy_offset": 999, + "iggy_stream": "spoofed", + "iggy_checksum": 999 + })); + + let prepared = sink + .prepare_document(&topic_metadata(), &messages_metadata(), message(payload)) + .expect("prepare document"); + + assert_eq!(prepared.document["iggy_offset"], 11); + assert_eq!(prepared.document["iggy_stream"], "orders.stream"); + assert_eq!(prepared.document["iggy_checksum"], "12"); + } + + #[test] + fn given_non_object_json_should_wrap_in_value_field() { + let sink = sink_with_config(base_config()); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!(["a", "b"]))), + ) + .expect("prepare document"); + + assert_eq!(prepared.document["value"], json!(["a", "b"])); + } + + #[test] + fn given_raw_json_payload_should_index_parsed_document() { + let sink = sink_with_config(base_config()); + let bytes = br#"{"name":"from-raw"}"#.to_vec(); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Raw(bytes)), + ) + .expect("prepare document"); + + assert_eq!(prepared.document["name"], "from-raw"); + } + + #[test] + fn given_raw_non_json_payload_should_base64_encode_original_bytes() { + let sink = sink_with_config(base_config()); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Raw(vec![0, 1, 2, 3])), + ) + .expect("prepare document"); + + assert_eq!(prepared.document["data"], "AAECAw=="); + assert_eq!(prepared.document["data_encoding"], ENCODING_BASE64); + assert_eq!(prepared.document["data_type"], "raw"); + } + + #[test] + fn given_truncated_json_raw_payload_should_preserve_original_bytes() { + let sink = sink_with_config(base_config()); + let bytes = b"[1,2,3".to_vec(); + let expected = general_purpose::STANDARD.encode(&bytes); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Raw(bytes)), + ) + .expect("prepare document"); + + assert_eq!(prepared.document["data"], expected); + } + + #[test] + fn given_message_with_headers_should_index_iggy_headers_field() { + let sink = sink_with_config(base_config()); + let mut message = message(Payload::Json(simd_json::json!({ "name": "Alice" }))); + message.headers = Some(BTreeMap::from([ + ( + HeaderKey::try_from("x-correlation-id").unwrap(), + HeaderValue::try_from("abc-123").unwrap(), + ), + ( + HeaderKey::try_from("x-raw").unwrap(), + HeaderValue::try_from([1u8, 2, 3].as_slice()).unwrap(), + ), + ])); + + let prepared = sink + .prepare_document(&topic_metadata(), &messages_metadata(), message) + .expect("prepare document"); + + let headers = &prepared.document["iggy_headers"]; + assert_eq!(headers["x-correlation-id"]["data"], "abc-123"); + assert_eq!(headers["x-correlation-id"]["data_encoding"], ENCODING_UTF8); + assert_eq!( + headers["x-raw"]["data"], + general_purpose::STANDARD.encode([1u8, 2, 3]) + ); + assert_eq!(headers["x-raw"]["data_encoding"], ENCODING_BASE64); + } + + // OpenSearch pins iggy_headers. to whatever type the first document + // uses, so a string header and a raw header must serialize to the same + // shape or the second kind fails with a mapper_parsing_exception. + #[test] + fn given_string_and_raw_header_values_should_use_one_document_shape() { + let headers = headers_to_json(&BTreeMap::from([ + ( + HeaderKey::try_from("x-text").unwrap(), + HeaderValue::try_from("plain").unwrap(), + ), + ( + HeaderKey::try_from("x-number").unwrap(), + HeaderValue::from(7u32), + ), + ( + HeaderKey::try_from("x-raw").unwrap(), + HeaderValue::try_from([1u8, 2, 3].as_slice()).unwrap(), + ), + ])); + + let entries = headers.as_object().expect("headers should be an object"); + assert_eq!(entries.len(), 3); + for (key, value) in entries { + let entry = value + .as_object() + .unwrap_or_else(|| panic!("header '{key}' should be an object")); + assert_eq!( + entry.keys().map(String::as_str).collect::>(), + ["data", "data_encoding"], + "header '{key}' has a divergent shape" + ); + assert!( + entry["data"].is_string(), + "header '{key}' data is not a string" + ); + } + assert_eq!(entries["x-text"]["data_encoding"], ENCODING_UTF8); + assert_eq!(entries["x-number"]["data_encoding"], ENCODING_UTF8); + assert_eq!(entries["x-raw"]["data_encoding"], ENCODING_BASE64); + } + + #[test] + fn given_message_without_headers_should_omit_iggy_headers_field() { + let sink = sink_with_config(base_config()); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "name": "Alice" }))), + ) + .expect("prepare document"); + + assert!(prepared.document.get("iggy_headers").is_none()); + } + + #[test] + fn given_text_payload_should_index_text_field() { + let sink = sink_with_config(base_config()); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Text("hello".to_string())), + ) + .expect("prepare document"); + + assert_eq!(prepared.document["text"], "hello"); + assert_eq!(prepared.document["data_type"], "text"); + } + + #[test] + fn given_unsupported_payload_should_return_invalid_record() { + let sink = sink_with_config(base_config()); + + let error = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Avro(vec![1, 2, 3])), + ) + .expect_err("unsupported payload should be rejected"); + + assert!(matches!(error, Error::InvalidRecordValue(_))); + } + + #[test] + fn given_document_id_field_should_use_payload_value_as_id() { + let mut config = base_config(); + config.document_id_field = Some("order_id".to_string()); + let sink = sink_with_config(config); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "order_id": "A-1" }))), + ) + .expect("prepare document"); + + assert_eq!(prepared.id, "A-1"); + } + + #[test] + fn given_numeric_document_id_field_should_stringify_it() { + let mut config = base_config(); + config.document_id_field = Some("order_id".to_string()); + let sink = sink_with_config(config); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "order_id": 17 }))), + ) + .expect("prepare document"); + + assert_eq!(prepared.id, "17"); + } + + #[test] + fn given_boolean_document_id_field_should_stringify_it() { + let mut config = base_config(); + config.document_id_field = Some("archived".to_string()); + let sink = sink_with_config(config); + + let prepared = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "archived": true }))), + ) + .expect("prepare document"); + + assert_eq!(prepared.id, "true"); + } + + #[test] + fn given_missing_document_id_field_should_fall_back_to_generated_id() { + let mut config = base_config(); + config.document_id_field = Some("order_id".to_string()); + let sink = sink_with_config(config); + let message = message(Payload::Json(simd_json::json!({ "name": "Alice" }))); + let expected_id = expected_generated_id(&message); + + let prepared = sink + .prepare_document(&topic_metadata(), &messages_metadata(), message) + .expect("prepare document"); + + assert_eq!(prepared.id, expected_id); + } + + #[test] + fn given_oversized_document_id_field_should_return_invalid_record() { + let mut config = base_config(); + config.document_id_field = Some("order_id".to_string()); + let sink = sink_with_config(config); + let oversized = "x".repeat(MAX_DOCUMENT_ID_BYTES + 1); + + let error = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "order_id": oversized }))), + ) + .expect_err("oversized document ID should be rejected"); + + assert!(matches!(error, Error::InvalidRecordValue(_))); + } + + #[test] + fn given_object_document_id_field_should_return_invalid_record() { + let mut config = base_config(); + config.document_id_field = Some("order_id".to_string()); + let sink = sink_with_config(config); + + let error = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json( + simd_json::json!({ "order_id": { "nested": true } }), + )), + ) + .expect_err("non-scalar document ID should be rejected"); + + assert!(matches!(error, Error::InvalidRecordValue(_))); + } + + // Bulk `index` upserts on a repeated `_id`, so a stable ID across replays + // of the same offset is what makes redelivery idempotent. + #[test] + fn given_replayed_message_at_same_offset_should_prepare_same_document_id() { + let sink = sink_with_config(base_config()); + + let first = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "name": "Alice" }))), + ) + .expect("prepare document"); + let second = sink + .prepare_document( + &topic_metadata(), + &messages_metadata(), + message(Payload::Json(simd_json::json!({ "name": "Alice" }))), + ) + .expect("prepare document"); + + assert_eq!(first.id, second.id); + assert!(first.id.starts_with(GENERATED_ID_PREFIX)); + } + + #[test] + fn given_same_message_should_generate_stable_id() { + let first = expected_generated_id(&message(Payload::Text("x".to_string()))); + let second = expected_generated_id(&message(Payload::Text("different".to_string()))); + + assert_eq!(first, second); + assert!(first.starts_with(GENERATED_ID_PREFIX)); + assert!( + first + .chars() + .all(|character| character.is_ascii_alphanumeric() + || character == '-' + || character == '_') + ); + assert!(first.len() <= MAX_DOCUMENT_ID_BYTES); + } + + #[test] + fn given_separator_shuffled_names_should_not_collide() { + let first = TopicMetadata { + stream: "orders.stream".to_string(), + topic: "created/topic".to_string(), + }; + let second = TopicMetadata { + stream: "orders/stream".to_string(), + topic: "created.topic".to_string(), + }; + + let first_id = generated_document_id(&first, &messages_metadata(), 11, 42).unwrap(); + let second_id = generated_document_id(&second, &messages_metadata(), 11, 42).unwrap(); + + assert_ne!(first_id, second_id); + } + + #[test] + fn given_max_u128_message_id_should_generate_id() { + let id = generated_document_id(&topic_metadata(), &messages_metadata(), 11, u128::MAX) + .expect("u128::MAX should serialize"); + + assert!(id.starts_with(GENERATED_ID_PREFIX)); + } + + #[test] + fn given_max_length_stream_and_topic_names_should_generate_id_within_byte_limit() { + // iggy_common::MAX_NAME_LENGTH is 255; encoding two such names + // verbatim (as the pre-hash implementation did) would exceed + // MAX_DOCUMENT_ID_BYTES once base64-expanded. Hashing keeps the ID a + // constant length regardless of input length. + let topic_metadata = TopicMetadata { + stream: "s".repeat(255), + topic: "t".repeat(255), + }; + + let id = generated_document_id(&topic_metadata, &messages_metadata(), u64::MAX, u128::MAX) + .expect("max-length names should serialize"); + + assert!(id.len() <= MAX_DOCUMENT_ID_BYTES); + assert!(id.starts_with(GENERATED_ID_PREFIX)); + } + + // The bulk-response fixtures below are verbatim captures from OpenSearch + // 3.8.0; the shapes are measured, not assumed. + + #[test] + fn given_clean_bulk_response_should_count_every_document_as_indexed() { + let response = json!({ + "took": 4, + "errors": false, + "items": [ + { "index": { "_index": "iggy_probe", "_id": "iggy_abc", "_version": 1, "result": "created", "status": 201 } }, + { "index": { "_index": "iggy_probe", "_id": "iggy_def", "_version": 1, "result": "created", "status": 201 } } + ] + }); + + let attempt = parse_bulk_response(&response); + + assert_eq!( + attempt, + BulkAttempt { + indexed: 2, + permanent_failed: 0, + first_permanent_failure: None, + retryable: Vec::new(), + first_retryable_failure: None, + } + ); + assert!(attempt.into_outcome().into_error("iggy_probe").is_none()); + } + + // Unlike the `errors: false` fast path (see the test above), this forces the + // per-item scan by mixing a replay with a fresh insert under `errors: true`, + // proving that scan also ignores `result` and counts both as indexed. + #[test] + fn given_replayed_bulk_response_should_count_updates_as_indexed() { + let response = json!({ + "took": 3, + "errors": true, + "items": [ + { "index": { "_id": "iggy_abc", "_version": 2, "result": "updated", "status": 200 } }, + { "index": { "_id": "iggy_def", "_version": 1, "result": "created", "status": 201 } } + ] + }); + + assert_eq!(parse_bulk_response(&response).indexed, 2); + } + + #[test] + fn given_mixed_bulk_response_should_account_per_item() { + let response = json!({ + "took": 6, + "errors": true, + "items": [ + { "index": { + "_id": "iggy_bad", + "status": 400, + "error": { + "type": "mapper_parsing_exception", + "reason": "failed to parse field [count] of type [integer] in document with id 'iggy_bad'. Preview of field's value: 'not-an-integer'" + } + } }, + { "index": { "_id": "iggy_ok", "_version": 1, "result": "created", "status": 201 } } + ] + }); + + let attempt = parse_bulk_response(&response); + + assert_eq!(attempt.indexed, 1); + assert_eq!(attempt.permanent_failed, 1); + assert!(attempt.retryable.is_empty()); + assert!( + attempt + .first_permanent_failure + .as_deref() + .is_some_and(|failure| failure.contains("mapper_parsing_exception") + || failure.contains("failed to parse field")) + ); + } + + #[test] + fn given_mapper_parsing_failure_should_map_to_permanent_error() { + let response = json!({ + "errors": true, + "items": [ + { "index": { "status": 400, "error": { "type": "mapper_parsing_exception", "reason": "bad field" } } } + ] + }); + + let error = parse_bulk_response(&response) + .into_outcome() + .into_error("iggy_probe") + .expect("failed items should produce an error"); + + assert!(matches!(error, Error::PermanentHttpError(_))); + } + + #[test] + fn given_rejected_execution_should_map_to_transient_error() { + let response = json!({ + "errors": true, + "items": [ + { "index": { "status": 429, "error": { "type": "es_rejected_execution_exception", "reason": "queue full" } } } + ] + }); + + let error = parse_bulk_response(&response) + .into_outcome() + .into_error("iggy_probe") + .expect("failed items should produce an error"); + + assert!(matches!(error, Error::HttpRequestFailed(_))); + } + + #[test] + fn given_any_transient_item_failure_should_prefer_retryable_error() { + let response = json!({ + "errors": true, + "items": [ + { "index": { "status": 400, "error": { "type": "mapper_parsing_exception", "reason": "bad field" } } }, + { "index": { "status": 503, "error": { "type": "unavailable_shards_exception", "reason": "primary shard unavailable" } } } + ] + }); + + let outcome = parse_bulk_response(&response).into_outcome(); + + assert_eq!(outcome.failed, 2); + assert!(outcome.transient); + assert!(matches!( + outcome.into_error("iggy_probe"), + Some(Error::HttpRequestFailed(_)) + )); + } + + #[test] + fn given_merged_outcomes_should_accumulate_totals_and_keep_first_failure() { + let mut outcome = BulkOutcome { + indexed: 2, + failed: 1, + transient: false, + first_failure: Some("first".to_string()), + }; + + outcome.merge(BulkOutcome { + indexed: 3, + failed: 4, + transient: true, + first_failure: Some("second".to_string()), + }); + + assert_eq!( + outcome, + BulkOutcome { + indexed: 5, + failed: 5, + transient: true, + first_failure: Some("first".to_string()), + } + ); + } + + #[test] + fn given_merge_into_clean_outcome_should_adopt_the_incoming_failure() { + let mut outcome = BulkOutcome::default(); + + outcome.merge(BulkOutcome { + indexed: 1, + failed: 1, + transient: false, + first_failure: Some("only".to_string()), + }); + outcome.merge(BulkOutcome::default()); + + assert_eq!(outcome.first_failure.as_deref(), Some("only")); + assert_eq!(outcome.indexed, 1); + assert_eq!(outcome.failed, 1); + } + + #[test] + fn given_bulk_response_without_items_should_report_nothing_indexed() { + assert_eq!(parse_bulk_response(&json!({})), BulkAttempt::default()); + } + + #[test] + fn given_transient_item_failures_should_report_their_positions_for_retry() { + let response = json!({ + "errors": true, + "items": [ + { "index": { "_id": "a", "_version": 1, "result": "created", "status": 201 } }, + { "index": { "_id": "b", "status": 429, "error": { "type": "es_rejected_execution_exception", "reason": "queue full" } } }, + { "index": { "_id": "c", "status": 400, "error": { "type": "mapper_parsing_exception", "reason": "bad field" } } }, + { "index": { "_id": "d", "status": 503, "error": { "type": "unavailable_shards_exception", "reason": "primary shard unavailable" } } } + ] + }); + + let attempt = parse_bulk_response(&response); + + assert_eq!(attempt.indexed, 1); + assert_eq!(attempt.permanent_failed, 1); + assert_eq!(attempt.retryable, vec![1, 3]); + } + + #[test] + fn given_only_permanent_item_failures_should_have_nothing_to_retry() { + let response = json!({ + "errors": true, + "items": [ + { "index": { "status": 400, "error": { "type": "mapper_parsing_exception", "reason": "bad field" } } } + ] + }); + + assert!(parse_bulk_response(&response).retryable.is_empty()); + } + + fn prepared(id: &str) -> PreparedDocument { + PreparedDocument { + id: id.to_string(), + document: json!({ "id": id }), + } + } + + #[test] + fn given_retryable_positions_should_select_matching_documents() { + let documents = [prepared("a"), prepared("b"), prepared("c")]; + let pending: Vec<&PreparedDocument> = documents.iter().collect(); + + let selected = documents_at(&pending, &[0, 2]); + + assert_eq!( + selected + .iter() + .map(|document| document.id.as_str()) + .collect::>(), + vec!["a", "c"] + ); + } + + // A response echoing more items than were sent must not panic: a panic + // crossing the plugin's FFI boundary aborts every connector in the runtime. + #[test] + fn given_out_of_range_positions_should_drop_them_instead_of_panicking() { + let documents = [prepared("a")]; + let pending: Vec<&PreparedDocument> = documents.iter().collect(); + + let selected = documents_at(&pending, &[0, 7]); + + assert_eq!(selected.len(), 1); + assert_eq!(selected[0].id, "a"); + assert!(documents_at(&pending, &[3, 4]).is_empty()); + } + + fn fast_retry_config(max_open_retries: u32) -> OpenSearchSinkConfig { + let mut config = base_config(); + config.max_open_retries = Some(max_open_retries); + config.retry_delay = Some("1ms".to_string()); + config.max_retry_delay = Some("2ms".to_string()); + config + } + + #[tokio::test] + async fn given_transient_open_failure_should_retry_until_success() { + let sink = sink_with_config(fast_retry_config(5)); + let attempts = AtomicU32::new(0); + + let result = sink + .retry_on_open("index existence check", || async { + if attempts.fetch_add(1, Ordering::Relaxed) < 2 { + return Err(Error::HttpRequestFailed("shard not ready".to_string())); + } + Ok(true) + }) + .await; + + assert!(result.expect("third attempt should succeed")); + assert_eq!(attempts.load(Ordering::Relaxed), 3); + } + + #[tokio::test] + async fn given_permanent_open_failure_should_not_retry() { + let sink = sink_with_config(fast_retry_config(5)); + let attempts = AtomicU32::new(0); + + let error = sink + .retry_on_open("index creation", || async { + attempts.fetch_add(1, Ordering::Relaxed); + Err::<(), Error>(Error::PermanentHttpError("invalid mapping".to_string())) + }) + .await + .expect_err("a permanent error should surface immediately"); + + assert!(matches!(error, Error::PermanentHttpError(_))); + assert_eq!(attempts.load(Ordering::Relaxed), 1); + } + + #[tokio::test] + async fn given_exhausted_open_retries_should_report_init_error() { + let sink = sink_with_config(fast_retry_config(2)); + let attempts = AtomicU32::new(0); + + let error = sink + .retry_on_open("health check", || async { + attempts.fetch_add(1, Ordering::Relaxed); + Err::<(), Error>(Error::HttpRequestFailed("connection refused".to_string())) + }) + .await + .expect_err("exhausted retries should fail"); + + assert!(matches!(error, Error::InitError(_))); + // The initial attempt plus max_open_retries retries. + assert_eq!(attempts.load(Ordering::Relaxed), 3); + } + + // The invocation counter is bumped before the client is looked up, so a + // batch that never reaches OpenSearch still counts as an invocation. + #[tokio::test] + async fn given_unopened_sink_when_consuming_should_count_invocation_and_fail() { + let sink = sink_with_config(base_config()); + + let error = sink + .consume(&topic_metadata(), messages_metadata(), Vec::new()) + .await + .expect_err("consuming without an open client should fail"); + + assert!(matches!(error, Error::Connection(_))); + assert_eq!(sink.invocations_count.load(Ordering::Relaxed), 1); + assert_eq!(sink.documents_indexed.load(Ordering::Relaxed), 0); + assert_eq!(sink.errors_count.load(Ordering::Relaxed), 0); + } + + async fn mock_health_sink(server: &MockServer, base_path: &str) -> OpenSearchSink { + let mut config = fast_retry_config(1); + config.url = format!("{}{base_path}", server.uri()); + sink_with_config(config) + } + + fn mock_client(sink: &OpenSearchSink) -> OpenSearch { + let normalized = normalize_url(&sink.config.url).expect("normalize"); + sink.create_client(&normalized).expect("build client") + } + + async fn mock_bulk_sink(server: &MockServer, max_retries: u32) -> OpenSearchSink { + let mut config = fast_retry_config(1); + config.url = server.uri(); + config.max_retries = Some(max_retries); + sink_with_config(config) + } + + // An index-scoped ingest user is not normally granted the cluster-scoped + // cluster:monitor/health privilege, and a 403 still proves the cluster + // answered and the credentials authenticated. + #[tokio::test] + async fn given_forbidden_health_check_should_treat_cluster_as_reachable() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/_cluster/health")) + .respond_with(ResponseTemplate::new(403).set_body_json(json!({ + "error": { + "type": "security_exception", + "reason": "no permissions for [cluster:monitor/health]" + }, + "status": 403 + }))) + .expect(1) + .mount(&server) + .await; + let sink = mock_health_sink(&server, "").await; + let client = mock_client(&sink); + + sink.check_connectivity(&client) + .await + .expect("a denied health check should not fail open()"); + } + + #[tokio::test] + async fn given_unauthorized_health_check_should_fail_open() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/_cluster/health")) + .respond_with(ResponseTemplate::new(401)) + .mount(&server) + .await; + let sink = mock_health_sink(&server, "").await; + let client = mock_client(&sink); + + let error = sink + .check_connectivity(&client) + .await + .expect_err("rejected credentials should still fail open()"); + + assert!(matches!(error, Error::PermanentHttpError(_))); + } + + // Proves the preserved base path reaches the wire, not just normalize_url: + // an unmatched request would land on the mock's 404 and fail this. + #[tokio::test] + async fn given_base_path_url_should_send_requests_under_that_path() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/opensearch/_cluster/health")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "status": "green" }))) + .expect(1) + .mount(&server) + .await; + let sink = mock_health_sink(&server, "/opensearch").await; + let client = mock_client(&sink); + + sink.check_connectivity(&client) + .await + .expect("health check should resolve under the base path"); + } + + #[tokio::test] + async fn given_transient_item_failure_should_retry_only_rejected_documents() { + let server = MockServer::start().await; + let sink = mock_bulk_sink(&server, 2).await; + let client = mock_client(&sink); + let documents = [prepared("a"), prepared("b")]; + let full_body = build_bulk_body(&sink.config.index, &[&documents[0], &documents[1]]) + .expect("build initial body"); + let retry_body = + build_bulk_body(&sink.config.index, &[&documents[1]]).expect("build retry body"); + + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(full_body)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": true, + "items": [ + { "index": { "_id": "a", "status": 201, "result": "created" } }, + { "index": { "_id": "b", "status": 429, "error": { "type": "es_rejected_execution_exception", "reason": "queue full" } } } + ] + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(retry_body)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": false, + "items": [ + { "index": { "_id": "b", "status": 200, "result": "updated" } } + ] + }))) + .expect(1) + .mount(&server) + .await; + + let outcome = sink + .index_chunk(&client, &documents) + .await + .expect("bulk retry should eventually succeed"); + + assert_eq!(outcome.indexed, 2); + assert_eq!(outcome.failed, 0); + } + + #[tokio::test] + async fn given_transient_bulk_status_should_resend_full_chunk() { + let server = MockServer::start().await; + let sink = mock_bulk_sink(&server, 2).await; + let client = mock_client(&sink); + let documents = [prepared("a"), prepared("b")]; + let full_body = build_bulk_body(&sink.config.index, &[&documents[0], &documents[1]]) + .expect("build body"); + + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(full_body.clone())) + .respond_with(ResponseTemplate::new(503)) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(full_body)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": false, + "items": [ + { "index": { "_id": "a", "status": 201, "result": "created" } }, + { "index": { "_id": "b", "status": 201, "result": "created" } } + ] + }))) + .expect(1) + .mount(&server) + .await; + + let outcome = sink + .index_chunk(&client, &documents) + .await + .expect("resending the identical body should eventually succeed"); + + assert_eq!(outcome.indexed, 2); + assert_eq!(outcome.failed, 0); + } + + #[tokio::test] + async fn given_bulk_retries_exhausted_should_return_ok_with_partial_outcome() { + let server = MockServer::start().await; + let sink = mock_bulk_sink(&server, 1).await; + let client = mock_client(&sink); + let documents = [prepared("a")]; + + Mock::given(method("POST")) + .and(path("/_bulk")) + .respond_with(ResponseTemplate::new(503)) + .expect(2) // the initial attempt plus one retry (max_retries = 1) + .mount(&server) + .await; + + let outcome = sink + .index_chunk(&client, &documents) + .await + .expect("exhausted retries should report Ok with a transient failure, not Err"); + + assert_eq!(outcome.indexed, 0); + assert_eq!(outcome.failed, 1); + assert!(outcome.transient); + } + + #[tokio::test] + async fn given_permanent_bulk_status_should_fail_without_retry() { + let server = MockServer::start().await; + let sink = mock_bulk_sink(&server, 3).await; + let client = mock_client(&sink); + let documents = [prepared("a")]; + + Mock::given(method("POST")) + .and(path("/_bulk")) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": { "type": "illegal_argument_exception", "reason": "malformed request" } + }))) + .expect(1) + .mount(&server) + .await; + + let outcome = sink + .index_chunk(&client, &documents) + .await + .expect("a permanent bulk status is reported through the outcome, not Err"); + + assert_eq!(outcome.indexed, 0); + assert_eq!(outcome.failed, 1); + assert!(!outcome.transient); + let error = outcome + .into_error(&sink.config.index) + .expect("a failed outcome must produce an error"); + assert!(matches!(error, Error::PermanentHttpError(_))); + } + + // Regression test: attempt 1 partially succeeds (document `a` indexed, + // `b` rejected transiently) and only the retry of `b` hits a permanent + // whole-request status. `a`'s success must survive into the final + // outcome instead of being discarded by the early `Err` return. + #[tokio::test] + async fn given_permanent_bulk_status_after_partial_retry_should_keep_earlier_indexed_count() { + let server = MockServer::start().await; + let sink = mock_bulk_sink(&server, 2).await; + let client = mock_client(&sink); + let documents = [prepared("a"), prepared("b")]; + let full_body = build_bulk_body(&sink.config.index, &[&documents[0], &documents[1]]) + .expect("build initial body"); + let retry_body = + build_bulk_body(&sink.config.index, &[&documents[1]]).expect("build retry body"); + + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(full_body)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": true, + "items": [ + { "index": { "_id": "a", "status": 201, "result": "created" } }, + { "index": { "_id": "b", "status": 429, "error": { "type": "es_rejected_execution_exception", "reason": "queue full" } } } + ] + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(retry_body)) + .respond_with(ResponseTemplate::new(401)) + .expect(1) + .mount(&server) + .await; + + let outcome = sink + .index_chunk(&client, &documents) + .await + .expect("a permanent failure on retry is reported through the outcome, not Err"); + + assert_eq!( + outcome.indexed, 1, + "document `a` from the first attempt must still count as indexed" + ); + assert_eq!(outcome.failed, 1); + assert!(!outcome.transient); + } + + #[tokio::test] + async fn given_unparsable_bulk_response_body_should_be_retried_as_transient() { + let server = MockServer::start().await; + let sink = mock_bulk_sink(&server, 1).await; + let client = mock_client(&sink); + let documents = [prepared("a")]; + + Mock::given(method("POST")) + .and(path("/_bulk")) + .respond_with(ResponseTemplate::new(200).set_body_raw("not json", "application/json")) + .up_to_n_times(1) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/_bulk")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": false, + "items": [ + { "index": { "_id": "a", "status": 201, "result": "created" } } + ] + }))) + .expect(1) + .mount(&server) + .await; + + let outcome = sink + .index_chunk(&client, &documents) + .await + .expect("an unparsable 200 body should be retried, not hard-failed"); + + assert_eq!(outcome.indexed, 1); + assert_eq!(outcome.failed, 0); + } + + #[tokio::test] + async fn given_batch_larger_than_batch_size_should_split_into_multiple_bulk_calls() { + let server = MockServer::start().await; + let mut config = fast_retry_config(1); + config.url = server.uri(); + config.batch_size = Some(2); + let sink = sink_with_config(config); + let client = mock_client(&sink); + let documents: Vec = ["a", "b", "c", "d", "e"] + .into_iter() + .map(prepared) + .collect(); + + Mock::given(method("POST")) + .and(path("/_bulk")) + .respond_with(|request: &wiremock::Request| { + let sent_documents = std::str::from_utf8(&request.body) + .expect("bulk body should be valid utf8") + .lines() + .filter(|line| !line.is_empty()) + .count() + / 2; + let items = vec![ + json!({ "index": { "status": 201, "result": "created" } }); + sent_documents + ]; + ResponseTemplate::new(200).set_body_json(json!({ "errors": false, "items": items })) + }) + .expect(3) // chunks of [2, 2, 1] for 5 documents at batch_size = 2 + .mount(&server) + .await; + + let indexed = sink + .index_documents(&client, documents) + .await + .expect("every chunk should be attempted and counted"); + + assert_eq!(indexed, 5); + } + + // Regression test for the specific claim the real-infra + // `given_missing_index_and_mapping_conflict_should_isolate_failures_from_healthy_sibling` + // integration test documents but cannot itself verify: a chunk failing + // must not stop `index_documents` from attempting the chunks queued + // behind it. Real OpenSearch's own per-item bulk semantics mean the + // final document count there is identical whether chunking happened or + // the whole batch went out in one `_bulk` call, so only a mocked + // per-chunk response (asserted here via wiremock's per-mock `.expect(1)`, + // which panics on drop if a mock is never hit) can prove every chunk was + // actually sent. + #[tokio::test] + async fn given_permanently_failing_chunk_should_not_abandon_later_chunks() { + let server = MockServer::start().await; + let mut config = fast_retry_config(1); + config.url = server.uri(); + config.batch_size = Some(2); + let sink = sink_with_config(config); + let client = mock_client(&sink); + let documents: Vec = ["a", "b", "c", "d", "e"] + .into_iter() + .map(prepared) + .collect(); + let chunk1_body = build_bulk_body(&sink.config.index, &[&documents[0], &documents[1]]) + .expect("build chunk 1 body"); + let chunk2_body = build_bulk_body(&sink.config.index, &[&documents[2], &documents[3]]) + .expect("build chunk 2 body"); + let chunk3_body = + build_bulk_body(&sink.config.index, &[&documents[4]]).expect("build chunk 3 body"); + + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(chunk1_body)) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": { "type": "illegal_argument_exception", "reason": "malformed request" } + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(chunk2_body)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": false, + "items": [ + { "index": { "_id": "c", "status": 201, "result": "created" } }, + { "index": { "_id": "d", "status": 201, "result": "created" } } + ] + }))) + .expect(1) + .mount(&server) + .await; + Mock::given(method("POST")) + .and(path("/_bulk")) + .and(body_bytes(chunk3_body)) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "errors": false, + "items": [ + { "index": { "_id": "e", "status": 201, "result": "created" } } + ] + }))) + .expect(1) + .mount(&server) + .await; + + let error = sink + .index_documents(&client, documents) + .await + .expect_err("a permanently failing first chunk should still surface an error"); + + assert_eq!( + error.indexed, 3, + "documents from the two chunks after the failing one must still be indexed" + ); + assert_eq!(error.failed, 2); + } + + #[tokio::test] + async fn given_concurrent_index_creation_should_not_fail_open() { + let server = MockServer::start().await; + let sink = mock_health_sink(&server, "").await; + let client = mock_client(&sink); + + Mock::given(method("PUT")) + .and(path(format!("/{}", sink.config.index))) + .respond_with(ResponseTemplate::new(400).set_body_json(json!({ + "error": { + "type": "resource_already_exists_exception", + "reason": "index [iggy_messages/abc] already exists" + }, + "status": 400 + }))) + .expect(1) + .mount(&server) + .await; + + sink.create_index(&client) + .await + .expect("another instance winning the create race should not fail open()"); + } + + #[test] + fn given_error_variants_should_classify_open_time_retryability() { + assert!(is_transient_error(&Error::HttpRequestFailed( + "503".to_string() + ))); + assert!(!is_transient_error(&Error::PermanentHttpError( + "400".to_string() + ))); + assert!(!is_transient_error(&Error::InitError( + "missing".to_string() + ))); + } + + #[test] + fn given_credentials_in_url_should_redact_them_from_logs() { + assert_eq!( + sanitize_url_for_log("https://admin:hunter2@opensearch.example.com:9200/path"), + "https://opensearch.example.com:9200/path" + ); + } + + #[test] + fn given_url_without_scheme_should_default_to_http() { + assert_eq!( + normalize_url("localhost:9200").expect("normalize"), + "http://localhost:9200" + ); + } + + #[test] + fn given_url_with_query_and_fragment_should_strip_them_and_keep_the_base_path() { + assert_eq!( + normalize_url("https://localhost:9200/opensearch?foo=bar#section").expect("normalize"), + "https://localhost:9200/opensearch" + ); + } + + // A reverse proxy exposing OpenSearch under a subpath is a supported + // topology: the transport joins every request path onto this base. + #[test] + fn given_url_with_base_path_should_preserve_it() { + assert_eq!( + normalize_url("https://proxy.example.com/opensearch/").expect("normalize"), + "https://proxy.example.com/opensearch" + ); + } + + #[test] + fn given_root_url_should_normalize_to_bare_origin() { + assert_eq!( + normalize_url("https://localhost:9200/").expect("normalize"), + "https://localhost:9200" + ); + } + + #[test] + fn given_blank_url_should_fail_normalization() { + assert!(matches!(normalize_url(" "), Err(Error::Connection(_)))); + } + + #[test] + fn given_malformed_url_should_fail_normalization() { + let error = normalize_url("http://[::1") + .expect_err("an unterminated IPv6 literal should not parse"); + + assert!(matches!(error, Error::Connection(_))); + } + + #[test] + fn given_credentials_embedded_in_url_should_fail_normalization() { + let error = normalize_url("https://admin:hunter2@opensearch.example.com:9200") + .expect_err("embedded credentials should be rejected"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_username_only_embedded_in_url_should_fail_normalization() { + let error = normalize_url("https://admin@opensearch.example.com:9200") + .expect_err("embedded username without a password should still be rejected"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_loopback_hosts_should_be_detected() { + assert!(is_loopback_host("localhost")); + assert!(is_loopback_host("127.0.0.1")); + assert!(is_loopback_host("::1")); + assert!(!is_loopback_host("opensearch.prod")); + } + + #[test] + fn given_refresh_values_should_match_opensearch_wire_format() { + assert_eq!(serde_json::to_string(&Refresh::True).unwrap(), "\"true\""); + assert_eq!(serde_json::to_string(&Refresh::False).unwrap(), "\"false\""); + assert_eq!( + serde_json::to_string(&Refresh::WaitFor).unwrap(), + "\"wait_for\"" + ); + } + + #[test] + fn given_debug_formatted_config_should_redact_password() { + let mut config = base_config(); + config.username = Some("admin".to_string()); + config.password = Some(SecretString::from("hunter2")); + + let rendered = format!("{config:?}"); + + assert!(!rendered.contains("hunter2"), "{rendered}"); + assert!(rendered.contains("admin"), "{rendered}"); + } + + #[test] + fn given_debug_formatted_sink_should_redact_password() { + let mut config = base_config(); + config.username = Some("admin".to_string()); + config.password = Some(SecretString::from("hunter2")); + + let rendered = format!("{:?}", sink_with_config(config)); + + assert!(!rendered.contains("hunter2"), "{rendered}"); + } + + #[test] + fn given_toml_config_should_deserialize_password_into_secret() { + let config: OpenSearchSinkConfig = toml::from_str( + r#" + url = "http://localhost:9200" + index = "iggy_messages" + username = "admin" + password = "hunter2" + refresh = "wait_for" + "#, + ) + .expect("config should deserialize"); + + assert_eq!( + config + .password + .as_ref() + .map(|password| password.expose_secret().to_string()), + Some("hunter2".to_string()) + ); + assert_eq!(config.refresh, Some(Refresh::WaitFor)); + } +} diff --git a/core/integration/tests/connectors/fixtures/mod.rs b/core/integration/tests/connectors/fixtures/mod.rs index e4992d6785..1467500740 100644 --- a/core/integration/tests/connectors/fixtures/mod.rs +++ b/core/integration/tests/connectors/fixtures/mod.rs @@ -26,6 +26,7 @@ mod iceberg; mod influxdb; mod meilisearch; mod mongodb; +mod opensearch; mod postgres; mod quickwit; mod s3; @@ -74,6 +75,7 @@ pub use mongodb::{ MongoDbOps, MongoDbSinkAutoCreateFixture, MongoDbSinkBatchFixture, MongoDbSinkFailpointFixture, MongoDbSinkFixture, MongoDbSinkJsonFixture, MongoDbSinkWriteConcernFixture, }; +pub use opensearch::{OpenSearchFailureFixture, OpenSearchOps, OpenSearchSinkFixture}; pub use postgres::{ PostgresOps, PostgresSinkByteaFixture, PostgresSinkFixture, PostgresSinkJsonFixture, PostgresSourceByteaFixture, PostgresSourceCdcFixture, PostgresSourceDeleteFixture, diff --git a/core/integration/tests/connectors/fixtures/opensearch/container.rs b/core/integration/tests/connectors/fixtures/opensearch/container.rs new file mode 100644 index 0000000000..56e9398ad8 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/opensearch/container.rs @@ -0,0 +1,201 @@ +// 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 = "opensearchproject/opensearch"; +// Pinned rather than a floating major tag: verified against this exact tag +// that DISABLE_SECURITY_PLUGIN=true still boots green over plain HTTP with no +// OPENSEARCH_INITIAL_ADMIN_PASSWORD (required since 2.12 unless the security +// plugin is genuinely disabled). +const OPENSEARCH_TAG: &str = "3.8.0"; +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, mirroring the Elasticsearch fixture: the first test +// creates it, later test processes attach by name. Per-test isolation comes +// from a unique index per fixture, not a fresh container. +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 DEFAULT_TEST_TOPIC_2: &str = "test_topic_2"; + +#[derive(Debug, Deserialize)] +pub struct OpenSearchCountResponse { + pub count: u64, +} + +pub struct OpenSearchContainer { + // Held so testcontainers' Drop runs on test exit; ReuseDirective::Always + // makes that Drop leave the container running for the next test to attach. + #[allow(dead_code)] + container: ContainerAsync, + pub base_url: String, +} + +impl OpenSearchContainer { + pub async fn start() -> Result { + 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("DISABLE_SECURITY_PLUGIN", "true") + .with_env_var("DISABLE_INSTALL_DEMO_CONFIG", "true") + .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 + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "OpenSearchContainer".to_string(), + message: format!("Failed to start container: {e}"), + })?; + + info!("Started OpenSearch container"); + + let mapped_port = container + .ports() + .await + .map_err(|e| TestBinaryError::FixtureSetup { + fixture_type: "OpenSearchContainer".to_string(), + message: format!("Failed to get ports: {e}"), + })? + .map_to_host_port_ipv4(OPENSEARCH_PORT) + .ok_or_else(|| TestBinaryError::FixtureSetup { + fixture_type: "OpenSearchContainer".to_string(), + message: "No mapping for OpenSearch port".to_string(), + })?; + + let base_url = format!("http://localhost:{mapped_port}"); + info!("OpenSearch container available at {base_url}"); + + Ok(Self { + container, + base_url, + }) + } +} + +pub fn create_http_client() -> HttpClient { + let retry_policy = ExponentialBackoff::builder().build_with_max_retries(3); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .expect("Failed to build HTTP client"); + reqwest_middleware::ClientBuilder::new(client) + .with(RetryTransientMiddleware::new_with_policy(retry_policy)) + .build() +} + +pub trait OpenSearchOps: Sync { + fn container(&self) -> &OpenSearchContainer; + fn http_client(&self) -> &HttpClient; + + fn count_documents( + &self, + index_name: &str, + ) -> impl std::future::Future> + Send { + async move { + let url = format!("{}/{}/_count", self.container().base_url, index_name); + + let response = self.http_client().get(&url).send().await.map_err(|e| { + TestBinaryError::InvalidState { + message: format!("Failed to count OpenSearch documents: {e}"), + } + })?; + + if response.status() == reqwest::StatusCode::NOT_FOUND { + // Index not created yet: treat as zero documents rather than + // an error, since polling starts before the sink has opened. + return Ok(0); + } + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(TestBinaryError::InvalidState { + message: format!( + "Failed to count OpenSearch documents: status={status}, body={body}" + ), + }); + } + + response + .json::() + .await + .map(|response| response.count) + .map_err(|e| TestBinaryError::InvalidState { + message: format!("Failed to parse OpenSearch count response: {e}"), + }) + } + } + + fn search_all( + &self, + index_name: &str, + ) -> impl std::future::Future> + Send { + async move { + let url = format!("{}/{}/_search", self.container().base_url, index_name); + let query = serde_json::json!({ "query": { "match_all": {} }, "size": 1000 }); + + let response = self + .http_client() + .post(&url) + .header("Content-Type", "application/json") + .json(&query) + .send() + .await + .map_err(|e| TestBinaryError::InvalidState { + message: format!("Failed to search OpenSearch index: {e}"), + })?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(TestBinaryError::InvalidState { + message: format!( + "Failed to search OpenSearch index: status={status}, body={body}" + ), + }); + } + + response + .json::() + .await + .map_err(|e| TestBinaryError::InvalidState { + message: format!("Failed to parse OpenSearch search response: {e}"), + }) + } + } +} diff --git a/core/integration/tests/connectors/fixtures/opensearch/failure.rs b/core/integration/tests/connectors/fixtures/opensearch/failure.rs new file mode 100644 index 0000000000..7de0f2458e --- /dev/null +++ b/core/integration/tests/connectors/fixtures/opensearch/failure.rs @@ -0,0 +1,129 @@ +// 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::{OpenSearchContainer, OpenSearchOps, create_http_client}; +use async_trait::async_trait; +use integration::harness::{TestBinaryError, TestFixture}; +use reqwest_middleware::ClientWithMiddleware as HttpClient; +use std::collections::HashMap; +use uuid::Uuid; + +const PLUGIN_PATH: &str = "../../target/debug/libiggy_connector_opensearch_sink"; + +const ENV_MISSING_INDEX_URL: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_MISSING_INDEX_PLUGIN_CONFIG_URL"; +const ENV_MISSING_INDEX_INDEX: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_MISSING_INDEX_PLUGIN_CONFIG_INDEX"; +const ENV_MISSING_INDEX_PATH: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_MISSING_INDEX_PATH"; + +const ENV_MAPPING_CONFLICT_URL: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_MAPPING_CONFLICT_PLUGIN_CONFIG_URL"; +const ENV_MAPPING_CONFLICT_INDEX: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_MAPPING_CONFLICT_PLUGIN_CONFIG_INDEX"; +const ENV_MAPPING_CONFLICT_PATH: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_MAPPING_CONFLICT_PATH"; + +const ENV_HEALTHY_URL: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_HEALTHY_PLUGIN_CONFIG_URL"; +const ENV_HEALTHY_INDEX: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_HEALTHY_PLUGIN_CONFIG_INDEX"; +const ENV_HEALTHY_PATH: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_HEALTHY_PATH"; + +/// Three OpenSearch sink connectors sharing one container, wired to prove the +/// failure-state assertions in `opensearch_sink_failures.rs`: +/// * `opensearch_missing_index` targets an index that will never exist, with +/// `create_index_if_not_exists = false` (static in its own TOML), so it +/// fails during `open()` and never starts consuming. +/// * `opensearch_mapping_conflict` has an explicit integer mapping for +/// `count` (static in its own TOML, so the type conflict is deterministic +/// rather than dependent on dynamic-mapping inference order) and +/// subscribes to `TOPIC_2`, receiving a batch containing a real +/// `mapper_parsing_exception`-triggering document partway through the +/// test. +/// * `opensearch_healthy` subscribes to `TOPIC` (a distinct topic in the +/// same stream, so it never sees the conflicting message) and stays +/// `Running` throughout, proving the runtime isolates one failing +/// connector from its siblings. +pub struct OpenSearchFailureFixture { + container: OpenSearchContainer, + http_client: HttpClient, + missing_index: String, + mapping_conflict_index: String, + healthy_index: String, +} + +impl OpenSearchOps for OpenSearchFailureFixture { + fn container(&self) -> &OpenSearchContainer { + &self.container + } + + fn http_client(&self) -> &HttpClient { + &self.http_client + } +} + +impl OpenSearchFailureFixture { + pub fn mapping_conflict_index(&self) -> &str { + &self.mapping_conflict_index + } + + pub fn healthy_index(&self) -> &str { + &self.healthy_index + } +} + +#[async_trait] +impl TestFixture for OpenSearchFailureFixture { + async fn setup() -> Result { + let container = OpenSearchContainer::start().await?; + let http_client = create_http_client(); + + Ok(Self { + container, + http_client, + missing_index: format!("iggy_missing_{}", Uuid::new_v4().simple()), + mapping_conflict_index: format!("iggy_conflict_{}", Uuid::new_v4().simple()), + healthy_index: format!("iggy_healthy_{}", Uuid::new_v4().simple()), + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + HashMap::from([ + ( + ENV_MISSING_INDEX_URL.to_string(), + self.container.base_url.clone(), + ), + ( + ENV_MISSING_INDEX_INDEX.to_string(), + self.missing_index.clone(), + ), + (ENV_MISSING_INDEX_PATH.to_string(), PLUGIN_PATH.to_string()), + ( + ENV_MAPPING_CONFLICT_URL.to_string(), + self.container.base_url.clone(), + ), + ( + ENV_MAPPING_CONFLICT_INDEX.to_string(), + self.mapping_conflict_index.clone(), + ), + ( + ENV_MAPPING_CONFLICT_PATH.to_string(), + PLUGIN_PATH.to_string(), + ), + (ENV_HEALTHY_URL.to_string(), self.container.base_url.clone()), + (ENV_HEALTHY_INDEX.to_string(), self.healthy_index.clone()), + (ENV_HEALTHY_PATH.to_string(), PLUGIN_PATH.to_string()), + ]) + } +} diff --git a/core/integration/tests/connectors/fixtures/opensearch/mod.rs b/core/integration/tests/connectors/fixtures/opensearch/mod.rs new file mode 100644 index 0000000000..22f38a3b2b --- /dev/null +++ b/core/integration/tests/connectors/fixtures/opensearch/mod.rs @@ -0,0 +1,24 @@ +// 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. + +mod container; +mod failure; +mod sink; + +pub use container::OpenSearchOps; +pub use failure::OpenSearchFailureFixture; +pub use sink::OpenSearchSinkFixture; diff --git a/core/integration/tests/connectors/fixtures/opensearch/sink.rs b/core/integration/tests/connectors/fixtures/opensearch/sink.rs new file mode 100644 index 0000000000..75b0e62111 --- /dev/null +++ b/core/integration/tests/connectors/fixtures/opensearch/sink.rs @@ -0,0 +1,161 @@ +// 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, DEFAULT_TEST_TOPIC_2, OpenSearchContainer, + OpenSearchOps, create_http_client, +}; +use async_trait::async_trait; +use integration::harness::{TestBinaryError, TestFixture}; +use reqwest_middleware::ClientWithMiddleware as HttpClient; +use std::collections::HashMap; +use std::time::Duration; +use tokio::time::sleep; +use tracing::info; +use uuid::Uuid; + +const ENV_SINK_URL: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_PLUGIN_CONFIG_URL"; +const ENV_SINK_INDEX: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_PLUGIN_CONFIG_INDEX"; +const ENV_SINK_DOCUMENT_ID_FIELD: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_PLUGIN_CONFIG_DOCUMENT_ID_FIELD"; +// Without this, a write is not visible to `_search`/`_count` until the next +// index refresh (default ~1s), which makes read-after-write assertions +// flaky. `wait_for` blocks the bulk response until the write is visible, +// exercising the `refresh` config option end-to-end. +const ENV_SINK_REFRESH: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_PLUGIN_CONFIG_REFRESH"; +const ENV_SINK_STREAMS_0_STREAM: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_0_STREAM"; +const ENV_SINK_STREAMS_0_TOPICS: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_0_TOPICS"; +const ENV_SINK_STREAMS_0_SCHEMA: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_0_SCHEMA"; +const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_0_CONSUMER_GROUP"; +// A second stream entry, subscribed to a distinct topic under the `raw` +// schema, so the same live-server test can prove `document_from_raw`'s two +// branches (valid-JSON-as-raw-bytes, and the base64 fallback for non-JSON +// bytes) alongside the `json`-schema coverage on stream 0. +const ENV_SINK_STREAMS_1_STREAM: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_1_STREAM"; +const ENV_SINK_STREAMS_1_TOPICS: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_1_TOPICS"; +const ENV_SINK_STREAMS_1_SCHEMA: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_1_SCHEMA"; +const ENV_SINK_STREAMS_1_CONSUMER_GROUP: &str = + "IGGY_CONNECTORS_SINK_OPENSEARCH_STREAMS_1_CONSUMER_GROUP"; +const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_OPENSEARCH_PATH"; + +const SINK_INDEX_PREFIX: &str = "iggy_messages"; +const POLL_ATTEMPTS: usize = 100; +const POLL_INTERVAL_MS: u64 = 50; + +pub struct OpenSearchSinkFixture { + container: OpenSearchContainer, + http_client: HttpClient, + // Unique per fixture so tests sharing the reused container never collide + // on the same index. The connector writes here via ENV_SINK_INDEX. + index: String, +} + +impl OpenSearchOps for OpenSearchSinkFixture { + fn container(&self) -> &OpenSearchContainer { + &self.container + } + + fn http_client(&self) -> &HttpClient { + &self.http_client + } +} + +impl OpenSearchSinkFixture { + pub fn index(&self) -> &str { + &self.index + } + + pub async fn wait_for_document_count( + &self, + expected_count: u64, + ) -> Result { + for _ in 0..POLL_ATTEMPTS { + if let Ok(count) = self.count_documents(&self.index).await + && count >= expected_count + { + info!("Found {count} documents in OpenSearch (expected {expected_count})"); + return Ok(count); + } + sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; + } + + let final_count = self.count_documents(&self.index).await.unwrap_or(0); + Err(TestBinaryError::InvalidState { + message: format!( + "Expected at least {expected_count} documents, found {final_count} after {POLL_ATTEMPTS} attempts" + ), + }) + } +} + +#[async_trait] +impl TestFixture for OpenSearchSinkFixture { + async fn setup() -> Result { + let container = OpenSearchContainer::start().await?; + let http_client = create_http_client(); + let index = format!("{SINK_INDEX_PREFIX}_{}", Uuid::new_v4().simple()); + + Ok(Self { + container, + http_client, + index, + }) + } + + fn connectors_runtime_envs(&self) -> HashMap { + HashMap::from([ + (ENV_SINK_URL.to_string(), self.container.base_url.clone()), + (ENV_SINK_INDEX.to_string(), self.index.clone()), + ( + ENV_SINK_DOCUMENT_ID_FIELD.to_string(), + "order_id".to_string(), + ), + (ENV_SINK_REFRESH.to_string(), "wait_for".to_string()), + ( + ENV_SINK_STREAMS_0_STREAM.to_string(), + DEFAULT_TEST_STREAM.to_string(), + ), + ( + ENV_SINK_STREAMS_0_TOPICS.to_string(), + format!("[{DEFAULT_TEST_TOPIC}]"), + ), + (ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string()), + ( + ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(), + "opensearch_sink".to_string(), + ), + ( + ENV_SINK_STREAMS_1_STREAM.to_string(), + DEFAULT_TEST_STREAM.to_string(), + ), + ( + ENV_SINK_STREAMS_1_TOPICS.to_string(), + format!("[{DEFAULT_TEST_TOPIC_2}]"), + ), + (ENV_SINK_STREAMS_1_SCHEMA.to_string(), "raw".to_string()), + ( + ENV_SINK_STREAMS_1_CONSUMER_GROUP.to_string(), + "opensearch_sink_raw".to_string(), + ), + ( + ENV_SINK_PATH.to_string(), + "../../target/debug/libiggy_connector_opensearch_sink".to_string(), + ), + ]) + } +} diff --git a/core/integration/tests/connectors/mod.rs b/core/integration/tests/connectors/mod.rs index 08b794fe8b..90944877e8 100644 --- a/core/integration/tests/connectors/mod.rs +++ b/core/integration/tests/connectors/mod.rs @@ -27,6 +27,7 @@ mod iceberg; mod influxdb; mod meilisearch; mod mongodb; +mod opensearch; mod postgres; mod quickwit; mod random; diff --git a/core/integration/tests/connectors/opensearch/failure_states.toml b/core/integration/tests/connectors/opensearch/failure_states.toml new file mode 100644 index 0000000000..68e0a2c705 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/failure_states.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "tests/connectors/opensearch/failure_states" diff --git a/core/integration/tests/connectors/opensearch/failure_states/opensearch_healthy.toml b/core/integration/tests/connectors/opensearch/failure_states/opensearch_healthy.toml new file mode 100644 index 0000000000..9adb9e6c89 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/failure_states/opensearch_healthy.toml @@ -0,0 +1,43 @@ +# 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. + +# Healthy companion sink: verifies that a sibling connector with valid +# configuration keeps running independently while opensearch_mapping_conflict +# absorbs a consume()-level failure (see that connector's own config comment) +# and opensearch_missing_index fails to open. Subscribes to test_topic, +# distinct from the mapping-conflict sink's test_topic_2. + +type = "sink" +key = "opensearch_healthy" +enabled = true +version = 0 +name = "OpenSearch healthy sink" +path = "../../target/debug/libiggy_connector_opensearch_sink" +verbose = true + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "opensearch_healthy" + +[plugin_config] +url = "http://localhost:9200" +index = "iggy_healthy" +create_index_if_not_exists = true diff --git a/core/integration/tests/connectors/opensearch/failure_states/opensearch_mapping_conflict.toml b/core/integration/tests/connectors/opensearch/failure_states/opensearch_mapping_conflict.toml new file mode 100644 index 0000000000..3ca3283881 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/failure_states/opensearch_mapping_conflict.toml @@ -0,0 +1,54 @@ +# 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. + +# Opens and indexes successfully, then receives a document that violates its +# own explicit mapping (count is pinned to integer at index-creation time, so +# the conflict is deterministic rather than dependent on dynamic-mapping +# inference order). Asserts the real OpenSearch 400 mapper_parsing_exception +# propagates as Error::PermanentHttpError and consume() returns Err, but the +# connectors runtime discards that return value (see opensearch_sink_failures.rs +# module doc), so the connector stays Running and keeps indexing later +# messages; this failure is only visible in its own tracing output. Subscribes +# to test_topic_2, distinct from the healthy sink's test_topic, so it never +# receives the healthy sink's messages or vice versa. + +type = "sink" +key = "opensearch_mapping_conflict" +enabled = true +version = 0 +name = "OpenSearch mapping-conflict sink" +path = "../../target/debug/libiggy_connector_opensearch_sink" +verbose = true + +[[streams]] +stream = "test_stream" +topics = ["test_topic_2"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "opensearch_mapping_conflict" + +[plugin_config] +url = "http://localhost:9200" +index = "iggy_mapping_conflict" +create_index_if_not_exists = true +# Small enough that one consume() batch spans several _bulk calls, so the test +# can prove a failing chunk does not abandon the chunks queued behind it. +batch_size = 2 + +[plugin_config.index_mapping.mappings.properties.count] +type = "integer" diff --git a/core/integration/tests/connectors/opensearch/failure_states/opensearch_missing_index.toml b/core/integration/tests/connectors/opensearch/failure_states/opensearch_missing_index.toml new file mode 100644 index 0000000000..803b1a3964 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/failure_states/opensearch_missing_index.toml @@ -0,0 +1,42 @@ +# 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. + +# Deliberately broken sink: targets an index that will never exist, with +# create_index_if_not_exists = false. Asserts that ensure_index_exists() +# fails open() with InitError against a real server, and that the runtime +# reports ConnectorStatus::Error instead of aborting startup. + +type = "sink" +key = "opensearch_missing_index" +enabled = true +version = 0 +name = "OpenSearch missing-index sink" +path = "../../target/debug/libiggy_connector_opensearch_sink" +verbose = true + +[[streams]] +stream = "test_stream" +topics = ["test_topic"] +schema = "json" +batch_length = 100 +poll_interval = "5ms" +consumer_group = "opensearch_missing_index" + +[plugin_config] +url = "http://localhost:9200" +index = "iggy_this_index_does_not_exist" +create_index_if_not_exists = false diff --git a/core/integration/tests/connectors/opensearch/mod.rs b/core/integration/tests/connectors/opensearch/mod.rs new file mode 100644 index 0000000000..aadf3e8b30 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/mod.rs @@ -0,0 +1,19 @@ +// 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. + +mod opensearch_sink; +mod opensearch_sink_failures; diff --git a/core/integration/tests/connectors/opensearch/opensearch_sink.rs b/core/integration/tests/connectors/opensearch/opensearch_sink.rs new file mode 100644 index 0000000000..732ad362d5 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/opensearch_sink.rs @@ -0,0 +1,325 @@ +// 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 crate::connectors::fixtures::{OpenSearchOps, OpenSearchSinkFixture}; +use base64::{Engine as _, engine::general_purpose}; +use bytes::Bytes; +use iggy::prelude::{IggyMessage, Partitioning}; +use iggy_common::{HeaderKey, HeaderValue, Identifier, MessageClient}; +use integration::harness::seeds; +use integration::iggy_harness; +use std::collections::BTreeMap; + +// Covers the natural-key path: `send_messages` always assigns fresh offsets, +// so generated-ID replay idempotency is unit-tested instead. +#[iggy_harness( + server(connectors_runtime(config_path = "tests/connectors/opensearch/sink.toml")), + seed = seeds::connector_multi_topic_stream +)] +async fn given_json_messages_when_sink_consumes_should_index_documents_and_upsert_by_natural_key( + harness: &TestHarness, + fixture: OpenSearchSinkFixture, +) { + let client = harness.root_client().await.unwrap(); + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + + let send = |payload: serde_json::Value, message_id: u128| { + let stream_id = stream_id.clone(); + let topic_id = topic_id.clone(); + let client = &client; + async move { + let mut messages = vec![ + IggyMessage::builder() + .id(message_id) + .payload(Bytes::from( + serde_json::to_vec(&payload).expect("serialize"), + )) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("send message"); + } + }; + + send( + serde_json::json!({ + "order_id": "A-1", + "name": "first", + "nested": { "deep": { "value": 1 } }, + "tags": ["rust", "iggy"], + }), + 1, + ) + .await; + send( + serde_json::json!({ "order_id": "A-2", "name": "second" }), + 2, + ) + .await; + + let count_after_first = fixture + .wait_for_document_count(2) + .await + .expect("wait for documents after first send"); + assert_eq!(count_after_first, 2); + + let documents = fixture + .search_all(fixture.index()) + .await + .expect("search index"); + let hits = documents["hits"]["hits"] + .as_array() + .expect("hits array") + .clone(); + let first_hit = hits + .iter() + .find(|hit| hit["_source"]["order_id"] == "A-1") + .expect("order A-1 indexed"); + assert_eq!(first_hit["_source"]["name"], "first"); + assert_eq!(first_hit["_source"]["nested"]["deep"]["value"], 1); + assert_eq!( + first_hit["_source"]["tags"], + serde_json::json!(["rust", "iggy"]) + ); + assert_eq!(first_hit["_id"], "A-1"); + assert!( + hits.iter() + .all(|hit| hit["_source"]["iggy_stream"] == seeds::names::STREAM) + ); + + // Resend order A-1 at a new offset: the natural key must upsert even when + // the offset differs. + send( + serde_json::json!({ "order_id": "A-1", "name": "first-updated" }), + 3, + ) + .await; + + // Poll on the actual updated content rather than a fixed sleep. + // `refresh=wait_for` blocks the sink's bulk response until the *next* + // scheduled OpenSearch refresh, not an immediate one, so the connector's + // own indexing latency for this batch can approach a full refresh + // interval (~1s default) even though the config is applied correctly. + // The document count alone cannot detect this: an upsert leaves the + // count at 2 whether or not the new content has landed yet, so only a + // content-based poll can distinguish "still stale" from "genuinely + // failed to upsert." + let updated_first_hit = wait_for_updated_name(&fixture, "first-updated").await; + assert_eq!(updated_first_hit["_source"]["order_id"], "A-1"); + + let count_after_update = fixture + .count_documents(fixture.index()) + .await + .expect("count after natural-key update"); + assert_eq!( + count_after_update, 2, + "resending the same order_id at a new offset must upsert, not duplicate" + ); + + // A message carrying both a string and a raw binary header, proving + // iggy_headers is actually indexed against a live server. The conversion + // itself (headers_to_json) is unit-tested in isolation; only a real + // send_messages round trip proves the wire-decoded headers survive all + // the way into the indexed document. + let string_header_key = HeaderKey::try_from("x-correlation-id").expect("header key"); + let string_header_value = HeaderValue::try_from("abc-123").expect("header value"); + let raw_header_key = HeaderKey::try_from("x-raw").expect("header key"); + let raw_header_value = HeaderValue::try_from([1u8, 2, 3].as_slice()).expect("header value"); + let user_headers = BTreeMap::from([ + (string_header_key, string_header_value), + (raw_header_key, raw_header_value), + ]); + + let mut messages_with_headers = vec![ + IggyMessage::builder() + .id(4) + .payload(Bytes::from( + serde_json::to_vec(&serde_json::json!({ "order_id": "A-3", "name": "third" })) + .expect("serialize"), + )) + .user_headers(user_headers) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages_with_headers, + ) + .await + .expect("send message with headers"); + + let count_after_headers = fixture + .wait_for_document_count(3) + .await + .expect("wait for documents after header send"); + assert_eq!(count_after_headers, 3); + + let documents = fixture + .search_all(fixture.index()) + .await + .expect("search index"); + let header_hit = documents["hits"]["hits"] + .as_array() + .expect("hits array") + .iter() + .find(|hit| hit["_source"]["order_id"] == "A-3") + .expect("order A-3 indexed") + .clone(); + // Both header kinds share the {data, data_encoding} shape. Asserted against + // a live server because a divergent shape per kind is only rejected once + // OpenSearch has pinned the dynamic mapping for iggy_headers.. + let headers = &header_hit["_source"]["iggy_headers"]; + assert_eq!(headers["x-correlation-id"]["data"], "abc-123"); + assert_eq!(headers["x-correlation-id"]["data_encoding"], "utf8"); + assert_eq!( + headers["x-raw"]["data"], + general_purpose::STANDARD.encode([1u8, 2, 3]) + ); + assert_eq!(headers["x-raw"]["data_encoding"], "base64"); + + // Payload::Raw coverage: stream 1 subscribes to TOPIC_2 under the `raw` + // schema, proving `document_from_raw`'s two branches against a live + // server instead of only the hand-built bytes the unit tests use. + let topic_2_id: Identifier = seeds::names::TOPIC_2.try_into().unwrap(); + + // Raw bytes that happen to parse as JSON: document_from_raw hands the + // parsed object to document_from_json exactly like the Payload::Json + // path, so the natural-key document ID must still resolve from + // `order_id` even though the message arrived through the raw decoder. + let mut raw_json_message = vec![ + IggyMessage::builder() + .id(5) + .payload(Bytes::from( + serde_json::to_vec(&serde_json::json!({ + "order_id": "R-1", + "name": "raw-json-object", + })) + .expect("serialize"), + )) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream_id, + &topic_2_id, + &Partitioning::partition_id(0), + &mut raw_json_message, + ) + .await + .expect("send raw JSON message"); + + // Raw bytes that are not valid JSON: the simd_json parse fails, falling + // back to base64-encoding the bytes verbatim. This message has no + // `order_id` field, so it also exercises the generated-document-ID + // fallback, which previously had no live-server coverage either. + let non_json_bytes: Vec = vec![0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0xFF]; + let mut raw_binary_message = vec![ + IggyMessage::builder() + .id(6) + .payload(Bytes::from(non_json_bytes.clone())) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream_id, + &topic_2_id, + &Partitioning::partition_id(0), + &mut raw_binary_message, + ) + .await + .expect("send raw binary message"); + + let count_after_raw = fixture + .wait_for_document_count(5) + .await + .expect("wait for documents after raw sends"); + assert_eq!(count_after_raw, 5); + + let documents = fixture + .search_all(fixture.index()) + .await + .expect("search index"); + let hits = documents["hits"]["hits"] + .as_array() + .expect("hits array") + .clone(); + + let raw_json_hit = hits + .iter() + .find(|hit| hit["_source"]["order_id"] == "R-1") + .expect("raw JSON document indexed"); + assert_eq!(raw_json_hit["_source"]["name"], "raw-json-object"); + assert_eq!( + raw_json_hit["_id"], "R-1", + "raw bytes that parse as JSON must still resolve the natural key" + ); + + let raw_binary_hit = hits + .iter() + .find(|hit| { + hit["_source"]["data_type"] == "raw" && hit["_source"]["data_encoding"] == "base64" + }) + .expect("raw binary document indexed via base64 fallback"); + assert_eq!( + raw_binary_hit["_source"]["data"], + general_purpose::STANDARD.encode(&non_json_bytes) + ); + assert!( + !raw_binary_hit["_id"] + .as_str() + .unwrap_or_default() + .is_empty(), + "fallback document must still get a generated ID" + ); +} + +async fn wait_for_updated_name( + fixture: &OpenSearchSinkFixture, + expected_name: &str, +) -> serde_json::Value { + const POLL_ATTEMPTS: usize = 40; + const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100); + + for _ in 0..POLL_ATTEMPTS { + if let Ok(documents) = fixture.search_all(fixture.index()).await + && let Some(hits) = documents["hits"]["hits"].as_array() + && let Some(hit) = hits.iter().find(|hit| hit["_source"]["order_id"] == "A-1") + && hit["_source"]["name"] == expected_name + { + return hit.clone(); + } + tokio::time::sleep(POLL_INTERVAL).await; + } + panic!( + "order A-1 did not show name={expected_name:?} within {} attempts", + POLL_ATTEMPTS + ); +} diff --git a/core/integration/tests/connectors/opensearch/opensearch_sink_failures.rs b/core/integration/tests/connectors/opensearch/opensearch_sink_failures.rs new file mode 100644 index 0000000000..3dc9319a1c --- /dev/null +++ b/core/integration/tests/connectors/opensearch/opensearch_sink_failures.rs @@ -0,0 +1,401 @@ +// 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. + +//! Real-infra failure-state coverage for the OpenSearch sink, mirroring +//! `runtime/error_isolation.rs` but driving the failures through this sink's +//! own code paths against a live OpenSearch server rather than a +//! misconfigured `stdout_sink`. +//! +//! Two failure states, both configured in `failure_states/`, with two +//! structurally different outcomes: +//! +//! * `opensearch_missing_index` never reaches `Running`. The target index +//! does not exist and `create_index_if_not_exists = false`, so `open()` +//! fails with `Error::InitError`. This goes through the sink init path +//! (`runtime/src/sink.rs::init` / `manager/sink.rs::set_error`), which +//! genuinely does flip `ConnectorStatus::Error`, observable via +//! `/sinks`. +//! +//! * `opensearch_mapping_conflict` reaches `Running`, indexes a first +//! document successfully, then receives a batch containing one document +//! that violates its own explicit `count: integer` mapping. OpenSearch +//! answers that bulk call with a `mapper_parsing_exception`: HTTP 200 at +//! the top level, with the failure in a per-item `items[]` entry, and +//! `Sink::consume()` returns `Err`. +//! +//! That `Err` does not halt the connector. `process_messages` in +//! `runtime/src/sink.rs` invokes the FFI `consume` callback and discards +//! its return code: +//! +//! ```ignore +//! (consume)(plugin_id, ..., messages.as_ptr(), messages.len()); +//! Ok(SinkBatchTiming { processed_count, decode_elapsed, ffi_elapsed }) +//! ``` +//! +//! `process_messages` always returns `Ok`, so `consume_messages`'s +//! `if let Err(error) = result { return Err(error); }` fires only on the +//! runtime's own internal failures (serialization, missing message +//! fields), never on a plugin-returned `Err`. `ConnectorStatus` never +//! leaves `Running`, and `/stats` `errors` (fed only by +//! `process_messages`'s own decode/transform/field-validation failures) +//! never increments. The malformed document is dropped with no +//! API-visible trace; the only record is this connector's own `tracing` +//! output. Subscribes to `test_topic_2`, distinct from the healthy +//! sink's `test_topic`, so it never receives the healthy sink's +//! messages or vice versa. +//! +//! `opensearch_healthy` is the control: it subscribes to a different topic +//! and must keep indexing normally throughout, proving the runtime process +//! (`/health`) and sibling connectors are unaffected. The stronger proof +//! point is that `opensearch_mapping_conflict` itself survives its own +//! failure and keeps indexing later messages. + +use crate::connectors::fixtures::{OpenSearchFailureFixture, OpenSearchOps}; +use bytes::Bytes; +use iggy::prelude::{IggyMessage, Partitioning}; +use iggy_common::{Identifier, MessageClient}; +use iggy_connector_sdk::api::{ConnectorError, ConnectorStatus, HealthResponse, SinkInfoResponse}; +use integration::harness::seeds; +use integration::iggy_harness; +use reqwest::Client; +use std::time::Duration; +use tokio::time::sleep; + +const POLL_ATTEMPTS: usize = 50; +const POLL_INTERVAL: Duration = Duration::from_millis(100); + +async fn assert_runtime_healthy(http_client: &Client, api_address: &str) { + let response = http_client + .get(format!("{api_address}/health")) + .send() + .await + .expect("Failed to query health endpoint"); + assert_eq!(response.status(), 200); + let health: HealthResponse = response + .json() + .await + .expect("Failed to parse health response"); + assert_eq!(health.status, "healthy"); +} + +async fn fetch_sinks(http_client: &Client, api_address: &str) -> Vec { + let response = http_client + .get(format!("{api_address}/sinks")) + .send() + .await + .expect("Failed to query /sinks"); + assert_eq!(response.status(), 200); + response.json().await.expect("Failed to parse sinks") +} + +/// Returns the matching sink's `last_error`, if any, once it reaches `status`. +/// `SinkInfoResponse` is not `Clone`, so only the field this test needs is +/// carried out of the poll loop. +async fn wait_for_status( + http_client: &Client, + api_address: &str, + key: &str, + status: ConnectorStatus, +) -> Option { + for _ in 0..POLL_ATTEMPTS { + let sinks = fetch_sinks(http_client, api_address).await; + if let Some(sink) = sinks.into_iter().find(|sink| sink.key == key) + && sink.status == status + { + return sink.last_error; + } + sleep(POLL_INTERVAL).await; + } + panic!("Sink '{key}' did not reach status {status:?} within {POLL_ATTEMPTS} attempts"); +} + +#[iggy_harness( + server(connectors_runtime( + config_path = "tests/connectors/opensearch/failure_states.toml" + )), + seed = seeds::connector_multi_topic_stream +)] +async fn given_missing_index_and_mapping_conflict_should_isolate_failures_from_healthy_sibling( + harness: &TestHarness, + fixture: OpenSearchFailureFixture, +) { + let api_address = harness + .connectors_runtime() + .expect("connector runtime should be available") + .http_url(); + let http_client = Client::new(); + let iggy_client = harness.root_client().await.unwrap(); + + // opensearch_missing_index never opens successfully: assert this first, + // since it requires no message traffic at all. + // `SinkContainer::open` (sdk/src/sink.rs) collapses any `Result<(), Error>` + // from a plugin's own `open()` to a bare 0/1 at the FFI boundary, and the + // runtime only ever records the generic "Plugin initialization failed + // (ID: N)" for that case (runtime/src/sink.rs), so this connector's + // `Error::InitError("... does not exist ...")` text never crosses the FFI + // call. Uniform SDK behavior, not fixable per-connector: meilisearch_sink + // has the identical gap. The only assertion the API can support here is + // that the connector reached Error with some last_error at all. + wait_for_status( + &http_client, + &api_address, + "opensearch_missing_index", + ConnectorStatus::Error, + ) + .await + .expect("missing-index sink should expose a last_error"); + + // opensearch_healthy and opensearch_mapping_conflict both open fine. + wait_for_status( + &http_client, + &api_address, + "opensearch_healthy", + ConnectorStatus::Running, + ) + .await; + wait_for_status( + &http_client, + &api_address, + "opensearch_mapping_conflict", + ConnectorStatus::Running, + ) + .await; + + let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap(); + let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap(); + let topic_2_id: Identifier = seeds::names::TOPIC_2.try_into().unwrap(); + + let send = |topic_id: Identifier, payload: serde_json::Value, message_id: u128| { + let stream_id = stream_id.clone(); + let client = &iggy_client; + async move { + let mut messages = vec![ + IggyMessage::builder() + .id(message_id) + .payload(Bytes::from( + serde_json::to_vec(&payload).expect("serialize"), + )) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream_id, + &topic_id, + &Partitioning::partition_id(0), + &mut messages, + ) + .await + .expect("send message"); + } + }; + + // A valid message to the healthy sink's topic, proving it indexes real + // traffic before the sibling failure is introduced. + send( + topic_id.clone(), + serde_json::json!({ "name": "healthy-before" }), + 1, + ) + .await; + + // A valid message on the mapping-conflict sink's own topic first, so its + // Running status and correct indexing are proven before it fails. + send( + topic_2_id.clone(), + serde_json::json!({ "name": "before-failure", "count": 1 }), + 2, + ) + .await; + + let index = fixture.mapping_conflict_index(); + wait_for_document_count(&fixture, index, 1).await; + + // One valid document plus one that violates the pinned `count: integer` + // mapping, sent together so both land in the same `_bulk` call. + let mut mixed_batch = vec![ + IggyMessage::builder() + .id(3) + .payload(Bytes::from( + serde_json::to_vec(&serde_json::json!({ "name": "also-ok", "count": 2 })) + .expect("serialize"), + )) + .build() + .expect("build message"), + IggyMessage::builder() + .id(4) + .payload(Bytes::from( + serde_json::to_vec(&serde_json::json!({ + "name": "should-fail", + "count": "not-a-number" + })) + .expect("serialize"), + )) + .build() + .expect("build message"), + ]; + iggy_client + .send_messages( + &stream_id, + &topic_2_id, + &Partitioning::partition_id(0), + &mut mixed_batch, + ) + .await + .expect("send mixed batch"); + + // The valid document from the mixed batch still landed: 1 from before + // the failure + 1 valid document from the failing batch = 2. Proves + // partial per-item credit against a real server, not just a JSON + // fixture. The malformed document never appears; it is gone permanently. + let document_count = wait_for_document_count(&fixture, index, 2).await; + assert_eq!( + document_count, 2, + "the valid document in the mixed batch should still be indexed" + ); + + // The connector itself survives its own consume() failure: no status + // change, no last_error, no /stats error count. See the module doc. + let sinks = fetch_sinks(&http_client, &api_address).await; + let mapping_conflict_sink = sinks + .iter() + .find(|sink| sink.key == "opensearch_mapping_conflict") + .expect("mapping-conflict sink should be reported"); + assert_eq!( + mapping_conflict_sink.status, + ConnectorStatus::Running, + "a consume()-level Err does not flip connector status; it is silently absorbed" + ); + assert!( + mapping_conflict_sink.last_error.is_none(), + "a consume()-level Err never populates last_error" + ); + + // And it keeps indexing normally afterward: a third valid message on the + // same topic must still land, proving this is not a lingering degraded + // state: the connector is fully healthy, just missing one document. + send( + topic_2_id.clone(), + serde_json::json!({ "name": "after-failure", "count": 3 }), + 5, + ) + .await; + let final_count = wait_for_document_count(&fixture, index, 3).await; + assert_eq!( + final_count, 3, + "the connector must keep indexing later messages after an unnoticed consume() failure" + ); + + // This sink runs batch_size = 2, so these six messages are intended to + // span three `_bulk` calls with the mapping violation in the first: + // [bad, ok] [ok, ok] [ok, ok]. Five of the six are indexable, and all + // five have to land: the runtime discards a consume()-level error and + // commits the offset regardless, so anything not indexed here is gone + // with no redelivery. + // + // This assertion alone can't distinguish "chunking happened as three + // separate `_bulk` calls" from "the whole batch went out in one `_bulk` + // call": OpenSearch's own per-item semantics make the final document + // count identical either way, since a single call still reports per-item + // success/failure. The deterministic proof that `index_documents` really + // does send N separate chunked calls and keeps going after a permanently + // failing one lives at the unit level, against a mocked server that can + // assert per-chunk call counts: + // `given_permanently_failing_chunk_should_not_abandon_later_chunks` in + // `opensearch_sink/src/lib.rs`. What this test proves that the unit test + // can't is that real documents genuinely survive in a live index across + // that chunk boundary. + let mut chunked_batch = vec![ + IggyMessage::builder() + .id(6) + .payload(Bytes::from( + serde_json::to_vec(&serde_json::json!({ + "name": "chunk-poison", + "count": "not-a-number" + })) + .expect("serialize"), + )) + .build() + .expect("build message"), + ]; + for index_in_batch in 0..5u128 { + chunked_batch.push( + IggyMessage::builder() + .id(7 + index_in_batch) + .payload(Bytes::from( + serde_json::to_vec(&serde_json::json!({ + "name": format!("chunk-ok-{index_in_batch}"), + "count": 10 + index_in_batch + })) + .expect("serialize"), + )) + .build() + .expect("build message"), + ); + } + iggy_client + .send_messages( + &stream_id, + &topic_2_id, + &Partitioning::partition_id(0), + &mut chunked_batch, + ) + .await + .expect("send chunked batch"); + + let chunked_count = wait_for_document_count(&fixture, index, 8).await; + assert_eq!( + chunked_count, 8, + "every chunk after a failing one must still be indexed" + ); + + // The healthy sibling, on a different topic, never saw the conflicting + // document, actually indexed the traffic it did receive, and must still + // be running. Refetched rather than reusing the snapshot taken above: + // two more sends have happened since then, so a stale snapshot would + // pass even if the runtime had since flipped its status. + wait_for_document_count(&fixture, fixture.healthy_index(), 1).await; + let sinks = fetch_sinks(&http_client, &api_address).await; + let healthy_sink = sinks + .iter() + .find(|sink| sink.key == "opensearch_healthy") + .expect("healthy sink should be reported"); + assert_eq!(healthy_sink.status, ConnectorStatus::Running); + assert!( + healthy_sink.last_error.is_none(), + "healthy sibling sink should have no last_error" + ); + + // And the runtime process itself is unaffected by either failure. + assert_runtime_healthy(&http_client, &api_address).await; +} + +async fn wait_for_document_count( + fixture: &OpenSearchFailureFixture, + index: &str, + expected: u64, +) -> u64 { + for _ in 0..POLL_ATTEMPTS { + if let Ok(count) = fixture.count_documents(index).await + && count >= expected + { + return count; + } + sleep(POLL_INTERVAL).await; + } + panic!("index '{index}' did not reach {expected} documents within {POLL_ATTEMPTS} attempts"); +} diff --git a/core/integration/tests/connectors/opensearch/sink.toml b/core/integration/tests/connectors/opensearch/sink.toml new file mode 100644 index 0000000000..e2d8eb5492 --- /dev/null +++ b/core/integration/tests/connectors/opensearch/sink.toml @@ -0,0 +1,20 @@ +# 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. + +[connectors] +config_type = "local" +config_dir = "../connectors/sinks/opensearch_sink" diff --git a/scripts/bump-version.sh b/scripts/bump-version.sh index 88413af9d2..ab63003ef8 100755 --- a/scripts/bump-version.sh +++ b/scripts/bump-version.sh @@ -87,7 +87,7 @@ EOF } RUST_COMPONENTS="rust-sdk rust-common rust-binary-protocol rust-server rust-cli rust-connector-sdk rust-mcp rust-bench rust-bench-dashboard-frontend rust-bench-dashboard-server rust-bench-report" -CONNECTOR_SINK_COMPONENTS="rust-connector-delta-sink rust-connector-elasticsearch-sink rust-connector-http-sink rust-connector-iceberg-sink rust-connector-influxdb-sink rust-connector-mongodb-sink rust-connector-postgres-sink rust-connector-quickwit-sink rust-connector-stdout-sink rust-connector-surrealdb-sink" +CONNECTOR_SINK_COMPONENTS="rust-connector-delta-sink rust-connector-elasticsearch-sink rust-connector-http-sink rust-connector-iceberg-sink rust-connector-influxdb-sink rust-connector-mongodb-sink rust-connector-opensearch-sink rust-connector-postgres-sink rust-connector-quickwit-sink rust-connector-stdout-sink rust-connector-surrealdb-sink" CONNECTOR_SOURCE_COMPONENTS="rust-connector-elasticsearch-source rust-connector-influxdb-source rust-connector-postgres-source rust-connector-random-source" CONNECTOR_COMPONENTS="rust-connector-runtime ${CONNECTOR_SINK_COMPONENTS} ${CONNECTOR_SOURCE_COMPONENTS}" SDK_COMPONENTS="sdk-python sdk-node sdk-go sdk-csharp sdk-java"