feat(connectors): opensearch source connector

[ryerraguntla]

Which issue does this PR close?
Closes #3503

Rationale
Iggy had no native way to ingest documents from OpenSearch clusters into Iggy streams. This connector fills that gap for observability pipelines, audit log forwarding, and data lake ingestion scenarios where OpenSearch is the source of truth.

What changed?
Before: no OpenSearch source connector existed. Users had to write custom consumers using the OpenSearch client to bridge documents into Iggy.

After: a new iggy_connector_opensearch_source cdylib plugin implements the Source trait and is loaded by the connectors runtime via FFI. Each poll() issues a search_after-paginated query on (timestamp_field, _id), publishes each hit's _source as a JSON ProducedMessage, and persists the cursor in ConnectorState.

Module layout

core/connectors/sources/opensearch_source/
├── Cargo.toml
├── README.md
├── dependencies.md
└── src/
    ├── lib.rs            # Source trait, poll loop, search_after cursor, config, 16 unit tests
    ├── state_manager.rs  # File-backed SourceState mirror, atomic rename protocol, 4 unit tests
    ├── retry.rs  # Retry logic and circuit breaker logic. 2 unit tests 
    └── http_tests.rs     # Axum mock server tests (20 tests, no Docker required)
core/connectors/runtime/example_config/connectors/
└── opensearch_source.toml
core/integration/tests/connectors/opensearch/
├── mod.rs
├── opensearch_source.rs      # 7 Docker-backed integration tests
├── opensearch_source_types.rs
├── opensearch_source_resilience.rs
└── fixtures/opensearch/      # Container start/stop, document indexing helpers

Key implementation decisionsCursor mechanism — 
search_after on (timestamp_field, _id) sorted ascending.
Two sort keys give stable pagination when timestamps collide; _id is the tiebreaker. No Scroll API, PIT, or offset paging.

State authority — runtime ConnectorState (MessagePack) is authoritative. An optional file-backed JSON mirror (SourceState) is written atomically on close() using write-tmp → fdatasync → rename → dir-fsync. On restart, if a valid runtime state exists it takes priority and file state is not loaded. If state deserialization fails, open() returns InitError to prevent a silent cursor reset.

Validation at open() — timestamp_field is required and non-empty (validated before any network call). batch_size = 0 is rejected. File state config is validated if state.enabled = true.

Error handling — transient OpenSearch errors fail the poll with the error counter incremented. No circuit breaker or HTTP retry in this version (documented limitation). poll() still sleeps polling_interval on error so the runtime does not spin.

At-least-once delivery — cursor advances in memory before ConnectorState is persisted by the runtime. A crash after poll() returns but before the runtime writes state can re-emit the last batch. Documented in README and Limitations.

Validation

cargo fmt --all
cargo sort --no-format --workspace
cargo clippy -p iggy_connector_opensearch_source --all-targets -- -D warnings
cargo test -p iggy_connector_opensearch_source
./scripts/ci/taplo.sh
./scripts/ci/license-headers.sh
# Docker-backed integration (requires running iggy-server and iggy-connectors binaries):
env CARGO_BIN_EXE_iggy-server=... CARGO_BIN_EXE_iggy-connectors=... \
  cargo test -p integration --test mod -- connectors::opensearch::opensearch_source

All 40 tests pass (16 unit + 4 state_manager unit + 20 axum HTTP mock + 7 Docker integration).

Local Execution
Passed
Pre-commit hooks ran

AI Usage

1. Tools used: Claude
2. Scope: scaffolding initial file structure and boilerplate; all logic reviewed and iterated manually
3. Verification: full local compilation, all tests executed, end-to-end tested with Docker OpenSearch + Iggy server + connector runtime
4. Can explain every line: yes

Configuration reference

type = "source"
key = "opensearch"
enabled = true
version = 0
name = "OpenSearch source"
path = "<BASE_DIR>/target/release/libiggy_connector_opensearch_source"
plugin_config_format = "json"
[[streams]]
stream = "opensearch_stream"
topic  = "documents"
schema = "json"
batch_length = 100
linger_time  = "5ms"
[plugin_config]
url              = "http://localhost:9200"
index            = "logs-*"
timestamp_field  = "@timestamp"       # required; must exist on every document
polling_interval = "10s"              # default
batch_size       = 100                # default
# username       = "admin"
# password       = "replace_with_secret"
# verbose_logging = false
# Optional: custom query DSL (default: match_all)
# [plugin_config.query]
# match = { "log.level" = "error" }
# Optional: file-backed state mirror (runtime msgpack is authoritative)
# [plugin_config.state]
# enabled      = true
# storage_type = "file"
# state_id     = "opensearch_logs_connector"
# [plugin_config.state.storage_config]
# base_path = "./connector_states"

Required Fields

Field	Type	Description
url	String	OpenSearch HTTP base URL
index	String	Index name or pattern (logs-* supported)
timestamp_field	String	Document field used as sort key and cursor; must exist on every document

Optional Fields

Field	Type	Default	Description
polling_interval	humantime	"10s"	Sleep after each completed poll cycle
batch_size	usize	100	Documents per search request (min 1)
query	JSON	{"match_all":{}}	OpenSearch query DSL applied on every poll
username / password	String	—	HTTP Basic auth
verbose_logging	bool	false	Log per-poll counts at info! instead of debug!

Poll cycle

Each poll() call:Issues POST /{index}/_search sorted by (timestamp_field asc, _id asc) with optional search_after.Maps each hit's _source to a JSON ProducedMessage.Advances the search_after cursor to the sort tuple of the last successfully published hit.Returns ProducedMessages with serialized ConnectorState.Sleeps polling_interval.Hits missing a sort tuple or _source are skipped with warn!; they do not block the cursor.

Known limitations

1. At-least-once delivery — cursor advances before the runtime persists ConnectorState; last batch may be re-emitted on crash.
2. Single sequential reader — one search_after cursor, one batch per poll; no parallel shard workers.
3. search_after only — no Scroll API, PIT, or sliced export.
4. Backfill gap — documents with timestamp_field older than the current cursor are not read without a state reset.
5. _source-disabled documents skipped permanently.Single-node transport — no cluster sniffing.

[github-actions]

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]

Codecov Report

❌ Patch coverage is 86.16279% with 119 lines in your changes missing coverage. Please review.
✅ Project coverage is 46.85%. Comparing base (4a48008) to head (f8b7e7b).

Files with missing lines	Patch %	Lines
...re/connectors/sources/opensearch_source/src/lib.rs	85.16%	83 Missing and 17 partials ⚠️
...ors/sources/opensearch_source/src/state_manager.rs	88.07%	10 Missing and 8 partials ⚠️
.../connectors/sources/opensearch_source/src/retry.rs	97.14%	1 Missing ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master    #3515       +/-   ##
=============================================
- Coverage     74.27%   46.85%   -27.42%     
  Complexity      937      937               
=============================================
  Files          1259     1259               
  Lines        125969   110774    -15195     
  Branches     101644    86493    -15151     
=============================================
- Hits          93558    51899    -41659     
- Misses        29396    56146    +26750     
+ Partials       3015     2729      -286     

Components	Coverage Δ
Rust Core	39.79% <86.16%> (-35.37%)	⬇️
Java SDK	58.57% <ø> (ø)
C# SDK	71.41% <ø> (-0.70%)	⬇️
Python SDK	88.88% <ø> (ø)
PHP SDK	84.29% <ø> (ø)
Node SDK	91.35% <ø> (+0.12%)	⬆️
Go SDK	40.36% <ø> (ø)

Files with missing lines	Coverage Δ
.../connectors/sources/opensearch_source/src/retry.rs	97.14% <97.14%> (ø)
...ors/sources/opensearch_source/src/state_manager.rs	88.07% <88.07%> (ø)
...re/connectors/sources/opensearch_source/src/lib.rs	85.16% <85.16%> (ø)

... and 364 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.

[ryerraguntla]

/ready