-
Notifications
You must be signed in to change notification settings - Fork 108
feat(processor): Migrate process_replay
#5580
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
29 commits
Select commit
Hold shift + click to select a range
076b0c4
boilerplate
tobias-wilfert 02bc9eb
initial logic changes (wip)
tobias-wilfert 37e35f3
update tests to send both event and recording in envelope
tobias-wilfert 89b9b11
cleanup + move code into modules
tobias-wilfert dcd9f86
remove dead code
tobias-wilfert aebb77c
appease clippy
tobias-wilfert 4c987b0
Merge branch 'master' into tobias-wilfert/feat/process_replay
tobias-wilfert d136068
cleanup
tobias-wilfert 4e73071
remove comments
tobias-wilfert 5362c0c
replace code comments with pr comments
tobias-wilfert 80fd5ad
changelog message
tobias-wilfert 5120b7e
use `for-loop` instead of `for_each`
tobias-wilfert 6241219
use `Items` instead of `Vec<Item>`
tobias-wilfert 13a1ebf
use `ContentType::OctetStream` instead of `ContentType::MsgPack`
tobias-wilfert cdc5bc4
move scrubbing after filter and quotas
tobias-wilfert 0a94211
update comments
tobias-wilfert 4223bfe
extract magic number into constant
tobias-wilfert 3ddc11e
use `#[from]` instead of `String` for errors
tobias-wilfert 666ff6b
combine `scrub` and `scrub_recording`
tobias-wilfert 9865ea7
add error logs to validate envelope content assumptions
tobias-wilfert d94b4d4
improve `ReplayError`
tobias-wilfert 5013f8a
add clarifying comment on the counting discrepancy
tobias-wilfert 8ea5bf6
add clarifying comment on the no event no recording case
tobias-wilfert a336cd6
fix up `reject_err`
tobias-wilfert 2888932
Merge branch 'master' into tobias-wilfert/feat/process_replay
tobias-wilfert fb33da6
add missing `?`
tobias-wilfert d6f60ca
Merge branch 'tobias-wilfert/feat/process_replay' of github.com:getse…
tobias-wilfert cf98063
readd old code and FF for partial rollout
tobias-wilfert 5e09667
Merge branch 'master' into tobias-wilfert/feat/process_replay
tobias-wilfert File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| use relay_dynamic_config::Feature; | ||
| use relay_statsd::metric; | ||
|
|
||
| use crate::managed::{Managed, Rejected}; | ||
| use crate::processing::Context; | ||
| use crate::processing::replays::{Error, ExpandedReplays, SerializedReplays}; | ||
| use crate::statsd::RelayCounters; | ||
|
|
||
| /// Maximum expected segment ID for a replay session, under normal operation. | ||
| const MAX_SEGMENTS_ID: u64 = 720; | ||
|
|
||
| /// Filters replays sent for a project which does not allow replay ingestion. | ||
| pub fn feature_flag( | ||
| replays: Managed<SerializedReplays>, | ||
| ctx: Context<'_>, | ||
| ) -> Result<Managed<SerializedReplays>, Rejected<Error>> { | ||
| match ctx.should_filter(Feature::SessionReplay) | ||
| || (ctx | ||
| .project_info | ||
| .has_feature(Feature::SessionReplayVideoDisabled) | ||
tobias-wilfert marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| && !replays.videos.is_empty()) | ||
| { | ||
| true => Err(replays.reject_err(Error::FilterFeatureFlag)), | ||
| false => Ok(replays), | ||
| } | ||
| } | ||
|
|
||
| /// Applies inbound filters to individual replays. | ||
| pub fn filter(replays: &mut Managed<ExpandedReplays>, ctx: Context<'_>) { | ||
| let client_addr = replays.headers.meta().client_addr(); | ||
| let event_id = replays.headers.event_id(); | ||
|
|
||
| replays.retain( | ||
| |replays| &mut replays.replays, | ||
| |replay, _| { | ||
| let event = replay.get_event().value().ok_or(Error::NoEventContent)?; | ||
|
|
||
| relay_filter::should_filter( | ||
| event, | ||
| client_addr, | ||
| &ctx.project_info.config.filter_settings, | ||
| ctx.global_config.filters(), | ||
| ) | ||
| .map_err(Error::Filtered)?; | ||
|
|
||
| // Log segments that exceed the hour limit so we can diagnose errant SDKs | ||
| // or exotic customer implementations. | ||
| if let Some(segment_id) = event.segment_id.value() | ||
| && *segment_id > MAX_SEGMENTS_ID | ||
| { | ||
| metric!(counter(RelayCounters::ReplayExceededSegmentLimit) += 1); | ||
|
|
||
| relay_log::debug!( | ||
| event_id = ?event_id, | ||
| project_id = ctx.project_info.project_id.map(|v| v.value()), | ||
| organization_id = ctx.project_info.organization_id.map(|o| o.value()), | ||
| segment_id = segment_id, | ||
| "replay segment-exceeded-limit" | ||
| ); | ||
| } | ||
|
|
||
| Ok::<_, Error>(()) | ||
| }, | ||
| ) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| use bytes::Bytes; | ||
| use smallvec::{SmallVec, smallvec}; | ||
|
|
||
| use crate::Envelope; | ||
| use crate::envelope::{ContentType, Item, ItemType, Items}; | ||
| #[cfg(feature = "processing")] | ||
| use crate::managed::ManagedEnvelope; | ||
| use crate::managed::{Managed, Rejected}; | ||
| use crate::processing::replays::{ | ||
| Error, ExpandedReplay, ExpandedReplays, ReplayVideoEvent, ReplaysOutput, | ||
| }; | ||
| use crate::processing::{self, Forward}; | ||
|
|
||
| /// Errors that can occur when serializing an expanded replay into envelope items. | ||
| #[derive(Debug, thiserror::Error)] | ||
| enum SerializeReplayError { | ||
| /// Failed to serialize the replay event to JSON. | ||
| #[error("json serialization failed")] | ||
| Json(#[from] serde_json::Error), | ||
| /// Failed to serialize the replay video event to MessagePack. | ||
| #[error("msgpack serialization failed")] | ||
| MsgPack(#[from] rmp_serde::encode::Error), | ||
| } | ||
|
|
||
| impl Forward for ReplaysOutput { | ||
| fn serialize_envelope( | ||
| self, | ||
| _ctx: processing::ForwardContext<'_>, | ||
| ) -> Result<Managed<Box<Envelope>>, Rejected<()>> { | ||
| Ok(self.0.map(|replays, records| { | ||
| let ExpandedReplays { headers, replays } = replays; | ||
| let mut items = Items::new(); | ||
|
|
||
| for replay in replays { | ||
| match serialize_replay(&replay) { | ||
| Ok(replay_items) => items.extend(replay_items), | ||
| Err(error) => { | ||
| relay_log::error!( | ||
| error = &error as &dyn std::error::Error, | ||
| event_id = ?headers.event_id(), | ||
| "failed to serialize replay" | ||
| ); | ||
| records.reject_err(Error::FailedToSerializeReplay, replay); | ||
| } | ||
| } | ||
tobias-wilfert marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| Envelope::from_parts(headers, items) | ||
| })) | ||
| } | ||
|
|
||
| #[cfg(feature = "processing")] | ||
| fn forward_store( | ||
| self, | ||
| s: processing::StoreHandle<'_>, | ||
| ctx: processing::ForwardContext<'_>, | ||
| ) -> Result<(), Rejected<()>> { | ||
| let envelope = self.serialize_envelope(ctx)?; | ||
| let envelope = ManagedEnvelope::from(envelope).into_processed(); | ||
|
|
||
| s.store(crate::services::store::StoreEnvelope { envelope }); | ||
|
|
||
| Ok(()) | ||
tobias-wilfert marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
| } | ||
|
|
||
| fn create_replay_event_item(payload: Bytes) -> Item { | ||
| let mut item = Item::new(ItemType::ReplayEvent); | ||
| item.set_payload(ContentType::Json, payload); | ||
| item | ||
| } | ||
|
|
||
| fn create_replay_recording_item(payload: Bytes) -> Item { | ||
| let mut item = Item::new(ItemType::ReplayRecording); | ||
| item.set_payload(ContentType::OctetStream, payload); | ||
| item | ||
| } | ||
|
|
||
| fn create_replay_video_item(payload: Bytes) -> Item { | ||
| let mut item = Item::new(ItemType::ReplayVideo); | ||
| item.set_payload(ContentType::OctetStream, payload); | ||
| item | ||
| } | ||
tobias-wilfert marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| fn serialize_replay(replay: &ExpandedReplay) -> Result<SmallVec<[Item; 2]>, SerializeReplayError> { | ||
| match replay { | ||
| ExpandedReplay::WebReplay { event, recording } => { | ||
| let event_bytes: Bytes = event.to_json()?.into_bytes().into(); | ||
|
|
||
| Ok(smallvec![ | ||
| create_replay_event_item(event_bytes), | ||
| create_replay_recording_item(recording.clone()), | ||
| ]) | ||
| } | ||
| ExpandedReplay::NativeReplay { | ||
| event, | ||
| recording, | ||
| video, | ||
| } => { | ||
| let event_bytes: Bytes = event.to_json()?.into_bytes().into(); | ||
|
|
||
| let video_event = ReplayVideoEvent { | ||
| replay_event: event_bytes, | ||
| replay_recording: recording.clone(), | ||
| replay_video: video.clone(), | ||
tobias-wilfert marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| }; | ||
|
|
||
| let payload = rmp_serde::to_vec_named(&video_event)?; | ||
| Ok(smallvec![create_replay_video_item(payload.into())]) | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.