feat(connectors): add OpenSearch sink connector - #3873
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3873 +/- ##
============================================
- Coverage 82.84% 81.98% -0.86%
Complexity 1299 1299
============================================
Files 1199 1200 +1
Lines 161885 163626 +1741
Branches 131360 133206 +1846
============================================
+ Hits 134120 134156 +36
- Misses 24225 25831 +1606
- Partials 3540 3639 +99
🚀 New features to boost your workflow:
|
|
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. |
| let Some(items) = response.get("items").and_then(Value::as_array) else { | ||
| return BulkAttempt::default(); | ||
| }; |
There was a problem hiding this comment.
A successful HTTP response with no items array is converted to an empty BulkAttempt, and malformed entries are skipped later in the loop. index_chunk() can therefore return a clean outcome even though none of the submitted documents were accounted for, causing the batch to be treated as successfully handled. Make the parser return a Result, require exactly one valid item result per pending document, and treat a missing, short, or malformed items array as a retryable response-parsing failure.
| pub verbose_logging: Option<bool>, | ||
| } | ||
|
|
||
| #[derive(Debug)] |
There was a problem hiding this comment.
Deriving Debug for the sink leaks the Basic-auth password after open(): OpenSearch derives Debug, its Transport includes credentials, and opensearch 2.4.0 derives Debug for Credentials::Basic(String, String) without redaction. The existing test formats only an unopened sink, where client is still None, so it misses this path. Replace the derive with a custom Debug implementation that omits or redacts client, and add a regression test that formats a sink after installing an authenticated client.
| // 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)), |
There was a problem hiding this comment.
Every raw payload is cloned in the consume hot path so that simd_json can mutate one copy while preserving the other for the base64 fallback. For large batches this temporarily doubles the payload memory and copies every byte. Parse the owned bytes non-destructively, for example with serde_json::from_slice, so the same buffer can be base64-encoded on parse failure without cloning it.
| let delay = jitter(exponential_backoff( | ||
| self.config.retry_delay, | ||
| retries, | ||
| self.config.max_retry_delay, | ||
| )); |
There was a problem hiding this comment.
exponential_backoff() computes base * 2^attempt, but the first retry passes retries == 1, so the configured retry_delay is never used and the first wait is already twice the base. Jitter is also applied after the helper caps at max_retry_delay, allowing the actual sleep to exceed that advertised maximum by up to 20%. Pass retries - 1 as the attempt index and clamp the jittered result to max_retry_delay.
| let with_scheme = if trimmed.starts_with("http://") || trimmed.starts_with("https://") { | ||
| trimmed.to_string() | ||
| } else { | ||
| format!("http://{trimmed}") | ||
| }; |
There was a problem hiding this comment.
Scheme detection is case-sensitive, so a valid URL such as HTTPS://host:9200 is rewritten as http://HTTPS://host:9200 and targets the wrong host; unsupported explicit schemes are similarly disguised instead of rejected. Detect an explicit scheme case-insensitively, accept only HTTP or HTTPS, and prepend http:// only when no scheme was supplied.
There was a problem hiding this comment.
Starting a fixed-name ReuseDirective::Always container is a check-then-create operation with no cross-process lock, so concurrent nextest invocations on a cold Docker daemon can race and all but one fail with 409 Conflict: name is already in use. Use the bounded create-or-attach retry now present in the Elasticsearch fixture so losers retry until they attach to the winning container.
github-actions
Bot
added
S-waiting-on-author
2 participants
Which issue does this PR address?
Closes #3504
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_sinkshape 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_fieldconstraints), thenretry_on_open-wraps a cluster health check, an index-exists check, and, whencreate_index_if_not_exists(defaulttrue), index creation with an optional custom mapping. Capped atmax_open_retrieswith exponential backoff and jitter (sharedretryhelpers).consume(): batches incoming messages intobatch_sizechunks, builds aniggy_*-enriched document per message (hashed or field-derived_id), and hands each chunk toindex_chunk.index_chunk(): POSTs a_bulkrequest and loops up tomax_retrieswith the same backoff._bulkanswers200even 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 viaReuseDirective::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);passwordis aSecretString, 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_sinkhas the same gap this PR fixes for OpenSearch:bulk_index_documents(elasticsearch_sink/src/lib.rs:205-219) tallies per-item_bulkfailures intoerrors_countbut never retries the transient subset (e.g. 429es_rejected_execution_exception). Worth a follow-up issue rather than folding into this PR.consume()return value:core/connectors/runtime/src/sink.rs:740-748invokes the FFIconsumecallback as a bare statement, never binding itsi32result, soprocess_messagesalways returnsOk. 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) andelasticsearch_sink(lib.rs:319-328) both silently dropiggy_headers/_iggy_headers:BTreeMap<HeaderKey, HeaderValue>can't serialize as a JSON object (serde_jsonrequires string keys), and both sinks swallow that error viaif let Ok(...)instead of surfacing it.core/commoneven ships aserialize_headersworkaround for this exact case that neither sink uses.elasticsearch_source, the client is built onSingleNodeConnectionPool(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
AI Usage