feat(connectors): add OpenSearch sink connector by mattp5657 · Pull Request #3873 · apache/iggy

mattp5657 · 2026-08-12T19:29:52Z

Which issue does this PR address?

Rationale

Iggy connectors ship sinks for several external systems but not for OpenSearch, a widely used search/analytics backend. This adds one, modeled on the existing elasticsearch_sink shape but closing a retry gap that sink still has (see Known trade-offs).

What changed?

Adds core/connectors/sinks/opensearch_sink/, following the sink lifecycle end to end:

open(): validates config (URL shape, credential pairing, document_id_field constraints), then retry_on_open-wraps a cluster health check, an index-exists check, and, when create_index_if_not_exists (default true), index creation with an optional custom mapping. Capped at max_open_retries with exponential backoff and jitter (shared retry helpers).
consume(): batches incoming messages into batch_size chunks, builds an iggy_*-enriched document per message (hashed or field-derived _id), and hands each chunk to index_chunk.
index_chunk(): POSTs a _bulk request and loops up to max_retries with the same backoff. _bulk answers 200 even when individual documents fail, so each response is parsed per item rather than trusting the top-level status: permanent failures (4xx, e.g. a mapping conflict) are recorded immediately, while transient ones (429/5xx) shrink the pending set to just the rejected documents and get resent, so a partial rejection under load doesn't re-index or lose the rest of the chunk. Counts from earlier attempts are merged into the outcome so a later failure doesn't erase already-indexed documents from the tally.
close(): drops the client, no special teardown.

Integration tests (core/integration/tests/connectors/opensearch/) run against a real container (testcontainers-modules, reused across tests via ReuseDirective::Always, per-test-unique index names for isolation), covering the happy path plus a static mapping conflict, a missing index with index-creation disabled, and confirming a failing chunk doesn't block chunks queued behind it.

Credentials: HTTP Basic auth only (username/password, both-or-neither validated at config time); password is a SecretString, never logged or serialized. AWS SigV4 (AWS-managed OpenSearch / Serverless) is not supported.

Known trade-offs, deliberately out of scope here, verified against current master:

elasticsearch_sink has the same gap this PR fixes for OpenSearch: bulk_index_documents (elasticsearch_sink/src/lib.rs:205-219) tallies per-item _bulk failures into errors_count but never retries the transient subset (e.g. 429 es_rejected_execution_exception). Worth a follow-up issue rather than folding into this PR.
The connectors runtime discards a sink's consume() return value: core/connectors/runtime/src/sink.rs:740-748 invokes the FFI consume callback as a bare statement, never binding its i32 result, so process_messages always returns Ok. Combined with offsets auto-committing at poll time (sink.rs:522), a plugin-level failure never reaches connector status, last_error, or /stats, and the batch is never redelivered. Pre-existing, repo-wide, affects every sink.
meilisearch_sink (lib.rs:451-454) and elasticsearch_sink (lib.rs:319-328) both silently drop iggy_headers/_iggy_headers: BTreeMap<HeaderKey, HeaderValue> can't serialize as a JSON object (serde_json requires string keys), and both sinks swallow that error via if let Ok(...) instead of surfacing it. core/common even ships a serialize_headers workaround for this exact case that neither sink uses.
Single-node transport: like elasticsearch_source, the client is built on SingleNodeConnectionPool (lib.rs:208) with no cluster sniffing or multi-node failover. A dead configured node fails every request rather than routing around it.

Local Execution

Passed
Pre-commit hooks ran

AI Usage

Tools: Claude
Use Used for scaffolding and iteration.
Verification: Full local compilation, extensive integration testing against a live container.
Reviewed and can explain every line if asked.

github-actions · 2026-08-12T19:30:01Z

Thanks for the PR. It is labeled S-waiting-on-review and queued for review.

Slash commands (own line, regular comment) move it around the queue:

/ready - back to S-waiting-on-review after addressing feedback
/author - flip to S-waiting-on-author while you finish changes
/request-review @user-or-team - request a reviewer

See CONTRIBUTING.md for details.

codecov · 2026-08-12T19:32:15Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 80.82%. Comparing base (d47ff5d) to head (e35204d).

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3873      +/-   ##
============================================
- Coverage     80.83%   80.82%   -0.01%     
+ Complexity     3107     3106       -1     
============================================
  Files          1467     1468       +1     
  Lines        356533   359013    +2480     
  Branches     300421   302902    +2481     
============================================
+ Hits         288194   290178    +1984     
- Misses        61617    62071     +454     
- Partials       6722     6764      +42

Components	Coverage Δ
Rust Core	`80.12% <ø> (-0.01%)`	⬇️
Java SDK	`78.15% <ø> (ø)`
C# SDK	`83.58% <ø> (-0.02%)`	⬇️
Python SDK	`85.67% <ø> (ø)`
PHP SDK	`90.71% <ø> (+0.04%)`	⬆️
Node SDK	`96.44% <ø> (ø)`
Go SDK	`69.24% <ø> (+0.11%)`	⬆️
C++ SDK	`85.51% <ø> (ø)`

Files with missing lines	Coverage Δ
core/connectors/sinks/opensearch_sink/src/lib.rs	`79.58% <ø> (ø)`

... and 35 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

mattp5657 · 2026-08-12T19:43:53Z

Getting some errors with a 503 returned when they passed locally, for example for Typo. Will try running again later:

  Connecting to github.com (github.com)|140.82.112.4|:443... connected.
  HTTP request sent, awaiting response... 503 Service Unavailable
  2026-08-12 19:30:09 ERROR 503: Service Unavailable.

Update: These seem to have resolved.

…success

…sitive

mattp5657 · 2026-08-22T01:15:12Z

/ready

slbotbm · 2026-08-23T22:33:58Z

+    // The top-level flag lets a clean batch skip the per-item scan entirely.
+    if !response
+        .get("errors")
+        .and_then(Value::as_bool)
+        .unwrap_or(true)
+    {
+        return Ok(BulkAttempt {
+            indexed: items.len(),
+            ..BulkAttempt::default()


This fast path still accepts malformed item entries without validating them. For example, {"errors":false,"items":[{}]} passes the length check and is reported as one indexed document, even though the response contains no recognizable index result or successful status. That leaves the malformed-response fix incomplete, and the runtime will commit the offset for a document that was never actually accounted for. Please validate that every item contains the expected operation result and a 2xx status before returning success, and add an errors: false regression case with a malformed item.

slbotbm · 2026-08-23T22:35:18Z

+
+[package]
+name = "iggy_connector_opensearch_sink"
+version = "0.5.0-edge.1"


update to "0.5.0-edge.4"

slbotbm · 2026-08-23T22:38:39Z

+impl std::fmt::Debug for OpenSearchSink {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("OpenSearchSink")
+            .field("id", &self.id)
+            .field("config", &self.config)
+            .field("client", &self.client.is_some())
+            .field("invocations_count", &self.invocations_count)
+            .field("documents_indexed", &self.documents_indexed)
+            .field("errors_count", &self.errors_count)
+            .finish()
+    }
+}


This hides the authenticated client and relies on SecretString for the dedicated password field, but it still prints self.config.url verbatim through self.config. Since embedded URL credentials are rejected only later in open(), formatting a newly constructed sink from https://admin:hunter2@host exposes hunter2. Let's redact or sanitize the URL in the resolved config's Debug output.

slbotbm · 2026-08-23T22:40:43Z

+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }


simd-json is used only by test fixtures inside the #[cfg(test)] module. It should be under [dev-dependencies].

# Conflicts: # Cargo.toml

mattp5657 · 2026-10-06T15:14:18Z

Fixed: lib.rs:470, Proto dropped. Now indexed as JSON or text, with unit and integration tests.
Deferred: lib.rs:464, deep copy. Waiting on perf(connectors): stop cloning JSON leaves when the caller already owns them #4399, then a one-line swap.
Fixed: lib.rs:839, warning spam. Warns once per instance, still counts drops.
Fixed: lib.rs:229, timeout = "0s". Clamped to 1s.
Fixed: lib.rs:98, comment. Names serialize_secret.
Fixed: lib.rs:799, verbose_logging. One log_progress! macro.
Fixed: lib.rs:1309, tuple hash. IDs unchanged, pinned by a test.
Skipped: lib.rs:846, preparation_errors. The compiler needs the catch-all arm anyway.
Fixed: lib.rs:1455, sanitize_url_for_log. Deleted.
Fixed: README:179, /stats. Says the counter rises per failed run.
Skipped: Cargo.toml:20, version. Keeping 0.5.0-edge.4 as requested by slbotbm.
Fixed: Cargo.toml:49, lints. Added [lints] workspace = true.
Fixed: opensearch_sink_failures:52, doc and assertion. Corrected, and the test asserts errors == 1.

mattp5657 · 2026-10-06T15:14:26Z

/ready

hubcio · 2026-10-12T15:24:21Z

+                                    });
+
+                                    let rejected = retry_set.len();
+                                    pending = retry_set;


critical: partial retries can overwrite a newer document with an older item that shares its ID. preserve source order across retries for the same ID.

hubcio · 2026-10-12T15:24:21Z

+        upsert_metadata_field(id, document, field, value);
+    }
+
+    if let Some(headers) = &message.headers


critical: a payload's iggy_headers survives when real headers are absent, so a scalar value can block later documents with headers. remove payload iggy_headers before adding metadata.

hubcio · 2026-10-12T15:24:21Z

+                // through `Serialize`, which only ever sees `&str` and so
+                // must copy `data` again even though it's already owned here.
+                Value::Object(Map::from_iter([
+                    ("data".to_string(), Value::String(data)),


critical: automatic date detection can map a header string as a date, then reject later documents with ordinary strings. explicitly map iggy_headers.*.data as text or keyword before the first write.

hubcio · 2026-10-12T15:24:21Z

+// Connector-owned rather than exposing `opensearch::params::Refresh` directly:
+// keeps this config's wire format under this connector's own control instead
+// of an internal detail of the opensearch crate.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]


warning: refresh=true and refresh=false environment overrides become booleans, which RefreshPolicy rejects at startup. accept booleans alongside the documented strings.

hubcio · 2026-10-12T15:24:21Z

+            );
+        }
+
+        if documents.is_empty() {


warning: consume() reports success after invalid document IDs drop records, so the runtime counts those records as processed without errors. return an error after valid documents finish if any record fails validation.

Should we hold off on this until the runtime-level discussion in #2927 and #2928 settles? Dropping a bad record, counting it and continuing is the runtime's own convention for records that fail decoding or transforms, and the README documents it for this sink. Failing the batch wouldn't recover the records anyway, since offsets are committed at poll time.

hubcio · 2026-10-12T15:24:22Z

+        let mut retries = 0u32;
+
+        loop {
+            let request = client.bulk(BulkParts::None).body(vec![body.clone()]);


warning: Bulk::body wraps the cached bytes in NdBody, which copies the whole body on every request. pass Bytes directly to client.transport().send and preserve the request parameters.

hubcio · 2026-10-12T15:24:22Z

+
+                                    let rejected = retry_set.len();
+                                    pending = retry_set;
+                                    body = build_bulk_body(&self.config.index, &pending)?;


warning: when every item returns 429, index_chunk() serializes the unchanged documents again. reuse the cached body unless the retry set removes documents.

hubcio · 2026-10-12T15:24:22Z

+/// 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>(


simplification: documents_at() checks positions that its caller already bounds to the pending documents. select retry documents directly and remove the impossible-position test.

hubcio · 2026-10-12T15:24:22Z

+        let messages_count = messages.len();
+        let mut documents = Vec::with_capacity(messages_count);
+        let mut invalid_records = 0usize;
+        let mut preparation_errors = 0usize;


simplification: prepare_document() only returns InvalidRecordValue, but the sink tracks separate preparation failures. use one counter and one error branch.

hubcio · 2026-10-12T15:24:22Z

+        assert_eq!(sink.errors_count.load(Ordering::Relaxed), 0);
+    }
+
+    async fn mock_health_sink(server: &MockServer, base_path: &str) -> OpenSearchSink {


simplification: mock_health_sink and mock_bulk_sink return futures even though they only do synchronous work. make both helpers synchronous and remove their callers' awaits.

…ation

mattp5657 · 2026-10-14T11:44:55Z

/ready

# Conflicts: # Cargo.lock # Cargo.toml # core/integration/tests/connectors/fixtures/mod.rs # core/integration/tests/connectors/mod.rs # scripts/bump-version.sh

feat(connectors): add OpenSearch sink connector

d92ea80

github-actions Bot added the S-waiting-on-review PR is waiting on a reviewer label Aug 12, 2026

Merge branch 'master' into feat/connectors-opensearch-sink

6f078e2

slbotbm requested changes Aug 16, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Aug 16, 2026

mattp5657 and others added 7 commits August 21, 2026 18:08

Merge branch 'master' into feat/connectors-opensearch-sink

fa4cc24

fix(connectors): don't count unparsable OpenSearch bulk responses as …

2817a4e

…success

fix(connectors): stop OpenSearch sink Debug from leaking the password

91ba9b0

perf(connectors): stop cloning every raw payload in OpenSearch sink

7adad92

fix(connectors): correct OpenSearch sink retry backoff and jitter cap

d9946e2

fix(connectors): make OpenSearch sink URL scheme detection case-insen…

5fbcf84

…sitive

fix(connectors): retry OpenSearch test container create-or-attach race

8fbad8f

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Aug 22, 2026

slbotbm requested changes Aug 23, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Aug 23, 2026

Matthew Patton added 6 commits September 1, 2026 18:50

fix(connectors): reject malformed items in OpenSearch bulk fast path

eef41fe

chore(connectors): bump OpenSearch sink to 0.5.0-edge.4

7427e83

fix(connectors): redact OpenSearch sink URL credentials in Debug

7fa604f

fix(connectors): move simd-json to dev-dependencies in OpenSearch sink

a1ffd8d

fix(connectors): report first failing chunk in OpenSearch batches

ee210fa

docs(connectors): clarify OpenSearch sink retry backoff scope

aedfdb1

mattp5657 force-pushed the feat/connectors-opensearch-sink branch from db6f3e3 to aedfdb1 Compare September 3, 2026 23:13

Merge branch 'master' into feat/connectors-opensearch-sink

07cbe3d

# Conflicts: # Cargo.toml

mattp5657 added 6 commits October 6, 2026 09:56

fix(connectors): index proto payloads in the OpenSearch sink

0ff70c9

fix(connectors): warn once when the OpenSearch sink drops records

e05f79e

fix(connectors): clamp the OpenSearch sink timeout to one second

32df6ce

refactor(connectors): trim redundant OpenSearch sink helpers

d2a6715

refactor(connectors): pick the OpenSearch sink progress log level once

86e0297

fix(connectors): correct OpenSearch sink docs and enable workspace lints

6010b43

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Oct 6, 2026

Merge branch 'master' into feat/connectors-opensearch-sink

57e6a9a

hubcio requested changes Oct 12, 2026

View reviewed changes

github-actions Bot added S-waiting-on-author PR is waiting on author response and removed S-waiting-on-review PR is waiting on a reviewer labels Oct 12, 2026

mattp5657 added 11 commits October 12, 2026 19:24

Merge branch 'master' into feat/connectors-opensearch-sink

d8336bf

fix(connectors): keep newest opensearch doc when batch repeats an _id

554c122

fix(connectors): drop payload iggy_headers when message has no headers

aea4598

fix(connectors): map opensearch header values as keyword on index cre…

80d9ab6

…ation

fix(connectors): accept boolean refresh values in opensearch config

dc832c8

docs(connectors): document opensearch sink payload variants

937f13d

Merge branch 'master' into feat/connectors-opensearch-sink

58cb135

perf(connectors): move owned JSON into serde_json in the opensearch sink

c770390

perf(connectors): send opensearch bulk body without copying it

9a35e1f

refactor(connectors): simplify opensearch sink retry and error paths

9affa6e

Merge branch 'master' into feat/connectors-opensearch-sink

24a1f31

github-actions Bot added S-waiting-on-review PR is waiting on a reviewer and removed S-waiting-on-author PR is waiting on author response labels Oct 14, 2026

mattp5657 added 2 commits October 26, 2026 19:42

Merge branch 'master' into feat/connectors-opensearch-sink

3661c12

# Conflicts: # Cargo.lock # Cargo.toml # core/integration/tests/connectors/fixtures/mod.rs # core/integration/tests/connectors/mod.rs # scripts/bump-version.sh

Merge branch 'master' into feat/connectors-opensearch-sink

e35204d

Conversation

mattp5657 commented Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Which issue does this PR address?

Rationale

What changed?

Local Execution

AI Usage

Uh oh!

github-actions Bot commented Aug 12, 2026

Uh oh!

codecov Bot commented Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

mattp5657 commented Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mattp5657 commented Aug 22, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mattp5657 commented Oct 6, 2026

Uh oh!

mattp5657 commented Oct 6, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mattp5657 commented Oct 14, 2026

Uh oh!

This branch has not been deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mattp5657 commented Aug 12, 2026 •

edited

Loading

codecov Bot commented Aug 12, 2026 •

edited

Loading

mattp5657 commented Aug 12, 2026 •

edited

Loading