feat(connectors): opensearch source connector

Cargo.toml

@@ -45,6 +45,7 @@ members = [
     "core/connectors/sinks/stdout_sink",
     "core/connectors/sources/elasticsearch_source",
     "core/connectors/sources/influxdb_source",
+    "core/connectors/sources/opensearch_source",
     "core/connectors/sources/postgres_source",
     "core/connectors/sources/random_source",
     "core/consensus",
@@ -212,6 +213,7 @@ nonzero_lit = "0.1.2"
 notify = "8.2.0"
 octocrab = "0.51.0"
 once_cell = "1.21.4"
+opensearch = { version = "2.4.0", features = ["rustls-tls"], default-features = false }
 opentelemetry = { version = "0.32.0", features = ["trace", "logs"] }
 opentelemetry-appender-tracing = { version = "0.32.0", features = ["log"] }
 opentelemetry-otlp = { version = "0.32.0", features = [

core/connectors/runtime/example_config/connectors/opensearch_source.toml

@@ -0,0 +1,63 @@
+# 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.
+
+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-*"
+polling_interval = "10s"
+batch_size = 100
+timestamp_field = "@timestamp"
+# username = "admin"
+# password = "replace_with_secret"
+# verbose_logging = false
+
+# HTTP resilience (defaults shown; see docs/RESILIENCE.md).
+# max_retries = 3
+# retry_delay = "1s"
+# retry_max_delay = "30s"
+# max_open_retries = 5
+# open_retry_max_delay = "30s"
+# circuit_breaker_threshold = 5
+# circuit_breaker_cool_down = "60s"
+
+# Optional: restrict which documents are polled (defaults to match_all).
+# [plugin_config.query]
+# match = { "log.level" = "error" }
+
+# Optional: mirror state to a local JSON file in addition to runtime msgpack state.
+# File state is secondary — runtime ConnectorState (msgpack) is authoritative on restart.
+# [plugin_config.state]
+# enabled = true
+# storage_type = "file"
+# state_id = "opensearch_logs_connector"
+# [plugin_config.state.storage_config]
+# base_path = "./connector_states"

core/connectors/sources/opensearch_source/Cargo.toml

@@ -0,0 +1,58 @@
+# 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_source"
+version = "0.4.1-edge.1"
+description = "Iggy OpenSearch source connector"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "opensearch"]
+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
+
+# dashmap and once_cell are not imported directly in this crate's source, but
+# the source_connector! macro (in iggy_connector_sdk::source) expands bare
+# `use dashmap::DashMap` and `use once_cell::sync::Lazy` into this crate's
+# namespace, so they must be listed here.
+[package.metadata.cargo-machete]
+ignored = ["dashmap", "once_cell"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+once_cell = { workspace = true }
+opensearch = { workspace = true }
+rmp-serde = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+
+[dev-dependencies]
+axum = { workspace = true }
+tempfile = { workspace = true }