feat(sinks): add iggy sink to publish OTLP to Apache Iggy#25846
feat(sinks): add iggy sink to publish OTLP to Apache Iggy#25846alexpacio wants to merge 1 commit into
iggy sink to publish OTLP to Apache Iggy#25846Conversation
Adds a new `iggy` sink that shards, encodes, and durably publishes OTLP logs, metrics, and traces (from the `opentelemetry` source configured with `use_otlp_decoding: true`) to Apache Iggy topics. The sink is self-contained: it depends only on the `iggy` client crate and existing Vector dependencies (rmp-serde, xxhash-rust), with no dependency on any downstream project. It is intended for pipelines that route observability data through Iggy as a durable, partitioned queue ahead of a storage backend, replacing a reference OpenTelemetry Collector plus a bespoke OTLP-to-Iggy adapter with a single Vector component. Gated behind the `sinks-iggy` feature flag (included in the `sinks-logs`/`sinks-metrics` meta-features and a new `iggy-integration-tests` feature).
github-actions
Bot
added
the
domain: sinks
|
Thank you for your contribution! Before we can merge this PR, please sign our Contributor License Agreement. To sign, copy and post the phrase below as a new comment on this PR.
I have read the CLA Document and I hereby sign the CLA You can retrigger this bot by commenting recheck in this Pull Request. Posted by the CLA Assistant Lite bot. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a58c9daef8
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| .or_else(|| obj.get("resourceMetrics")) | ||
| .or_else(|| obj.get("resourceSpans"))? | ||
| .as_array()?; | ||
| let first = resources.first()?.as_object()?; |
There was a problem hiding this comment.
When one decoded OTLP request contains multiple resourceLogs, resourceMetrics, or resourceSpans entries, this selects only the first resource's tenant while decode_event appends rows from every resource in the request to that same WriteBatch. Multi-tenant exporters commonly batch multiple resources together, so later resources with a different tenant_attribute will be published under the first tenant, causing cross-tenant misrouting/data leakage.
Useful? React with 👍 / 👎.
| rmpv = { version = "1.3.0", default-features = false, features = ["with-serde"], optional = true } | ||
|
|
||
| # Obstack `iggy` sink: Apache Iggy client + wire-format hashing. | ||
| iggy = { version = "=0.10.0", optional = true } |
There was a problem hiding this comment.
This adds the iggy dependency and enables it through sinks-iggy (also pulled in by the sinks meta-feature), but Cargo.lock still has no name = "iggy" entry. Any CI or release build using --locked with sinks-iggy, sinks-logs, sinks-metrics, or sinks will fail before compilation because Cargo is not allowed to update the lockfile.
Useful? React with 👍 / 👎.
| let sends = prepared | ||
| .into_iter() | ||
| .map(|(partition, mut messages)| async move { | ||
| let partitioning = Partitioning::partition_id(partition); |
There was a problem hiding this comment.
This passes the zero-based storage shard directly as the Iggy partition id, but Apache Iggy partition ids start at 1. Any batch assigned to shard 0 will be sent/flushed against partition 0, which is not a valid topic partition, so normal traffic will fail whenever a row hashes to the first shard.
Useful? React with 👍 / 👎.
| fn as_id(v: Option<&Value>) -> String { | ||
| let s = as_string(v); |
There was a problem hiding this comment.
With the opentelemetry source's OTLP decoding enabled, traceId/spanId fields are stored as raw bytes, but this path turns those bytes into a lossy UTF-8 string instead of hex. Real traces and log correlation metadata will therefore be published with invalid/non-hex IDs and hashed for shard placement using the wrong value, while the tests miss it because they build IDs as JSON strings.
Useful? React with 👍 / 👎.
| SampleRow { | ||
| fingerprint: Default::default(), | ||
| timestamp_ns: ts, |
There was a problem hiding this comment.
Every encoded sample row currently carries Fingerprint(0) even though the surrounding label set has the actual series fingerprint and the row fingerprint is serialized as part of the wire format. For metrics (and the analogous log rows), downstream storage that keys/upserts by the row's fingerprint will collapse distinct series that share a timestamp because they all arrive with the same zero fingerprint.
Useful? React with 👍 / 👎.
| let status = if result.is_ok() { | ||
| EventStatus::Delivered | ||
| } else { | ||
| EventStatus::Rejected |
There was a problem hiding this comment.
When send_messages or flush_unsaved_buffer fails because the broker/network is temporarily unavailable, this finalizes the events as Rejected, which Vector treats as a permanent failure rather than a retriable Errored status. In configurations relying on source acknowledgements or disk buffers, those transient Iggy outages will therefore drop the batch instead of allowing it to be replayed.
Useful? React with 👍 / 👎.
| as_f64(field(dp, "count")).unwrap_or(0.0), | ||
| ); | ||
| } | ||
| } else if let Some(h) = mobj.and_then(|o| o.get("exponentialHistogram")) { |
There was a problem hiding this comment.
For OTLP exponentialHistogram metrics, this branch emits only _sum and _count and ignores the positive/negative bucket arrays that carry the actual distribution. Any latency/size distribution sent as an exponential histogram will arrive downstream without buckets, making quantiles and histogram queries impossible even though the sink claims to publish OTLP metrics.
Useful? React with 👍 / 👎.
| labels.sort(); | ||
| labels.reverse(); | ||
| labels.dedup_by(|a, b| a.name == b.name); | ||
| labels.reverse(); |
There was a problem hiding this comment.
This sort/reverse/dedup sequence keeps the lexicographically largest value for duplicate label names, not the later label as the comment and call sites assume. If OTLP attributes collide after sanitization or a datapoint has a __name__ attribute, the user-provided label can replace the metric name or another intended override based solely on string ordering, routing samples under the wrong series.
Useful? React with 👍 / 👎.
1 participant
Summary
Adds a new
iggysink that publishes OTLP logs, metrics, and traces toApache Iggy topics. It's intended for
pipelines that want Iggy as a durable, partitioned queue ahead of a
storage/observability backend — pair it with the
opentelemetrysourceconfigured with
use_otlp_decoding: trueso the OTLP structure ispreserved end to end through the sink's mapping.
Gated behind a new
sinks-iggyfeature flag (included in thesinks-logs/sinks-metricsmeta-features, plus a newiggy-integration-testsfeature for future CI integration coverage).Design
logs & metrics, by trace id for spans) and encodes them into a
compact, versioned MessagePack envelope with per-envelope label
interning, before appending to the matching Iggy partition and
issuing an explicit flush for durability.
iggyclient crate (=0.10.0) andxxhash-rust(used for series fingerprinting).rmp-serdeandhexwere already optional dependencies used elsewhere in the tree.
format and OTLP→row mapping (metric name normalization, histogram
and summary explosion, severity mapping, span structure) live
entirely in
src/sinks/iggy/{proto,otlp}.rs.connection_string,stream,topic,shards,tenant/tenant_attribute(multi-tenant routing via a resource attribute),max_message_bytes,lanes(concurrent Iggy connections), andbatch_events.Testing
cargo test --features sinks-iggy sinks::iggy) coverwire encoding + shard-placement round-tripping and OTLP→row mapping
for logs (severity, trace id), metrics (monotonic sum →
_total,gauge), and the config's
generate_configround trip.vectorbinary builtwith
--features sinks-iggyagainst a local Apache Iggy broker anda downstream consumer that decodes the sink's wire format, drove
OTLP logs/metrics/traces through the
opentelemetrysource, andconfirmed all three signals were received, correctly sharded, and
queryable on the consumer side.
Notes for reviewers
1.95viarust-toolchain.toml),adding the
iggy/xxhash-rustdependencies triggered cargo torelock a large number of unrelated packages to their "latest 1.95
compatible" versions (unrelated to this change — the same relock
happens even when checking only this feature in isolation). I've
left
Cargo.lockout of this PR so CI/maintainers can regenerate itin a clean environment rather than carrying an unrelated, very large
diff.
already an
iggy-integration-testsfeature flag stubbed in forthis) if that's the preferred path for this kind of sink — let me
know the convention you'd like me to follow.