ref(server): Introduces a new processing pipeline and implements it for logs (#4777)

Introduces a new processing pipeline and implements it for logs.
All of the log processing logic has been moved to the new pipeline.

A `Processor` is designed to build its unit of work from multiple
sources, for now it only supports a `ManagedEnvelope` as this is what we
do today. In the future it's possible to imagine a custom transport
between Relays and processor will be able to process from different
formats.

Similar to the input, the output is now also an additional layer which
can be serialized into multiple formats, for now this only supports
`ManagedEnvelope` again, but in the future, an additional serialization
will be introduced to serialize to a format which the `StoreService`
(Kafka) expects. This finally means we no longer need to needlessly
`serialize->deserialize->serialize` certain items for Kafka.

Rate limiting is now no longer envelope based, which makes it much much
easier, but has the downside that envelope based rate limiting is still
currently necessary for the fast path. We might want to explore more in
this direction in the future.

The processor also forces a correct outcome handling, through the type
system (usage of `Rejected<>`), which is much easier to use. As well as
just runtime, debug validations of outcome counts. For every change in
items must be accounted for.

The processing code itself is structured to include more strongly typed
containers, instead of an untyped envelope with a lot of implicit state.

Processing is now longer conditionally compiled, ideally all of this
should be hidden behind either normal `if` statements or a facade which
conditionally does nothing or directs to somewhere else (like the new
rate limiter). This means we can maybe finally reduce the cancer which
is `cfg(feature = "processing")`, which is only really useful for Kafka
and Redis dependencies.

Refs: INGEST-395

Author: Dav1dde

CHANGELOG.md

@@ -9,6 +9,7 @@
 
 **Internal**:
 
+- Introduces a new processing pipeline and implements it for logs. ([#4777](https://github.com/getsentry/relay/pull/4777))
 - Produce spans to the items topic. ([#4735](https://github.com/getsentry/relay/pull/4735))
 - Update opentelemetry-proto and sentry-protos dependencies. ([#4847](https://github.com/getsentry/relay/pull/4847))
 - Take into account more types of tokens when doing AI cost calculation. ([#4840](https://github.com/getsentry/relay/pull/4840))

Cargo.lock

@@ -109,7 +109,7 @@ pub fn otel_to_sentry_log(otel_log: OtelLog, received_at: DateTime<Utc>) -> Resu
     Ok(ourlog)
 }
 
-/// This fills attributes with OTel specific fields to be compatible with the otel schema.
+/// This fills attributes with OTel specific fields to be compatible with the OTel schema.
 pub fn ourlog_merge_otel(ourlog: &mut Annotated<OurLog>, received_at: DateTime<Utc>) {
     let Some(ourlog_value) = ourlog.value_mut() else {
         return;

relay-ourlogs/src/ourlog.rs

@@ -17,15 +17,16 @@ redis = ["dep:thiserror", "dep:relay-log", "relay-redis/impl"]
 workspace = true
 
 [dependencies]
+async-trait = { workspace = true }
 hashbrown = { workspace = true }
+itertools = { workspace = true }
 relay-base-schema = { workspace = true }
 relay-common = { workspace = true }
 relay-log = { workspace = true, optional = true }
 relay-redis = { workspace = true, optional = true }
 serde = { workspace = true }
 smallvec = { workspace = true }
 thiserror = { workspace = true, optional = true }
-itertools = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }
 
 [dev-dependencies]

relay-quotas/Cargo.toml

@@ -299,6 +299,14 @@ impl RateLimits {
         }
     }
 
+    /// Merges all limits from another [`RateLimits`] with this one.
+    ///
+    /// See also: [`Self::merge`].
+    pub fn merge_with(mut self, other: Self) -> Self {
+        self.merge(other);
+        self
+    }
+
     /// Returns `true` if this instance contains no active limits.
     ///
     /// This is the opposite of [`is_limited`](Self::is_limited).
@@ -468,6 +476,10 @@ impl CachedRateLimits {
     ///
     /// This is a thread-safe wrapper around [`RateLimits::merge`].
     pub fn merge(&self, limits: RateLimits) {
+        if limits.is_empty() {
+            return;
+        }
+
         let mut inner = self.0.lock().unwrap_or_else(PoisonError::into_inner);
         let current = Arc::make_mut(&mut inner);
         for limit in limits {

relay-quotas/src/rate_limit.rs

@@ -225,6 +225,7 @@ impl std::ops::Deref for RedisQuota<'_> {
 /// attachments. For more information on quota parameters, see [`Quota`].
 ///
 /// Requires the `redis` feature.
+#[derive(Clone)]
 pub struct RedisRateLimiter<T> {
     client: AsyncRedisClient,
     script: &'static Script,

relay-quotas/src/redis.rs

@@ -163,6 +163,13 @@ impl EnvelopeHeaders<PartialMeta> {
     }
 }
 
+impl<M> EnvelopeHeaders<M> {
+    /// Returns a reference to the contained meta.
+    pub fn meta(&self) -> &M {
+        &self.meta
+    }
+}
+
 #[doc(hidden)]
 #[derive(Clone, Debug)]
 pub struct Envelope {

relay-server/src/envelope/mod.rs

@@ -261,6 +261,7 @@ mod http;
 mod metrics;
 mod metrics_extraction;
 mod middlewares;
+mod processing;
 mod service;
 mod services;
 mod statsd;

relay-server/src/lib.rs

@@ -0,0 +1,98 @@
+use relay_event_schema::protocol::OurLog;
+use relay_protocol::Annotated;
+use relay_quotas::DataCategory;
+use smallvec::SmallVec;
+
+use crate::Envelope;
+use crate::envelope::Item;
+use crate::utils::EnvelopeSummary;
+
+/// A list of data categories and amounts.
+pub type Quantities = SmallVec<[(DataCategory, usize); 1]>;
+
+/// A counted item.
+///
+/// An item may represent multiple categories with different counts at once.
+pub trait Counted {
+    /// Returns the contained item quantities.
+    ///
+    /// Implementation are expected to be pure.
+    fn quantities(&self) -> Quantities;
+}
+
+impl Counted for () {
+    fn quantities(&self) -> Quantities {
+        Quantities::new()
+    }
+}
+
+impl Counted for Item {
+    fn quantities(&self) -> Quantities {
+        self.quantities()
+    }
+}
+
+impl Counted for Box<Envelope> {
+    fn quantities(&self) -> Quantities {
+        let mut quantities = Quantities::new();
+
+        // This matches the implementation of `ManagedEnvelope::reject`.
+        let summary = EnvelopeSummary::compute(self);
+        if let Some(category) = summary.event_category {
+            quantities.push((category, 1));
+            if let Some(category) = category.index_category() {
+                quantities.push((category, 1));
+            }
+        }
+
+        let data = [
+            (DataCategory::Attachment, summary.attachment_quantity),
+            (DataCategory::Profile, summary.profile_quantity),
+            (DataCategory::ProfileIndexed, summary.profile_quantity),
+            (DataCategory::Span, summary.span_quantity),
+            (DataCategory::SpanIndexed, summary.span_quantity),
+            (
+                DataCategory::Transaction,
+                summary.secondary_transaction_quantity,
+            ),
+            (DataCategory::Span, summary.secondary_span_quantity),
+            (DataCategory::Replay, summary.replay_quantity),
+            (DataCategory::ProfileChunk, summary.profile_chunk_quantity),
+            (
+                DataCategory::ProfileChunkUi,
+                summary.profile_chunk_ui_quantity,
+            ),
+            (DataCategory::LogItem, summary.log_item_quantity),
+            (DataCategory::LogByte, summary.log_byte_quantity),
+        ];
+
+        for (category, quantity) in data {
+            if quantity > 0 {
+                quantities.push((category, quantity));
+            }
+        }
+
+        quantities
+    }
+}
+
+impl Counted for Annotated<OurLog> {
+    fn quantities(&self) -> Quantities {
+        smallvec::smallvec![
+            (DataCategory::LogItem, 1),
+            (
+                DataCategory::LogByte,
+                crate::processing::logs::DUMMY_LOG_SIZE
+            )
+        ]
+    }
+}
+
+impl<T> Counted for &T
+where
+    T: Counted,
+{
+    fn quantities(&self) -> Quantities {
+        (*self).quantities()
+    }
+}