diff --git a/devolutions-gateway/src/api/jrec.rs b/devolutions-gateway/src/api/jrec.rs index 508d32ff5..48f9a3776 100644 --- a/devolutions-gateway/src/api/jrec.rs +++ b/devolutions-gateway/src/api/jrec.rs @@ -31,7 +31,7 @@ use crate::api::heartbeat::recording_storage_health; use crate::artifacts::JrecArtifacts; use crate::extract::{JrecToken, RecordingDeleteScope, RecordingsReadScope}; use crate::http::{HttpError, HttpErrorBuilder}; -use crate::recording::{PushOutcome, RecordingMessageSender}; +use crate::recording::{PushOutcome, PushTarget, RecordingMessageSender}; use crate::token::{JrecTokenClaims, RecordingFileType, RecordingOperation}; /// Read chunk size when streaming a finished session ZIP from the temp file. @@ -66,6 +66,14 @@ pub fn make_router(state: DgwState) -> Router { #[serde(rename_all = "camelCase")] struct JrecPushQueryParam { file_type: RecordingFileType, + artifact: Option, +} + +/// An artifact a recorder pushes next to its recording, instead of a recording clip. +#[derive(Deserialize)] +#[serde(rename_all = "lowercase")] +enum PushArtifact { + Events, } #[derive(Deserialize)] @@ -92,6 +100,14 @@ async fn jrec_push( return Err(HttpError::forbidden().msg("expected push operation")); } + let target = match (query.artifact, query.file_type) { + (None, file_type) => PushTarget::Clip(file_type), + (Some(PushArtifact::Events), RecordingFileType::SessionRecordingLog) => PushTarget::Events, + (Some(PushArtifact::Events), _) => { + return Err(HttpError::bad_request().msg("the events artifact is a session recording log")); + } + }; + let conf = conf_handle.get_conf(); // Pre-flight: refuse the upgrade up-front when the recording storage cannot accept @@ -138,7 +154,7 @@ async fn jrec_push( recordings, shutdown_signal, claims, - query.file_type, + target, session_id, source_addr, Duration::from_secs(conf_handle.get_conf().debug.ws_keep_alive_interval), @@ -154,7 +170,7 @@ async fn handle_jrec_push( recordings: RecordingMessageSender, shutdown_signal: ShutdownSignal, claims: JrecTokenClaims, - file_type: RecordingFileType, + target: PushTarget, session_id: Uuid, source_addr: SocketAddr, keep_alive_interval: Duration, @@ -169,7 +185,7 @@ async fn handle_jrec_push( .client_stream(stream) .recordings(recordings) .claims(claims) - .file_type(file_type) + .target(target) .session_id(session_id) .shutdown_signal(shutdown_signal) .build() diff --git a/devolutions-gateway/src/artifacts.rs b/devolutions-gateway/src/artifacts.rs index a2ecd5fa8..e46c7a548 100644 --- a/devolutions-gateway/src/artifacts.rs +++ b/devolutions-gateway/src/artifacts.rs @@ -9,22 +9,25 @@ use crate::token::RecordingFileType; pub(crate) struct JrecArtifacts { #[serde(default, skip_serializing_if = "Vec::is_empty")] ai_analysis: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + events: Vec, } impl JrecArtifacts { pub(crate) fn is_empty(&self) -> bool { - let Self { ai_analysis } = self; - ai_analysis.is_empty() + let Self { ai_analysis, events } = self; + ai_analysis.is_empty() && events.is_empty() } pub(crate) fn into_file_names(self) -> impl IntoIterator { - let Self { ai_analysis } = self; - ai_analysis.into_iter().map(|artifact| artifact.file_name) + let Self { ai_analysis, events } = self; + ai_analysis.into_iter().chain(events).map(|artifact| artifact.file_name) } pub(crate) fn of_kind_mut(&mut self, kind: ArtifactKind) -> &mut Vec { match kind { ArtifactKind::AiAnalysis => &mut self.ai_analysis, + ArtifactKind::Events => &mut self.events, } } } @@ -39,18 +42,21 @@ pub(crate) struct JrecArtifact { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ArtifactKind { AiAnalysis, + /// The session events log a recorder pushes next to its recording. + Events, } impl ArtifactKind { pub const fn as_str(self) -> &'static str { match self { ArtifactKind::AiAnalysis => "ai-analysis", + ArtifactKind::Events => "events", } } pub(crate) const fn file_type(self) -> RecordingFileType { match self { - ArtifactKind::AiAnalysis => RecordingFileType::SessionRecordingLog, + ArtifactKind::AiAnalysis | ArtifactKind::Events => RecordingFileType::SessionRecordingLog, } } @@ -62,11 +68,14 @@ impl ArtifactKind { #[cfg(test)] mod tests { + use rstest::rstest; + use super::*; - #[test] - fn manifest_key_is_the_kind_name() { - let kind = ArtifactKind::AiAnalysis; + #[rstest] + #[case::ai_analysis(ArtifactKind::AiAnalysis)] + #[case::events(ArtifactKind::Events)] + fn manifest_key_is_the_kind_name(#[case] kind: ArtifactKind) { let mut artifacts = JrecArtifacts::default(); artifacts.of_kind_mut(kind).push(JrecArtifact { file_name: "file".to_owned(), diff --git a/devolutions-gateway/src/recording.intent.md b/devolutions-gateway/src/recording.intent.md index 14bb0371e..721a777ed 100644 --- a/devolutions-gateway/src/recording.intent.md +++ b/devolutions-gateway/src/recording.intent.md @@ -1,10 +1,15 @@ # Recording -Gateway recording is a continuous stream of bytes pushed from the client who pocesses the valid recording token. +Gateway recording is a continuous stream of bytes pushed from the client who possesses the valid recording token. The url is `/jet/jrec/push/{sessionId}?fileType={fileType}`. We expect the `fileType` to be one of the following file types: `webm`, `cast`, `trp` and `slog`, which must be specified. When connection is established with request of recordings for the session, if recording is not enabled within a short period of time, the connection will be closed with indication of violation of the recording policy. +Gateway only allows one recording to be pushed at a time per session. # Artifacts -Artifacts are files that are not recordings, currently only have `ai-analysis` with combination to `slog` file type. -Currently, we do not have an api endpoint to push artifacts, but we will have one in the future. \ No newline at end of file +Artifacts are files that are not recordings, listed under `artifacts` in manifest instead of `files`. +The url is `/jet/jrec/push/{sessionId}?fileType=slog&artifact=events`; `events` is the only artifact that can be pushed, as the session events log. +`ai-analysis` is not pushed: it is only kept so older manifests still list it. +Artifacts of file type slog can be pushed by producers alongside the recordings, one artifact at a time. +Pushing an artifact does not satisfy the recording policy. + diff --git a/devolutions-gateway/src/recording.rs b/devolutions-gateway/src/recording.rs index 8ece9b9b6..f806d9984 100644 --- a/devolutions-gateway/src/recording.rs +++ b/devolutions-gateway/src/recording.rs @@ -79,11 +79,20 @@ pub struct ClientPush { recordings: RecordingMessageSender, claims: JrecTokenClaims, client_stream: S, - file_type: RecordingFileType, + target: PushTarget, session_id: Uuid, shutdown_signal: ShutdownSignal, } +/// What a push writes. +#[derive(Debug, Clone, Copy)] +pub enum PushTarget { + /// A clip of the recording, listed in `files`. + Clip(RecordingFileType), + /// The session events log, an artifact the recorder pushes next to its recording. + Events, +} + impl ClientPush where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, @@ -93,7 +102,7 @@ where recordings, claims, mut client_stream, - file_type, + target, session_id, mut shutdown_signal, } = self; @@ -109,10 +118,18 @@ where } }; - let (recording_file, progress) = match recordings.connect(session_id, file_type, disconnected_ttl).await { + let connected = match target { + PushTarget::Clip(file_type) => recordings + .connect(session_id, file_type, disconnected_ttl) + .await + .map(|(path, progress)| (path, Some(progress))), + PushTarget::Events => recordings.connect_events(session_id).await.map(|path| (path, None)), + }; + + let (recording_file, progress) = match connected { Ok(connected) => connected, Err(e) => { - warn!(error = format!("{e:#}"), "Unable to start recording"); + warn!(error = format!("{e:#}"), ?target, "Unable to start recording"); client_stream.shutdown().await.context("shutdown")?; return Ok(PushOutcome::Done); } @@ -122,7 +139,15 @@ where let mut open_options = fs::OpenOptions::new(); - open_options.read(false).write(true).truncate(true).create(true); + open_options.read(false).write(true).create(true); + + // UNTESTED: a reconnecting events push appends to the session's single events log. + // Truncating it would drop every line the recorder pushed before the reconnect. + // When changing these options, keep `append` for `PushTarget::Events` and `truncate` for clips. + match target { + PushTarget::Clip(_) => open_options.truncate(true), + PushTarget::Events => open_options.append(true), + }; #[cfg(windows)] { @@ -135,7 +160,9 @@ where let res = match open_options.open(&recording_file).await { Ok(file) => { - progress.mark_recording(); + if let Some(progress) = &progress { + progress.mark_recording(); + } // Wrap WriteProgressWriter inside a BufWriter to reduce the number of flushes. let (file, flush_signal) = WriteProgressWriter::new(file); // larger buffer size to reduce the number of flushes @@ -148,7 +175,9 @@ where loop { tokio::select! { _ = flush_signal.notified() => { - progress.notify_data(); + if let Some(progress) = &progress { + progress.notify_data(); + } }, _ = shutdown_signal_clone.wait() => { break; @@ -179,7 +208,7 @@ where let _ = signal_loop.await; let flush_result = file.flush().await; - if flush_result.is_ok() { + if let (Ok(()), Some(progress)) = (&flush_result, &progress) { progress.notify_data(); } @@ -196,9 +225,15 @@ where Err(e) => Err(anyhow::Error::new(e).context(format!("failed to open file at {recording_file}"))), }; - info!(?res, "Recording finished"); + info!(?target, ?res, "Recording finished"); - recordings.disconnect(session_id).await.context("disconnect")?; + match target { + PushTarget::Clip(_) => recordings.disconnect(session_id).await.context("disconnect")?, + PushTarget::Events => recordings + .disconnect_events(session_id) + .await + .context("disconnect events")?, + } res } @@ -360,6 +395,13 @@ enum RecordingManagerMessage { Disconnect { id: Uuid, }, + ConnectEvents { + id: Uuid, + channel: oneshot::Sender>, + }, + DisconnectEvents { + id: Uuid, + }, GetState { id: Uuid, channel: oneshot::Sender>, @@ -397,6 +439,12 @@ impl fmt::Debug for RecordingManagerMessage { .field("kind", kind) .finish_non_exhaustive(), RecordingManagerMessage::Disconnect { id } => f.debug_struct("Disconnect").field("id", id).finish(), + RecordingManagerMessage::ConnectEvents { id, channel: _ } => { + f.debug_struct("ConnectEvents").field("id", id).finish_non_exhaustive() + } + RecordingManagerMessage::DisconnectEvents { id } => { + f.debug_struct("DisconnectEvents").field("id", id).finish() + } RecordingManagerMessage::GetState { id, channel: _ } => { f.debug_struct("GetState").field("id", id).finish_non_exhaustive() } @@ -467,6 +515,25 @@ impl RecordingMessageSender { .context("couldn't send Remove message") } + /// Returns the session events log to append to. + async fn connect_events(&self, id: Uuid) -> anyhow::Result { + let (tx, rx) = oneshot::channel(); + self.channel + .send(RecordingManagerMessage::ConnectEvents { id, channel: tx }) + .await + .ok() + .context("couldn't send ConnectEvents message")?; + rx.await.context("couldn't receive ConnectEvents result")? + } + + async fn disconnect_events(&self, id: Uuid) -> anyhow::Result<()> { + self.channel + .send(RecordingManagerMessage::DisconnectEvents { id }) + .await + .ok() + .context("couldn't send DisconnectEvents message") + } + pub async fn get_state(&self, id: Uuid) -> anyhow::Result> { let (tx, rx) = oneshot::channel(); self.channel @@ -573,6 +640,8 @@ impl Ord for DisconnectedTtl { pub struct RecordingManagerTask { rx: RecordingMessageReceiver, ongoing_recordings: HashMap, + /// Sessions whose recorder is pushing its events log. + events_pushes: HashSet, recordings_path: Utf8PathBuf, session_manager_handle: SessionMessageSender, job_queue_handle: JobQueueHandle, @@ -588,6 +657,7 @@ impl RecordingManagerTask { Self { rx, ongoing_recordings: HashMap::new(), + events_pushes: HashSet::new(), recordings_path, session_manager_handle, job_queue_handle, @@ -819,6 +889,45 @@ impl RecordingManagerTask { Ok(artifact_path) } + fn handle_connect_events(&mut self, id: Uuid) -> anyhow::Result { + if self.events_pushes.contains(&id) { + anyhow::bail!("concurrent events push for the same session is not supported"); + } + + let recording_path = self.recordings_path.join(id.to_string()); + let manifest_path = recording_path.join("recording.json"); + + // The session folder and manifest belong to the recording, so the events push never creates them. + if !manifest_path.exists() { + anyhow::bail!("an events push needs the recording of the session to start first"); + } + + let mut manifest = JrecManifest::read_from_file(&manifest_path).context("read manifest from disk")?; + + let events = manifest.artifacts.of_kind_mut(ArtifactKind::Events); + let file_name = match events.last() { + Some(log) => log.file_name.clone(), + None => { + let file_name = ArtifactKind::Events.file_name(0); + events.push(JrecArtifact { + file_name: file_name.clone(), + }); + manifest + .save_to_file(&manifest_path) + .context("write manifest to disk")?; + file_name + } + }; + + self.events_pushes.insert(id); + + Ok(recording_path.join(file_name)) + } + + fn handle_disconnect_events(&mut self, id: Uuid) { + self.events_pushes.remove(&id); + } + fn handle_remove(&mut self, id: Uuid) { if let Some(ongoing) = self.ongoing_recordings.get(&id) { match ongoing.state { @@ -960,6 +1069,12 @@ async fn recording_manager_task( } } } + RecordingManagerMessage::ConnectEvents { id, channel } => { + let _ = channel.send(manager.handle_connect_events(id)); + } + RecordingManagerMessage::DisconnectEvents { id } => { + manager.handle_disconnect_events(id); + } RecordingManagerMessage::GetState { id, channel } => { let response = manager.ongoing_recordings.get(&id).map(|ongoing| ongoing.state.clone()); let _ = channel.send(response); @@ -1008,11 +1123,15 @@ async fn recording_manager_task( }; debug!(?msg, "Received message"); - if let RecordingManagerMessage::Disconnect { id } = msg { - if let Err(e) = manager.handle_disconnect(id).await { - error!(error = format!("{e:#}"), "handle_disconnect"); + match msg { + RecordingManagerMessage::Disconnect { id } => { + if let Err(e) = manager.handle_disconnect(id).await { + error!(error = format!("{e:#}"), "handle_disconnect"); + } + manager.ongoing_recordings.remove(&id); } - manager.ongoing_recordings.remove(&id); + RecordingManagerMessage::DisconnectEvents { id } => manager.handle_disconnect_events(id), + _ => {} } } @@ -1179,6 +1298,21 @@ mod tests { fn read_file(&self, id: Uuid, file_name: &str) -> String { std::fs::read_to_string(self.recordings_path.join(id.to_string()).join(file_name)).expect("read file") } + + async fn try_connect_events(&self, id: Uuid) -> anyhow::Result { + let path = self.sender.connect_events(id).await?; + Ok(path.file_name().expect("file name").to_owned()) + } + + async fn connect_events(&self, id: Uuid) -> String { + self.try_connect_events(id).await.expect("connect events") + } + + async fn disconnect_events(&self, id: Uuid) { + self.sender.disconnect_events(id).await.expect("disconnect events"); + // Messages are processed in order, so this waits for the disconnect to be handled. + self.sender.get_count().await.expect("sync with manager"); + } } #[tokio::test] @@ -1421,6 +1555,84 @@ mod tests { assert_eq!(harness.sender.get_count().await.expect("count"), 1); } + #[tokio::test] + async fn events_log_goes_to_artifacts() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + let clip = harness.connect(id, WEBM).await; + let log = harness.connect_events(id).await; + + assert_eq!(clip, "recording-0.webm"); + assert_eq!(log, "events-0.slog"); + let manifest = harness.read_manifest(id); + let files = manifest["files"].as_array().expect("files"); + assert_eq!(files.len(), 1); + assert_eq!(files[0]["fileName"], "recording-0.webm"); + assert_eq!( + manifest["artifacts"], + json!({ "events": [{ "fileName": "events-0.slog" }] }) + ); + } + + #[tokio::test] + async fn an_events_push_before_the_recording_is_refused() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + assert!(harness.try_connect_events(id).await.is_err()); + assert!(!harness.recordings_path.join(id.to_string()).exists()); + } + + #[tokio::test] + async fn a_second_events_push_is_refused() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + harness.connect(id, WEBM).await; + harness.connect_events(id).await; + + assert!(harness.try_connect_events(id).await.is_err()); + } + + #[tokio::test] + async fn a_reconnecting_events_push_continues_the_same_log() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + harness.connect(id, WEBM).await; + let first = harness.connect_events(id).await; + harness.disconnect_events(id).await; + let second = harness.connect_events(id).await; + + assert_eq!(first, second); + assert_eq!( + harness.read_manifest(id)["artifacts"], + json!({ "events": [{ "fileName": "events-0.slog" }] }) + ); + } + + #[tokio::test] + async fn an_events_push_never_keeps_the_recording_alive() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + harness.sender.connect(id, WEBM, Duration::ZERO).await.expect("connect"); + harness.connect_events(id).await; + assert!(harness.sender.active_recordings.contains(id)); + assert_eq!(harness.sender.get_count().await.expect("count"), 1); + + harness.disconnect(id).await; + tokio::time::timeout(Duration::from_secs(5), async { + while harness.sender.get_state(id).await.expect("get state").is_some() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the recording ends"); + assert!(!harness.sender.active_recordings.contains(id)); + } + #[test] fn manifest_without_artifacts_round_trips_byte_for_byte() { let dir = tempfile::tempdir().expect("temp dir");