relay_server/managed/
envelope.rs

1use std::fmt::{Debug, Display};
2use std::marker::PhantomData;
3use std::mem::size_of;
4use std::ops::{Deref, DerefMut};
5use std::time::Duration;
6
7use chrono::{DateTime, Utc};
8use relay_quotas::{DataCategory, Scoping};
9use relay_system::Addr;
10
11use crate::envelope::{Envelope, Item};
12use crate::extractors::RequestMeta;
13use crate::managed::Counted as _;
14use crate::services::outcome::{DiscardReason, Outcome, TrackOutcome};
15use crate::services::processor::{Processed, ProcessingGroup};
16use crate::statsd::{RelayCounters, RelayTimers};
17use crate::utils::EnvelopeSummary;
18
19/// Denotes the success of handling an envelope.
20#[derive(Clone, Copy, Debug)]
21enum Handling {
22    /// The envelope was handled successfully.
23    ///
24    /// This can be the case even if the envelpoe was dropped. For example, if a rate limit is in
25    /// effect or if the corresponding project is disabled.
26    Success,
27    /// Handling the envelope failed due to an error or bug.
28    Failure,
29}
30
31impl Handling {
32    fn from_outcome(outcome: &Outcome) -> Self {
33        if outcome.is_unexpected() {
34            Self::Failure
35        } else {
36            Self::Success
37        }
38    }
39
40    fn as_str(&self) -> &str {
41        match self {
42            Handling::Success => "success",
43            Handling::Failure => "failure",
44        }
45    }
46}
47
48/// Represents the decision on whether or not to keep an envelope item.
49#[derive(Debug, Clone)]
50pub enum ItemAction {
51    /// Keep the item.
52    Keep,
53    /// Drop the item and log an outcome for it.
54    Drop(Outcome),
55    /// Drop the item without logging an outcome.
56    DropSilently,
57}
58
59#[derive(Debug)]
60struct EnvelopeContext {
61    summary: EnvelopeSummary,
62    scoping: Scoping,
63    partition_key: Option<u32>,
64    done: bool,
65}
66
67/// Error emitted when converting a [`ManagedEnvelope`] and a processing group into a [`TypedEnvelope`].
68#[derive(Debug)]
69pub struct InvalidProcessingGroupType(pub ManagedEnvelope, pub ProcessingGroup);
70
71impl Display for InvalidProcessingGroupType {
72    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
73        f.write_fmt(format_args!(
74            "failed to convert to the processing group {} based on the provided type",
75            self.1.variant()
76        ))
77    }
78}
79
80impl std::error::Error for InvalidProcessingGroupType {}
81
82/// A wrapper for [`ManagedEnvelope`] with assigned processing group type.
83pub struct TypedEnvelope<G>(ManagedEnvelope, PhantomData<G>);
84
85impl<G> TypedEnvelope<G> {
86    /// Changes the typed of the current envelope to processed.
87    ///
88    /// Once it's marked processed it can be submitted to upstream.
89    pub fn into_processed(self) -> TypedEnvelope<Processed> {
90        TypedEnvelope::new(self.0)
91    }
92
93    /// Accepts the envelope and drops the internal managed envelope with its context.
94    ///
95    /// This should be called if the envelope has been accepted by the upstream, which means that
96    /// the responsibility for logging outcomes has been moved. This function will not log any
97    /// outcomes.
98    pub fn accept(self) {
99        self.0.accept()
100    }
101
102    /// Creates a new typed envelope.
103    ///
104    /// Note: this method is private to make sure that only `TryFrom` implementation is used, which
105    /// requires the check for the error if conversion is failing.
106    fn new(managed_envelope: ManagedEnvelope) -> Self {
107        Self(managed_envelope, Default::default())
108    }
109}
110
111impl<G: TryFrom<ProcessingGroup>> TryFrom<(ManagedEnvelope, ProcessingGroup)> for TypedEnvelope<G> {
112    type Error = InvalidProcessingGroupType;
113    fn try_from(
114        (envelope, group): (ManagedEnvelope, ProcessingGroup),
115    ) -> Result<Self, Self::Error> {
116        match <ProcessingGroup as TryInto<G>>::try_into(group) {
117            Ok(_) => Ok(TypedEnvelope::new(envelope)),
118            Err(_) => Err(InvalidProcessingGroupType(envelope, group)),
119        }
120    }
121}
122
123impl<G> Debug for TypedEnvelope<G> {
124    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
125        f.debug_tuple("TypedEnvelope").field(&self.0).finish()
126    }
127}
128
129impl<G> Deref for TypedEnvelope<G> {
130    type Target = ManagedEnvelope;
131
132    fn deref(&self) -> &Self::Target {
133        &self.0
134    }
135}
136
137impl<G> DerefMut for TypedEnvelope<G> {
138    fn deref_mut(&mut self) -> &mut Self::Target {
139        &mut self.0
140    }
141}
142
143/// Tracks the lifetime of an [`Envelope`] in Relay.
144///
145/// The managed envelope accompanies envelopes through the processing pipeline in Relay and ensures
146/// that outcomes are recorded when the Envelope is dropped. They can be dropped in one of three
147/// ways:
148///
149///  - By calling [`accept`](Self::accept). Responsibility of the envelope has been transferred to
150///    another service, and no further outcomes need to be recorded.
151///  - By calling [`reject`](Self::reject). The entire envelope was dropped, and the outcome
152///    specifies the reason.
153///  - By dropping the managed envelope. This indicates an issue or a bug and raises the
154///    `"internal"` outcome. There should be additional error handling to report an error to Sentry.
155///
156/// The managed envelope also holds a processing queue permit which is used for backpressure
157/// management. It is automatically reclaimed when the context is dropped along with the envelope.
158#[derive(Debug)]
159pub struct ManagedEnvelope {
160    envelope: Box<Envelope>,
161    context: EnvelopeContext,
162    outcome_aggregator: Addr<TrackOutcome>,
163}
164
165impl ManagedEnvelope {
166    /// Computes a managed envelope from the given envelope and binds it to the processing queue.
167    ///
168    /// To provide additional scoping, use [`ManagedEnvelope::scope`].
169    pub fn new(envelope: Box<Envelope>, outcome_aggregator: Addr<TrackOutcome>) -> Self {
170        let meta = &envelope.meta();
171        let summary = EnvelopeSummary::compute(envelope.as_ref());
172        let scoping = meta.get_partial_scoping().into_scoping();
173
174        Self {
175            envelope,
176            context: EnvelopeContext {
177                summary,
178                scoping,
179                partition_key: None,
180                done: false,
181            },
182            outcome_aggregator,
183        }
184    }
185
186    /// An untracked instance which does not emit outcomes, useful for testing.
187    #[cfg(test)]
188    pub fn untracked(envelope: Box<Envelope>, outcome_aggregator: Addr<TrackOutcome>) -> Self {
189        let mut envelope = Self::new(envelope, outcome_aggregator);
190        envelope.context.done = true;
191        envelope
192    }
193
194    /// Returns a reference to the contained [`Envelope`].
195    pub fn envelope(&self) -> &Envelope {
196        self.envelope.as_ref()
197    }
198
199    /// Returns a mutable reference to the contained [`Envelope`].
200    pub fn envelope_mut(&mut self) -> &mut Envelope {
201        self.envelope.as_mut()
202    }
203
204    /// Consumes itself returning the managed envelope.
205    pub fn into_envelope(mut self) -> Box<Envelope> {
206        self.context.done = true;
207        self.take_envelope()
208    }
209
210    /// Converts current managed envelope into processed envelope.
211    ///
212    /// Once it's marked processed it can be submitted to upstream.
213    pub fn into_processed(self) -> TypedEnvelope<Processed> {
214        TypedEnvelope::new(self)
215    }
216
217    /// Take the envelope out of the context and replace it with a dummy.
218    ///
219    /// Note that after taking out the envelope, the envelope summary is incorrect.
220    pub(crate) fn take_envelope(&mut self) -> Box<Envelope> {
221        Box::new(self.envelope.take_items())
222    }
223
224    /// Update the context with envelope information.
225    ///
226    /// This updates the item summary as well as the event id.
227    pub fn update(&mut self) -> &mut Self {
228        self.context.summary = EnvelopeSummary::compute(self.envelope());
229        self
230    }
231
232    /// Retains or drops items based on the [`ItemAction`].
233    ///
234    ///
235    /// This method operates in place and preserves the order of the retained items.
236    pub fn retain_items<F>(&mut self, mut f: F)
237    where
238        F: FnMut(&mut Item) -> ItemAction,
239    {
240        let mut outcomes = Vec::new();
241        self.envelope.retain_items(|item| match f(item) {
242            ItemAction::Keep => true,
243            ItemAction::DropSilently => false,
244            ItemAction::Drop(outcome) => {
245                for (category, quantity) in item.quantities() {
246                    outcomes.push((outcome.clone(), category, quantity));
247                }
248
249                false
250            }
251        });
252        for (outcome, category, quantity) in outcomes {
253            self.track_outcome(outcome, category, quantity);
254        }
255        // TODO: once `update` is private, it should be called here.
256    }
257
258    /// Drops every item in the envelope.
259    pub fn drop_items_silently(&mut self) {
260        self.envelope.drop_items_silently();
261    }
262
263    /// Re-scopes this context to the given scoping.
264    pub fn scope(&mut self, scoping: Scoping) -> &mut Self {
265        self.context.scoping = scoping;
266        self
267    }
268
269    /// Removes event item(s) and logs an outcome.
270    ///
271    /// Note: This function relies on the envelope summary being correct.
272    pub fn reject_event(&mut self, outcome: Outcome) {
273        if let Some(event_category) = self.event_category() {
274            self.envelope.retain_items(|item| !item.creates_event());
275            if let Some(indexed) = event_category.index_category() {
276                self.track_outcome(outcome.clone(), indexed, 1);
277            }
278            self.track_outcome(outcome, event_category, 1);
279        }
280    }
281
282    /// Records an outcome scoped to this envelope's context.
283    ///
284    /// This managed envelope should be updated using [`update`](Self::update) soon after this
285    /// operation to ensure that subsequent outcomes are consistent.
286    pub fn track_outcome(&self, outcome: Outcome, category: DataCategory, quantity: usize) {
287        self.outcome_aggregator.send(TrackOutcome {
288            timestamp: self.received_at(),
289            scoping: self.context.scoping,
290            outcome,
291            event_id: self.envelope.event_id(),
292            remote_addr: self.meta().remote_addr(),
293            category,
294            // Quantities are usually `usize` which lets us go all the way to 64-bit on our
295            // machines, but the protocol and data store can only do 32-bit.
296            quantity: quantity as u32,
297        });
298    }
299
300    /// Accepts the envelope and drops the context.
301    ///
302    /// This should be called if the envelope has been accepted by the upstream, which means that
303    /// the responsibility for logging outcomes has been moved. This function will not log any
304    /// outcomes.
305    pub fn accept(mut self) {
306        if !self.context.done {
307            self.finish(RelayCounters::EnvelopeAccepted, Handling::Success);
308        }
309    }
310
311    /// Returns the data category of the event item in the envelope.
312    fn event_category(&self) -> Option<DataCategory> {
313        self.context.summary.event_category
314    }
315
316    /// Records rejection outcomes for all items stored in this context.
317    ///
318    /// This does not send outcomes for empty envelopes or request-only contexts.
319    pub fn reject(&mut self, outcome: Outcome) {
320        if self.context.done {
321            return;
322        }
323
324        // Errors are only logged for what we consider failed request handling. In other cases, we
325        // "expect" errors and log them as debug level.
326        let handling = Handling::from_outcome(&outcome);
327        match handling {
328            Handling::Success => relay_log::debug!("dropped envelope: {outcome}"),
329            Handling::Failure => {
330                let summary = &self.context.summary;
331
332                relay_log::error!(
333                    tags.project_key = self.scoping().project_key.to_string(),
334                    tags.has_attachments = summary.attachment_quantities.bytes() > 0,
335                    tags.has_sessions = summary.session_quantity > 0,
336                    tags.has_profiles = summary.profile_quantity > 0,
337                    tags.has_transactions = summary.secondary_transaction_quantity > 0,
338                    tags.has_span_metrics = summary.secondary_span_quantity > 0,
339                    tags.has_replays = summary.replay_quantity > 0,
340                    tags.has_user_reports = summary.user_report_quantity > 0,
341                    tags.has_trace_metrics = summary.trace_metric_quantity > 0,
342                    tags.has_checkins = summary.monitor_quantity > 0,
343                    tags.event_category = ?summary.event_category,
344                    cached_summary = ?summary,
345                    recomputed_summary = ?EnvelopeSummary::compute(self.envelope()),
346                    "dropped envelope: {outcome}"
347                );
348            }
349        }
350
351        for (category, quantity) in self.context.summary.quantities() {
352            self.track_outcome(outcome.clone(), category, quantity);
353        }
354
355        self.finish(RelayCounters::EnvelopeRejected, handling);
356    }
357
358    /// Returns scoping stored in this context.
359    pub fn scoping(&self) -> Scoping {
360        self.context.scoping
361    }
362
363    /// Returns the partition key, which is set on upstream requests in the `X-Sentry-Relay-Shard` header.
364    pub fn partition_key(&self) -> Option<u32> {
365        self.context.partition_key
366    }
367
368    /// Sets a new [`Self::partition_key`].
369    pub fn set_partition_key(&mut self, partition_key: Option<u32>) -> &mut Self {
370        self.context.partition_key = partition_key;
371        self
372    }
373
374    /// Returns the contained original request meta.
375    pub fn meta(&self) -> &RequestMeta {
376        self.envelope().meta()
377    }
378
379    /// Returns estimated size of this envelope.
380    ///
381    /// This is just an estimated size, which in reality can be somewhat bigger, depending on the
382    /// list of additional attributes allocated on all of the inner types.
383    ///
384    /// NOTE: Current implementation counts in only the size of the items payload and stack
385    /// allocated parts of [`Envelope`] and [`ManagedEnvelope`]. All the heap allocated fields
386    /// within early mentioned types are skipped.
387    pub fn estimated_size(&self) -> usize {
388        // Always round it up to next 1KB.
389        (f64::ceil(
390            (self.context.summary.payload_size + size_of::<Self>() + size_of::<Envelope>()) as f64
391                / 1000.,
392        ) * 1000.) as usize
393    }
394
395    /// Returns the time at which the envelope was received at this Relay.
396    ///
397    /// This is the date time equivalent to [`start_time`](Self::received_at).
398    pub fn received_at(&self) -> DateTime<Utc> {
399        self.envelope.received_at()
400    }
401
402    /// Returns the time elapsed in seconds since the envelope was received by this Relay.
403    ///
404    /// In case the elapsed time is negative, it is assumed that no time elapsed.
405    pub fn age(&self) -> Duration {
406        self.envelope.age()
407    }
408
409    /// Escape hatch for the [`super::Managed`] type, to make it possible to construct
410    /// from a managed envelope.
411    pub(super) fn outcome_aggregator(&self) -> &Addr<TrackOutcome> {
412        &self.outcome_aggregator
413    }
414
415    /// Resets inner state to ensure there's no more logging.
416    fn finish(&mut self, counter: RelayCounters, handling: Handling) {
417        relay_statsd::metric!(counter(counter) += 1, handling = handling.as_str());
418        relay_statsd::metric!(timer(RelayTimers::EnvelopeTotalTime) = self.age());
419
420        self.context.done = true;
421    }
422}
423
424impl Drop for ManagedEnvelope {
425    fn drop(&mut self) {
426        self.reject(Outcome::Invalid(DiscardReason::Internal));
427    }
428}
429
430impl<G> From<TypedEnvelope<G>> for ManagedEnvelope {
431    fn from(value: TypedEnvelope<G>) -> Self {
432        value.0
433    }
434}
435
436#[cfg(test)]
437mod tests {
438    use super::*;
439    use bytes::Bytes;
440
441    #[test]
442    fn span_metrics_are_reported() {
443        let bytes =
444            Bytes::from(r#"{"dsn":"https://e12d836b15bb49d7bbf99e64295d995b:@sentry.io/42"}"#);
445        let envelope = Envelope::parse_bytes(bytes).unwrap();
446
447        let (outcome_aggregator, mut rx) = Addr::custom();
448        let mut env = ManagedEnvelope::new(envelope, outcome_aggregator);
449        env.context.summary.span_quantity = 123;
450        env.context.summary.secondary_span_quantity = 456;
451
452        env.reject(Outcome::Abuse);
453
454        rx.close();
455
456        let outcome = rx.blocking_recv().unwrap();
457        assert_eq!(outcome.category, DataCategory::Span);
458        assert_eq!(outcome.quantity, 123);
459        assert_eq!(outcome.outcome, Outcome::Abuse);
460
461        let outcome = rx.blocking_recv().unwrap();
462        assert_eq!(outcome.category, DataCategory::SpanIndexed);
463        assert_eq!(outcome.quantity, 123);
464        assert_eq!(outcome.outcome, Outcome::Abuse);
465
466        let outcome = rx.blocking_recv().unwrap();
467        assert_eq!(outcome.category, DataCategory::Span);
468        assert_eq!(outcome.quantity, 456);
469        assert_eq!(outcome.outcome, Outcome::Abuse);
470
471        assert!(rx.blocking_recv().is_none());
472    }
473}
relay_server/managed/envelope.rs

relay_server/managed/
envelope.rs
