feat(eap): Produce logs to the items topic (#4707)

Author: phacops

CHANGELOG.md

@@ -20,6 +20,7 @@
 
 - Remove threads with 1 non-idle sample in profiling chunks. ([#4694](https://github.com/getsentry/relay/pull/4694))
 - Migrate all Rust workspace crates to edition 2024. ([#4705](https://github.com/getsentry/relay/pull/4705))
+- Produce logs to the items topic. ([#4707](https://github.com/getsentry/relay/pull/4707))
 
 ## 25.4.0
 

Cargo.lock

@@ -43,6 +43,7 @@ relay-common = { path = "relay-common" }
 relay-config = { path = "relay-config" }
 relay-crash = { path = "relay-crash" }
 relay-dynamic-config = { path = "relay-dynamic-config" }
+relay-event-derive = { path = "relay-event-derive" }
 relay-event-normalization = { path = "relay-event-normalization" }
 relay-event-schema = { path = "relay-event-schema" }
 relay-ffi = { path = "relay-ffi" }
@@ -58,6 +59,7 @@ relay-pii = { path = "relay-pii" }
 relay-profiling = { path = "relay-profiling" }
 relay-prosperoconv = { path = "relay-prosperoconv" }
 relay-protocol = { path = "relay-protocol" }
+relay-protocol-derive = { path = "relay-protocol-derive" }
 relay-quotas = { path = "relay-quotas" }
 relay-redis = { path = "relay-redis" }
 relay-replays = { path = "relay-replays" }
@@ -66,23 +68,21 @@ relay-server = { path = "relay-server" }
 relay-spans = { path = "relay-spans" }
 relay-statsd = { path = "relay-statsd" }
 relay-system = { path = "relay-system" }
+relay-test = { path = "relay-test" }
 relay-threading = { path = "relay-threading" }
 relay-ua = { path = "relay-ua" }
-relay-test = { path = "relay-test" }
-relay-protocol-derive = { path = "relay-protocol-derive" }
-relay-event-derive = { path = "relay-event-derive" }
 
 ahash = "0.8.11"
 android_trace_log = { version = "0.3.0", features = ["serde"] }
 # This version is pinned because upgrading it enables backtrace by default without the possibility to disable it
 # which will increase processing time for transactions massively.
 # Keep it pinned until it's possible to disable backtrace.
 anyhow = "=1.0.69"
+arc-swap = "1.7.1"
+async-trait = "0.1.83"
 axum = "0.8.1"
 axum-extra = "0.10.0"
 axum-server = "0.7.1"
-arc-swap = "1.7.1"
-async-trait = "0.1.83"
 backoff = "0.4.0"
 bindgen = "0.71.1"
 brotli = "6.0.0"
@@ -91,8 +91,8 @@ bytes = "1.10.1"
 bzip2 = "0.4.4"
 cadence = "1.5.0"
 chrono = { version = "0.4.38", default-features = false, features = [
-    "std",
-    "serde",
+  "std",
+  "serde",
 ] }
 clap = { version = "4.4.18" }
 clap_complete = "4.1.6"
@@ -110,19 +110,19 @@ dynfmt = "0.1.5"
 ed25519-dalek = "2.1.1"
 enumset = "1.0.13"
 flate2 = "1.0.35"
+flume = { version = "0.11.1", default-features = false }
 fnv = "1.0.7"
 futures = { version = "0.3", default-features = false, features = ["std"] }
-flume = { version = "0.11.1", default-features = false }
 globset = "0.4.15"
 hash32 = "0.3.1"
 hashbrown = "0.14.5"
 hex = "0.4.3"
 hmac = "0.12.1"
 hostname = "0.4.0"
-human-size = "0.4.3"
 http = "1.1.0"
-hyper-util = { version = "0.1.10", features = ["tokio"] }
+human-size = "0.4.3"
 hyper = "1.5.1"
+hyper-util = { version = "0.1.10", features = ["tokio"] }
 indexmap = "2.2.6"
 insta = { version = "1.31.0", features = ["json", "redactions", "ron"] }
 ipnetwork = "0.20.0"
@@ -132,8 +132,8 @@ libc = "0.2.167"
 liblzma = "0.3.5"
 lru = "0.12.5"
 maxminddb = "0.24.0"
-memchr = "2.7.4"
 md5 = "0.7.0"
+memchr = "2.7.4"
 mime = "0.3.17"
 minidump = "0.22.2"
 multer = "3.1.0"
@@ -151,6 +151,7 @@ pretty-hex = "0.4.1"
 priority-queue = "2.0.3"
 proc-macro2 = "1.0.92"
 prost = "0.13.3"
+prost-types = "0.13.3"
 psl = "2.1.65"
 quote = "1.0.37"
 rand = "0.8.5"
@@ -162,12 +163,13 @@ regex = "1.11.1"
 regex-lite = "0.1.6"
 reqwest = "0.12.9"
 rmp-serde = "1.3.0"
+semver = "1.0.23"
 sentry = "0.36.0"
 sentry-core = "0.36.0"
-sentry-kafka-schemas = { version = "0.1.129", default-features = false }
+sentry-kafka-schemas = { version = "1.3.2", default-features = false }
 sentry-release-parser = { version = "1.3.2", default-features = false }
 sentry-types = "0.36.0"
-semver = "1.0.23"
+sentry_protos = "0.2.0"
 serde = { version = "1.0.215", features = ["derive", "rc"] }
 serde-transcode = "1.1.1"
 serde_bytes = "0.11"
@@ -205,6 +207,6 @@ unescaper = "0.1.5"
 unicase = "2.8.0"
 url = "2.5.4"
 utf16string = "0.2.0"
-uuid = { version = "1.11.0", features = ["serde", "v4"] }
+uuid = { version = "1.11.0", features = ["serde", "v4", "v7"] }
 walkdir = "2.5.0"
 zstd = { version = "0.13.2", features = ["experimental"] }

Cargo.toml

@@ -21,11 +21,11 @@ rmp-serde = { workspace = true, optional = true }
 serde = { workspace = true }
 serde_json = { workspace = true, optional = true }
 thiserror = { workspace = true }
-sentry-kafka-schemas = { workspace = true, default_features = false, optional = true }
+sentry-kafka-schemas = { workspace = true, default-features = false, optional = true }
 
 [dev-dependencies]
 serde_yaml = { workspace = true }
-sentry-kafka-schemas = { workspace = true }
+sentry-kafka-schemas = { workspace = true, default-features = false }
 
 [features]
 default = []

relay-kafka/Cargo.toml

@@ -51,14 +51,16 @@ pub enum KafkaTopic {
     Spans,
     /// Feedback events topic.
     Feedback,
+    /// Items topic
+    Items,
 }
 
 impl KafkaTopic {
     /// Returns iterator over the variants of [`KafkaTopic`].
     /// It will have to be adjusted if the new variants are added.
     pub fn iter() -> std::slice::Iter<'static, Self> {
         use KafkaTopic::*;
-        static TOPICS: [KafkaTopic; 14] = [
+        static TOPICS: [KafkaTopic; 15] = [
             Events,
             Attachments,
             Transactions,
@@ -73,6 +75,7 @@ impl KafkaTopic {
             OurLogs,
             Spans,
             Feedback,
+            Items,
         ];
         TOPICS.iter()
     }
@@ -138,6 +141,7 @@ define_topic_assignments! {
     monitors: (KafkaTopic::Monitors, "ingest-monitors", "Monitor check-ins."),
     spans: (KafkaTopic::Spans, "snuba-spans", "Standalone spans without a transaction."),
     feedback: (KafkaTopic::Feedback, "ingest-feedback-events", "Feedback events topic."),
+    items: (KafkaTopic::Items, "snuba-items", "Items topic."),
 }
 
 /// Configuration for a "logical" topic/datasink that Relay should forward data into.

relay-kafka/src/config.rs

@@ -64,6 +64,11 @@ pub enum ClientError {
     /// Failed to validate the topic.
     #[error("failed to validate the topic with name {0}: {1:?}")]
     TopicError(String, rdkafka_sys::rd_kafka_resp_err_t),
+
+    /// Failed to encode the protobuf into the buffer
+    /// because the buffer is too small.
+    #[error("failed to encode protobuf because the buffer is too small")]
+    ProtobufEncodingFailed,
 }
 
 /// Describes the type which can be sent using kafka producer provided by this crate.