logs(replay): Base64 encode recording payload on error and log to GCP (…

…#4252)

We're seeing a lot of payloads being rejected for reasons that should
never happen. By logging the payloads we can examine these payloads and
root cause the error.

#skip-changelog

---------

Co-authored-by: David Herberth <[email protected]>

relay-server/src/services/processor/replay.rs

@@ -20,6 +20,7 @@ use crate::envelope::{ContentType, ItemType};
 use crate::services::outcome::DiscardReason;
 use crate::services::processor::{ProcessEnvelopeState, ProcessingError, ReplayGroup};
 use crate::statsd::{RelayCounters, RelayTimers};
+use crate::utils::sample;
 
 /// Removes replays if the feature flag is not enabled.
 pub fn process(
@@ -262,13 +263,55 @@ fn handle_replay_recording_item(
     })
     .map(Into::into)
     .map_err(|error| {
-        relay_log::error!(
-            error = &error as &dyn Error,
-            event_id = ?config.event_id,
-            project_id = config.project_id.map(|v| v.value()),
-            organization_id = config.organization_id.map(|o| o.value()),
-            "invalid replay recording"
-        );
+        match &error {
+            relay_replays::recording::ParseRecordingError::Compression(e) => {
+                // 20k errors per day at 0.1% sample rate == 20 logs per day
+                if sample(0.001) {
+                    relay_log::with_scope(
+                        move |scope| {
+                            scope.add_attachment(relay_log::protocol::Attachment {
+                                buffer: payload.into(),
+                                filename: "payload".to_owned(),
+                                content_type: Some("application/octet-stream".to_owned()),
+                                ty: None,
+                            });
+                        },
+                        || {
+                            relay_log::error!(
+                                error = e as &dyn Error,
+                                event_id = ?config.event_id,
+                                project_id = config.project_id.map(|v| v.value()),
+                                organization_id = config.organization_id.map(|o| o.value()),
+                                "ParseRecordingError::Compression"
+                            )
+                        },
+                    );
+                }
+            }
+            relay_replays::recording::ParseRecordingError::Message(e) => {
+                // Only 118 errors in the past 30 days. We log everything.
+                relay_log::with_scope(
+                    move |scope| {
+                        scope.add_attachment(relay_log::protocol::Attachment {
+                            buffer: payload.into(),
+                            filename: "payload".to_owned(),
+                            content_type: Some("application/octet-stream".to_owned()),
+                            ty: None,
+                        });
+                    },
+                    || {
+                        relay_log::error!(
+                            error = e,
+                            event_id = ?config.event_id,
+                            project_id = config.project_id.map(|v| v.value()),
+                            organization_id = config.organization_id.map(|o| o.value()),
+                            "ParseRecordingError::Message"
+                        )
+                    },
+                );
+            }
+            _ => (),
+        };
         ProcessingError::InvalidReplay(DiscardReason::InvalidReplayRecordingEvent)
     })
 }