ryerraguntla opened a new pull request, #3515:
URL: https://github.com/apache/iggy/pull/3515
**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"
```
<h2>Required Fields</h2>
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
</body></html>
<h2 class="text-text-100 mt-3 -mb-1 text-[1.125rem] font-bold"
data-sourcepos="11:1-11:19;335-353">Optional Fields</h2>
<div class="overflow-x-auto w-full px-2 mb-6"
data-sourcepos="13:1-19:92;355-829">
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!
</div>
**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.
--
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]
