feat(attachments): Migrate standalone logic

CHANGELOG.md

@@ -25,6 +25,7 @@
 - Strip performance metric specs from extraction while keeping extraction interfaces intact. ([#5674](https://github.com/getsentry/relay/pull/5674))
 - Allow deferred lengths to the `/upload` endpoint when the sender is trusted. ([#5658](https://github.com/getsentry/relay/pull/5658))
 - Use new processor architecture to process client reports. ([#5686](https://github.com/getsentry/relay/pull/5686))
+- Use new processor architecture to process standalone attachments. ([#5703](https://github.com/getsentry/relay/pull/5703))
 - Prevent timeouts on the `/upload` endpoint. ([#5692](https://github.com/getsentry/relay/pull/5692))
 - Handle traffic bursts in the objectstore service. ([#5689](https://github.com/getsentry/relay/pull/5689))
 - Disable `fetch_materials` on GoCD `pipeline-complete` stages. ([#5697](https://github.com/getsentry/relay/pull/5697))

relay-server/src/processing/attachments/forward.rs

@@ -0,0 +1,55 @@
+use crate::Envelope;
+use crate::managed::{Managed, Rejected};
+use crate::processing::attachments::AttachmentsOutput;
+use crate::processing::{self, Forward};
+
+impl Forward for AttachmentsOutput {
+    fn serialize_envelope(
+        self,
+        _: processing::ForwardContext<'_>,
+    ) -> Result<Managed<Box<Envelope>>, Rejected<()>> {
+        let Self(attachments) = self;
+        Ok(attachments.map(|attachments, _| {
+            Envelope::from_parts(attachments.headers, attachments.attachments)
+        }))
+    }
+
+    #[cfg(feature = "processing")]
+    fn forward_store(
+        self,
+        s: processing::StoreHandle<'_>,
+        ctx: processing::ForwardContext<'_>,
+    ) -> Result<(), Rejected<()>> {
+        use crate::processing::attachments::Error;
+
+        let Self(attachments) = self;
+
+        let Some(event_id) = attachments.headers.event_id() else {
+            return Err(attachments.reject_err(Error::NoEventId).map(drop));
+        };
+
+        let use_objectstore = {
+            let options = &ctx.global_config.options;
+            crate::utils::sample(options.objectstore_attachments_sample_rate).is_keep()
+        };
+
+        for attachment in attachments.split(|attachment| attachment.attachments) {
+            let store_attachment = attachment.map(|attachment, _| {
+                use crate::services::store::StoreAttachment;
+                let quantities = attachment.quantities();
+                StoreAttachment {
+                    event_id,
+                    attachment,
+                    quantities,
+                }
+            });
+            if use_objectstore {
+                s.send_to_objectstore(store_attachment);
+            } else {
+                s.send_to_store(store_attachment);
+            }
+        }
+
+        Ok(())
+    }
+}

relay-server/src/processing/attachments/mod.rs

@@ -0,0 +1,138 @@
+use std::sync::Arc;
+
+use relay_quotas::RateLimits;
+
+use crate::envelope::{EnvelopeHeaders, Item, ItemType, Items};
+use crate::managed::{Counted, Managed, ManagedEnvelope, OutcomeError, Quantities, Rejected};
+use crate::processing::{self, CountRateLimited, Output, QuotaRateLimiter};
+#[cfg(feature = "processing")]
+use crate::services::outcome::DiscardReason;
+use crate::services::outcome::Outcome;
+use crate::statsd::RelayCounters;
+
+mod forward;
+mod process;
+
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// The Attachment was rate limited.
+    #[error("rate limited")]
+    RateLimited(RateLimits),
+
+    /// The envelope did not contain an event ID.
+    #[cfg(feature = "processing")]
+    #[error("missing event ID")]
+    NoEventId,
+}
+
+impl OutcomeError for Error {
+    type Error = Self;
+
+    fn consume(self) -> (Option<crate::services::outcome::Outcome>, Self::Error) {
+        let outcome = match &self {
+            Self::RateLimited(limits) => {
+                let reason_code = limits.longest().and_then(|limit| limit.reason_code.clone());
+                Some(Outcome::RateLimited(reason_code))
+            }
+            #[cfg(feature = "processing")]
+            Self::NoEventId => Some(Outcome::Invalid(DiscardReason::Internal)),
+        };
+        (outcome, self)
+    }
+}
+
+impl From<RateLimits> for Error {
+    fn from(value: RateLimits) -> Self {
+        Self::RateLimited(value)
+    }
+}
+
+/// A processor for Attachments.
+pub struct AttachmentProcessor {
+    limiter: Arc<QuotaRateLimiter>,
+}
+
+impl AttachmentProcessor {
+    /// Creates a new [`Self`].
+    pub fn new(limiter: Arc<QuotaRateLimiter>) -> Self {
+        Self { limiter }
+    }
+}
+
+impl processing::Processor for AttachmentProcessor {
+    type UnitOfWork = SerializedAttachments;
+    type Output = AttachmentsOutput;
+    type Error = Error;
+
+    fn prepare_envelope(
+        &self,
+        envelope: &mut ManagedEnvelope,
+    ) -> Option<Managed<Self::UnitOfWork>> {
+        debug_assert!(
+            !envelope.envelope().items().any(Item::creates_event),
+            "AttachmentProcessor should not receive items that create events"
+        );
+
+        let attachments = envelope
+            .envelope_mut()
+            .take_items_by(|i| i.requires_event() && matches!(i.ty(), ItemType::Attachment));
+
+        if attachments.is_empty() {
+            return None;
+        }
+
+        let headers = envelope.envelope().headers().clone();
+        let work = SerializedAttachments {
+            headers,
+            attachments,
+        };
+        Some(Managed::with_meta_from(envelope, work))
+    }
+
+    async fn process(
+        &self,
+        attachments: Managed<Self::UnitOfWork>,
+        ctx: processing::Context<'_>,
+    ) -> Result<processing::Output<Self::Output>, Rejected<Self::Error>> {
+        for item in &attachments.attachments {
+            let attachment_type_tag = match item.attachment_type() {
+                Some(t) => &t.to_string(),
+                None => "",
+            };
+            relay_statsd::metric!(
+                counter(RelayCounters::StandaloneItem) += 1,
+                processor = "new",
+                item_type = item.ty().name(),
+                attachment_type = attachment_type_tag,
+            );
+        }
+
+        let mut attachments = self.limiter.enforce_quotas(attachments, ctx).await?;
+        process::scrub(&mut attachments, ctx)?;
+
+        Ok(Output::just(AttachmentsOutput(attachments)))
+    }
+}
+
+/// Serialized attachments extracted from an envelope.
+#[derive(Debug)]
+pub struct SerializedAttachments {
+    /// Original envelope headers.
+    headers: EnvelopeHeaders,
+    /// A list of attachments.
+    attachments: Items,
+}
+
+impl Counted for SerializedAttachments {
+    fn quantities(&self) -> Quantities {
+        self.attachments.quantities()
+    }
+}
+
+impl CountRateLimited for Managed<SerializedAttachments> {
+    type Error = Error;
+}
+
+/// Output produced by the [`AttachmentProcessor`].
+#[derive(Debug)]
+pub struct AttachmentsOutput(Managed<SerializedAttachments>);

relay-server/src/processing/attachments/process.rs

@@ -0,0 +1,18 @@
+use crate::managed::{Managed, Rejected};
+use crate::processing::attachments::{Error, SerializedAttachments};
+use crate::processing::{self, utils};
+
+/// Runs PiiProcessors on the attachments.
+pub fn scrub(
+    attachments: &mut Managed<SerializedAttachments>,
+    ctx: processing::Context<'_>,
+) -> Result<(), Rejected<Error>> {
+    attachments.try_modify(|attachments, records| {
+        utils::attachments::scrub(
+            attachments.attachments.iter_mut(),
+            ctx.project_info,
+            Some(records),
+        );
+        Ok::<_, Error>(())
+    })
+}

relay-server/src/processing/common.rs

@@ -1,6 +1,7 @@
 use crate::Envelope;
 use crate::managed::{Managed, Rejected};
 use crate::processing::ForwardContext;
+use crate::processing::attachments::AttachmentProcessor;
 use crate::processing::check_ins::CheckInsProcessor;
 use crate::processing::errors::ErrorsProcessor;
 use crate::processing::logs::LogsProcessor;
@@ -68,4 +69,5 @@ outputs!(
     TraceAttachments => TraceAttachmentsProcessor,
     TraceMetrics => TraceMetricsProcessor,
     Replays => ReplaysProcessor,
+    Attachments => AttachmentProcessor,
 );

relay-server/src/processing/mod.rs

@@ -23,6 +23,7 @@ pub use self::common::*;
 pub use self::forward::*;
 pub use self::limits::*;
 
+pub mod attachments;
 pub mod check_ins;
 pub mod client_reports;
 pub mod errors;

relay-server/src/processing/trace_attachments/store.rs

@@ -13,15 +13,15 @@ use crate::processing::utils::store::{
     AttributeMeta, extract_client_sample_rate, extract_meta_attributes, proto_timestamp,
     quantities_to_trace_item_outcomes, uuid_to_item_id,
 };
-use crate::services::objectstore::StoreAttachment;
+use crate::services::objectstore::StoreTraceAttachment;
 use crate::services::outcome::{DiscardReason, Outcome};
 
 /// Converts an expanded attachment to a storable unit.
 pub fn convert(
     attachment: Managed<ExpandedAttachment>,
     retention: Retention,
     server_sample_rate: Option<f64>,
-) -> Result<Managed<StoreAttachment>, Rejected<()>> {
+) -> Result<Managed<StoreTraceAttachment>, Rejected<()>> {
     let scoping = attachment.scoping();
     let received_at = attachment.received_at();
     attachment.try_map(|attachment, _record_keeper| {
@@ -42,7 +42,7 @@ pub fn convert(
         let trace_item = attachment_to_trace_item(meta, quantities, ctx)
             .ok_or(Outcome::Invalid(DiscardReason::InvalidTraceAttachment))?;
 
-        Ok::<_, Outcome>(StoreAttachment { trace_item, body })
+        Ok::<_, Outcome>(StoreTraceAttachment { trace_item, body })
     })
 }