feat(connectors): add OTLP gRPC 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/otlp_source",
     "core/connectors/sources/postgres_source",
     "core/connectors/sources/random_source",
     "core/consensus",
@@ -222,6 +223,7 @@ opentelemetry-otlp = { version = "0.32.0", features = [
     "http-proto",
     "reqwest-client",
 ] }
+opentelemetry-proto = { version = "0.32.0", default-features = false }
 opentelemetry-semantic-conventions = "0.32.0"
 opentelemetry_sdk = { version = "0.32.1", features = [
     "logs",
@@ -308,6 +310,7 @@ tokio-rustls = "0.26.4"
 tokio-tungstenite = { version = "0.29", features = ["rustls-tls-webpki-roots"] }
 tokio-util = { version = "0.7.18", features = ["compat"] }
 toml = "1.1.2"
+tonic = { version = "0.14.6", default-features = false }
 tower-http = { version = "0.7.0", features = ["add-extension", "cors", "trace"] }
 tracing = "0.1.44"
 tracing-appender = "0.2.5"

Dockerfile.connectors

@@ -0,0 +1,79 @@
+# 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.
+
+# syntax=docker/dockerfile:1.7
+#
+# iggy-connectors runtime image — ships the iggy-connectors binary and the
+# OTLP/gRPC source, OTLP/gRPC sink, quickwit sink plugins (.so). Config files are injected
+# via K8s ConfigMaps.
+#
+# Builder: rust:1-slim-bookworm (mold linker, BuildKit cache mounts)
+# Runtime: debian:bookworm-slim + libssl3
+
+# ---- Build ----
+FROM rust:1-slim-bookworm AS build
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        pkg-config libssl-dev clang mold \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY . .
+
+ARG TARGETARCH
+RUN --mount=type=cache,id=iggy-connectors-registry,sharing=locked,target=/usr/local/cargo/registry \
+    --mount=type=cache,id=iggy-connectors-git,sharing=locked,target=/usr/local/cargo/git \
+    --mount=type=cache,id=iggy-connectors-target-${TARGETARCH},target=/app/target \
+    RUSTFLAGS="-C linker=clang -C link-arg=-fuse-ld=mold" \
+    cargo build --release --bin iggy-connectors \
+ && cargo build --release -p iggy_connector_otlp_source \
+ && cargo build --release -p iggy_connector_quickwit_sink \
+ && cargo build --release -p iggy_connector_otlp_sink \
+ && cp target/release/iggy-connectors /usr/local/bin/iggy-connectors \
+ && cp target/release/libiggy_connector_otlp_source.so /usr/local/lib/libiggy_connector_otlp_source.so \
+ && cp target/release/libiggy_connector_quickwit_sink.so /usr/local/lib/libiggy_connector_quickwit_sink.so \
+ && cp target/release/libiggy_connector_otlp_sink.so /usr/local/lib/libiggy_connector_otlp_sink.so \
+ && strip /usr/local/bin/iggy-connectors
+
+# ---- Runtime ----
+FROM debian:bookworm-slim
+RUN apt-get update && apt-get install -y --no-install-recommends \
+        ca-certificates libssl3 \
+    && rm -rf /var/lib/apt/lists/* \
+    && useradd --system --uid 10001 --home-dir /connectors \
+        --shell /usr/sbin/nologin iggy \
+    && mkdir -p /connectors/plugins /connectors/state \
+    && chown -R iggy:iggy /connectors
+
+COPY --from=build /usr/local/bin/iggy-connectors          /usr/local/bin/iggy-connectors
+COPY --from=build /usr/local/lib/libiggy_connector_otlp_source.so \
+                                                           /connectors/plugins/libiggy_connector_otlp_source.so
+COPY --from=build /usr/local/lib/libiggy_connector_quickwit_sink.so \
+                                                           /connectors/plugins/libiggy_connector_quickwit_sink.so
+COPY --from=build /usr/local/lib/libiggy_connector_otlp_sink.so \
+                                                           /connectors/plugins/libiggy_connector_otlp_sink.so
+
+USER iggy
+WORKDIR /connectors
+
+# Runtime config: IGGY_CONNECTORS_CONFIG_PATH → /connectors/runtime.toml (ConfigMap)
+# Plugin configs: /connectors/plugins/otlp_source.toml (ConfigMap)
+ENV IGGY_CONNECTORS_CONFIG_PATH=/connectors/runtime.toml
+
+EXPOSE 8081
+
+ENTRYPOINT ["/usr/local/bin/iggy-connectors"]

core/connectors/sources/otlp_source/Cargo.toml

@@ -0,0 +1,59 @@
+# 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_otlp_source"
+version = "0.4.1-edge.1"
+description = "Iggy OTLP/gRPC source connector - receives logs, metrics, and traces from any OpenTelemetry SDK or Collector and writes them to Iggy streams as JSON"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "opentelemetry", "otlp"]
+categories = ["command-line-utilities", "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. Remove them only after the SDK macro
+# is updated to use `$crate::connector_macro_support::{DashMap, Lazy}`.
+[package.metadata.cargo-machete]
+ignored = ["dashmap", "once_cell"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+dashmap = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+once_cell = { workspace = true }
+prost = { workspace = true }
+opentelemetry-proto = { workspace = true, features = [
+    "gen-tonic",
+    "logs",
+    "metrics",
+    "trace",
+] }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tokio = { workspace = true }
+tonic = { workspace = true, features = ["transport", "router", "gzip"] }
+tracing = { workspace = true }

core/connectors/sources/otlp_source/README.md

@@ -0,0 +1,44 @@
+# OTLP Source
+
+Receives logs, metrics, and traces from any OpenTelemetry SDK or Collector
+over gRPC (OTLP/gRPC protocol) and writes them to an Iggy stream as JSON
+messages.
+
+## How it works
+
+The connector binds a gRPC server (default port 4317, the OTLP standard) and
+implements all three collector services: `LogsService`, `MetricsService`, and
+`TraceService`. Each incoming export request is deserialized from the
+`opentelemetry-proto` wire format and serialized to JSON. The JSON messages are
+buffered in an in-process channel and drained by the runtime via the `poll()`
+call.
+
+Gzip compression is enabled on every service. OTel SDKs and the Collector's
+OTLP exporter compress payloads by default, so the connector accepts and sends
+gzip on all three services.
+
+## JSON schema
+
+Each message has a `signal` field (`"log"`, `"metric"`, or `"trace"`) plus
+signal-specific fields:
+
+**Logs** — `timestamp_ns`, `observed_timestamp_ns`, `severity`,
+`severity_text`, `body`, `trace_id`, `span_id`, `service_name`, `attributes`
+
+**Metrics** — `name`, `description`, `unit`, `kind` (gauge/sum/histogram/…),
+`data_points`, `resource`, `attributes`
+
+**Traces** — `trace_id`, `span_id`, `parent_span_id`, `name`, `kind`,
+`start_time_ns`, `end_time_ns`, `status`, `service_name`, `attributes`,
+`events`, `links`
+
+## Configuration
+
+```toml
+[plugin_config]
+listen_addr = "0.0.0.0:4317"   # gRPC bind address
+channel_capacity = 50000         # in-process buffer (messages)
+batch_size = 1000                # max messages returned per poll()
+```
+
+Point any OTel SDK or Collector at `grpc://host:4317` (no TLS by default).

core/connectors/sources/otlp_source/config.toml

@@ -0,0 +1,36 @@
+# 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 = "otlp"
+enabled = true
+version = 0
+name = "OTLP source"
+path = "../../target/release/libiggy_connector_otlp_source"
+verbose = false
+
+[[streams]]
+stream = "otel"
+topic = "signals"
+schema = "json"
+batch_length = 1000
+linger_time = "5ms"
+
+[plugin_config]
+listen_addr = "0.0.0.0:4317"
+channel_capacity = 50000
+batch_size = 1000