github-actions[bot] commented on code in PR #3873: URL: https://github.com/apache/iggy/pull/3873#discussion_r4126792079
########## 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.4" Review Comment: nit: `version = "0.5.0-edge.4"` is the only `-edge` version in the workspace; the runtime, the SDK, and every sibling sink use `0.5.0`. Set it to `0.5.0` and let `scripts/bump-version.sh` own later bumps. ########## core/connectors/sinks/opensearch_sink/README.md: ########## @@ -0,0 +1,205 @@ +# 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, shared by both bulk requests and `open()`-time requests + (see `max_open_retries` below). 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`. Paced by the same `retry_delay` / + `max_retry_delay` backoff bounds used for bulk requests. 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: <base64>, 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: <string>, 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.<key>` 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 Review Comment: warning: A `consume()` failure does reach `/stats`. The runtime increments the sink's error counter on a non-zero FFI result (`core/connectors/runtime/src/sink.rs:806`), and `core/connectors/runtime/src/stats.rs:91` reports that counter as the connector's `errors`. Drop `/stats` from the list of places the failure is invisible, and say the counter rises once per failed run. ########## core/integration/tests/connectors/fixtures/opensearch/container.rs: ########## @@ -0,0 +1,238 @@ +// 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 std::time::Duration; +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 tokio::time::sleep; +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<GenericImage>, + pub base_url: String, +} + +/// See [`start_shared_container`]: same create-or-attach race as the +/// Elasticsearch fixture, same bound. +const CONTAINER_START_ATTEMPTS: u32 = 30; +const CONTAINER_START_RETRY_DELAY: Duration = Duration::from_secs(1); + +async fn start_shared_container() -> Result<ContainerAsync<GenericImage>, TestBinaryError> { + let mut conflict = String::new(); + for attempt in 1..=CONTAINER_START_ATTEMPTS { + // Rebuilt per attempt because `start` consumes the request. + let result = 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(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; + + match result { + Ok(container) => return Ok(container), + Err(error) => { + let message = error.to_string(); + if !message.contains("is already in use") { + return Err(TestBinaryError::FixtureSetup { + fixture_type: "OpenSearchContainer".to_string(), + message: format!("Failed to start container: {message}"), + }); + } + info!( + "OpenSearch container name taken by another test (attempt {attempt}), retrying to attach" + ); + conflict = message; + sleep(CONTAINER_START_RETRY_DELAY).await; + } + } + } + + Err(TestBinaryError::FixtureSetup { + fixture_type: "OpenSearchContainer".to_string(), + message: format!( + "Failed to attach to container '{OPENSEARCH_CONTAINER_NAME}' after \ + {CONTAINER_START_ATTEMPTS} attempts: {conflict}" + ), + }) +} + +impl OpenSearchContainer { + pub async fn start() -> Result<Self, TestBinaryError> { Review Comment: nit: `start()` returns the reused container without checking that it is serving. The Elasticsearch fixture re-runs `/_cluster/health` on every attach, since reuse can hand back a days-old container (`core/integration/tests/connectors/fixtures/elasticsearch/container.rs:256`). Add the same readiness poll before returning. ########## 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.4" +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 } +tokio = { workspace = true } +tracing = { workspace = true } +url = { workspace = true } + +[dev-dependencies] +simd-json = { workspace = true } +toml = { workspace = true } +wiremock = { workspace = true } Review Comment: nit: The crate omits `[lints] workspace = true`, so it does not inherit `warnings = "deny"` (`Cargo.toml:83`), unlike every other connector crate. Add the table at the end of the file. ########## 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 Review Comment: nit: The module doc says `/stats` `errors` never increments, and the test skips that assertion. The runtime does increment that counter on a failed `consume()` (`core/connectors/runtime/src/sink.rs:806`). Correct the doc and assert the counter, so the test proves the failure is observable. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
