[Feat] follow logs (#1714)

* tail logs * add cli * add FE * abstract http to shared * batch new logs * file download for logs * fix modal error when no config Co-authored-by: Chris Guida <chrisguida@users.noreply.github.com> Co-authored-by: Aiden McClelland <me@drbonez.dev> Co-authored-by: Matt Hill <matthewonthemoon@gmail.com> Co-authored-by: BluJ <mogulslayer@gmail.com>
2026-04-01 21:13:09 +00:00 · 2022-08-03 13:06:25 -05:00
parent c44eb3a2c3
commit 2f8d825970
70 changed files with 2202 additions and 1795 deletions
--- a/backend/src/logs.rs
+++ b/backend/src/logs.rs
@@ -1,22 +1,117 @@
+use std::future::Future;
+use std::marker::PhantomData;
+use std::ops::Deref;
+use std::ops::DerefMut;
 use std::process::Stdio;
 use std::time::{Duration, UNIX_EPOCH};

 use chrono::{DateTime, Utc};
-use clap::ArgMatches;
 use color_eyre::eyre::eyre;
-use futures::TryStreamExt;
+use futures::stream::BoxStream;
+use futures::Stream;
+use futures::{FutureExt, SinkExt, StreamExt, TryStreamExt};
+use hyper::upgrade::Upgraded;
+use hyper::Error as HyperError;
 use rpc_toolkit::command;
+use rpc_toolkit::yajrc::RpcError;
 use serde::{Deserialize, Serialize};
 use tokio::io::{AsyncBufReadExt, BufReader};
-use tokio::process::Command;
+use tokio::process::{Child, Command};
+use tokio::task::JoinError;
 use tokio_stream::wrappers::LinesStream;
+use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
+use tokio_tungstenite::tungstenite::protocol::CloseFrame;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::WebSocketStream;
 use tracing::instrument;

+use crate::context::{CliContext, RpcContext};
+use crate::core::rpc_continuations::{RequestGuid, RpcContinuation};
 use crate::error::ResultExt;
 use crate::procedure::docker::DockerProcedure;
 use crate::s9pk::manifest::PackageId;
-use crate::util::serde::Reversible;
-use crate::Error;
+use crate::util::{display_none, serde::Reversible};
+use crate::{Error, ErrorKind};
+
+#[pin_project::pin_project]
+struct LogStream {
+    _child: Child,
+    #[pin]
+    entries: BoxStream<'static, Result<JournalctlEntry, Error>>,
+}
+impl Deref for LogStream {
+    type Target = BoxStream<'static, Result<JournalctlEntry, Error>>;
+    fn deref(&self) -> &Self::Target {
+        &self.entries
+    }
+}
+impl DerefMut for LogStream {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.entries
+    }
+}
+impl Stream for LogStream {
+    type Item = Result<JournalctlEntry, Error>;
+    fn poll_next(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Option<Self::Item>> {
+        let this = self.project();
+        Stream::poll_next(this.entries, cx)
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        self.entries.size_hint()
+    }
+}
+
+#[instrument(skip(logs, ws_fut))]
+async fn ws_handler<
+    WSFut: Future<Output = Result<Result<WebSocketStream<Upgraded>, HyperError>, JoinError>>,
+>(
+    first_entry: Option<LogEntry>,
+    mut logs: LogStream,
+    ws_fut: WSFut,
+) -> Result<(), Error> {
+    let mut stream = ws_fut
+        .await
+        .with_kind(crate::ErrorKind::Network)?
+        .with_kind(crate::ErrorKind::Unknown)?;
+
+    if let Some(first_entry) = first_entry {
+        stream
+            .send(Message::Text(
+                serde_json::to_string(&first_entry).with_kind(ErrorKind::Serialization)?,
+            ))
+            .await
+            .with_kind(ErrorKind::Network)?;
+    }
+
+    while let Some(entry) = tokio::select! {
+        a = logs.try_next() => Some(a?),
+        a = stream.try_next() => { a.with_kind(crate::ErrorKind::Network)?; None }
+    } {
+        if let Some(entry) = entry {
+            let (_, log_entry) = entry.log_entry()?;
+            stream
+                .send(Message::Text(
+                    serde_json::to_string(&log_entry).with_kind(ErrorKind::Serialization)?,
+                ))
+                .await
+                .with_kind(ErrorKind::Network)?;
+        }
+    }
+
+    stream
+        .close(Some(CloseFrame {
+            code: CloseCode::Normal,
+            reason: "Log Stream Finished".into(),
+        }))
+        .await
+        .with_kind(ErrorKind::Network)?;
+
+    Ok(())
+}

 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
 #[serde(rename_all = "kebab-case")]
@@ -25,6 +120,12 @@ pub struct LogResponse {
    start_cursor: Option<String>,
    end_cursor: Option<String>,
 }
+#[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
+#[serde(rename_all = "kebab-case", tag = "type")]
+pub struct LogFollowResponse {
+    start_cursor: Option<String>,
+    guid: RequestGuid,
+}

 #[derive(serde::Serialize, serde::Deserialize, Debug, Clone)]
 pub struct LogEntry {
@@ -111,38 +212,145 @@ pub enum LogSource {
    Container(PackageId),
 }

-pub fn display_logs(all: LogResponse, _: &ArgMatches) {
-    for entry in all.entries.iter() {
-        println!("{}", entry);
-    }
-}
-
-#[command(display(display_logs))]
+#[command(
+    custom_cli(cli_logs(async, context(CliContext))),
+    subcommands(self(logs_nofollow(async)), logs_follow),
+    display(display_none)
+)]
 pub async fn logs(
    #[arg] id: PackageId,
-    #[arg] limit: Option<usize>,
-    #[arg] cursor: Option<String>,
-    #[arg] before_flag: Option<bool>,
+    #[arg(short = 'l', long = "limit")] limit: Option<usize>,
+    #[arg(short = 'c', long = "cursor")] cursor: Option<String>,
+    #[arg(short = 'B', long = "before", default)] before: bool,
+    #[arg(short = 'f', long = "follow", default)] follow: bool,
+) -> Result<(PackageId, Option<usize>, Option<String>, bool, bool), Error> {
+    Ok((id, limit, cursor, before, follow))
+}
+pub async fn cli_logs(
+    ctx: CliContext,
+    (id, limit, cursor, before, follow): (PackageId, Option<usize>, Option<String>, bool, bool),
+) -> Result<(), RpcError> {
+    if follow {
+        if cursor.is_some() {
+            return Err(RpcError::from(Error::new(
+                eyre!("The argument '--cursor <cursor>' cannot be used with '--follow'"),
+                crate::ErrorKind::InvalidRequest,
+            )));
+        }
+        if before {
+            return Err(RpcError::from(Error::new(
+                eyre!("The argument '--before' cannot be used with '--follow'"),
+                crate::ErrorKind::InvalidRequest,
+            )));
+        }
+        cli_logs_generic_follow(ctx, "package.logs.follow", Some(id), limit).await
+    } else {
+        cli_logs_generic_nofollow(ctx, "package.logs", Some(id), limit, cursor, before).await
+    }
+}
+pub async fn logs_nofollow(
+    _ctx: (),
+    (id, limit, cursor, before, _): (PackageId, Option<usize>, Option<String>, bool, bool),
 ) -> Result<LogResponse, Error> {
-    Ok(fetch_logs(
-        LogSource::Container(id),
-        limit,
-        cursor,
-        before_flag.unwrap_or(false),
-    )
-    .await?)
+    fetch_logs(LogSource::Container(id), limit, cursor, before).await
+}
+#[command(rpc_only, rename = "follow", display(display_none))]
+pub async fn logs_follow(
+    #[context] ctx: RpcContext,
+    #[parent_data] (id, limit, _, _, _): (PackageId, Option<usize>, Option<String>, bool, bool),
+) -> Result<LogFollowResponse, Error> {
+    follow_logs(ctx, LogSource::Container(id), limit).await
 }

-#[instrument]
-pub async fn fetch_logs(
-    id: LogSource,
+pub async fn cli_logs_generic_nofollow(
+    ctx: CliContext,
+    method: &str,
+    id: Option<PackageId>,
    limit: Option<usize>,
    cursor: Option<String>,
-    before_flag: bool,
-) -> Result<LogResponse, Error> {
-    let mut cmd = Command::new("journalctl");
+    before: bool,
+) -> Result<(), RpcError> {
+    let res = rpc_toolkit::command_helpers::call_remote(
+        ctx.clone(),
+        method,
+        serde_json::json!({
+            "id": id,
+            "limit": limit,
+            "cursor": cursor,
+            "before": before,
+        }),
+        PhantomData::<LogResponse>,
+    )
+    .await?
+    .result?;

-    let limit = limit.unwrap_or(50);
+    for entry in res.entries.iter() {
+        println!("{}", entry);
+    }
+
+    Ok(())
+}
+
+pub async fn cli_logs_generic_follow(
+    ctx: CliContext,
+    method: &str,
+    id: Option<PackageId>,
+    limit: Option<usize>,
+) -> Result<(), RpcError> {
+    let res = rpc_toolkit::command_helpers::call_remote(
+        ctx.clone(),
+        method,
+        serde_json::json!({
+            "id": id,
+            "limit": limit,
+        }),
+        PhantomData::<LogFollowResponse>,
+    )
+    .await?
+    .result?;
+
+    let mut base_url = ctx.base_url.clone();
+    let ws_scheme = match base_url.scheme() {
+        "https" => "wss",
+        "http" => "ws",
+        _ => {
+            return Err(Error::new(
+                eyre!("Cannot parse scheme from base URL"),
+                crate::ErrorKind::ParseUrl,
+            )
+            .into())
+        }
+    };
+    base_url.set_scheme(ws_scheme).or_else(|_| {
+        Err(Error::new(
+            eyre!("Cannot set URL scheme"),
+            crate::ErrorKind::ParseUrl,
+        ))
+    })?;
+    let (mut stream, _) =
+                // base_url is "http://127.0.0.1/", with a trailing slash, so we don't put a leading slash in this path:
+                tokio_tungstenite::connect_async(format!("{}ws/rpc/{}", base_url, res.guid)).await?;
+    while let Some(log) = stream.try_next().await? {
+        match log {
+            Message::Text(log) => {
+                println!("{}", serde_json::from_str::<LogEntry>(&log)?);
+            }
+            _ => (),
+        }
+    }
+
+    Ok(())
+}
+
+async fn journalctl(
+    id: LogSource,
+    limit: usize,
+    cursor: Option<&str>,
+    before: bool,
+    follow: bool,
+) -> Result<LogStream, Error> {
+    let mut cmd = Command::new("journalctl");
+    cmd.kill_on_drop(true);

    cmd.arg("--output=json");
    cmd.arg("--output-fields=MESSAGE");
@@ -163,16 +371,15 @@ pub async fn fetch_logs(
        }
    };

-    let cursor_formatted = format!("--after-cursor={}", cursor.clone().unwrap_or("".to_owned()));
-    let mut get_prev_logs_and_reverse = false;
+    let cursor_formatted = format!("--after-cursor={}", cursor.clone().unwrap_or(""));
    if cursor.is_some() {
        cmd.arg(&cursor_formatted);
-        if before_flag {
-            get_prev_logs_and_reverse = true;
+        if before {
+            cmd.arg("--reverse");
        }
    }
-    if get_prev_logs_and_reverse {
-        cmd.arg("--reverse");
+    if follow {
+        cmd.arg("--follow");
    }

    let mut child = cmd.stdout(Stdio::piped()).spawn()?;
@@ -185,7 +392,7 @@ pub async fn fetch_logs(

    let journalctl_entries = LinesStream::new(out.lines());

-    let mut deserialized_entries = journalctl_entries
+    let deserialized_entries = journalctl_entries
        .map_err(|e| Error::new(e, crate::ErrorKind::Journald))
        .and_then(|s| {
            futures::future::ready(
@@ -194,16 +401,37 @@ pub async fn fetch_logs(
            )
        });

+    Ok(LogStream {
+        _child: child,
+        entries: deserialized_entries.boxed(),
+    })
+}
+
+#[instrument]
+pub async fn fetch_logs(
+    id: LogSource,
+    limit: Option<usize>,
+    cursor: Option<String>,
+    before: bool,
+) -> Result<LogResponse, Error> {
+    let limit = limit.unwrap_or(50);
+    let mut stream = journalctl(id, limit, cursor.as_deref(), before, false).await?;
+
    let mut entries = Vec::with_capacity(limit);
    let mut start_cursor = None;

-    if let Some(first) = deserialized_entries.try_next().await? {
+    if let Some(first) = tokio::time::timeout(Duration::from_secs(1), stream.try_next())
+        .await
+        .ok()
+        .transpose()?
+        .flatten()
+    {
        let (cursor, entry) = first.log_entry()?;
        start_cursor = Some(cursor);
        entries.push(entry);
    }

-    let (mut end_cursor, entries) = deserialized_entries
+    let (mut end_cursor, entries) = stream
        .try_fold(
            (start_cursor.clone(), entries),
            |(_, mut acc), entry| async move {
@@ -215,7 +443,7 @@ pub async fn fetch_logs(
        .await?;
    let mut entries = Reversible::new(entries);
    // reverse again so output is always in increasing chronological order
-    if get_prev_logs_and_reverse {
+    if cursor.is_some() && before {
        entries.reverse();
        std::mem::swap(&mut start_cursor, &mut end_cursor);
    }
@@ -226,21 +454,81 @@ pub async fn fetch_logs(
    })
 }

+#[instrument(skip(ctx))]
+pub async fn follow_logs(
+    ctx: RpcContext,
+    id: LogSource,
+    limit: Option<usize>,
+) -> Result<LogFollowResponse, Error> {
+    let limit = limit.unwrap_or(50);
+    let mut stream = journalctl(id, limit, None, false, true).await?;
+
+    let mut start_cursor = None;
+    let mut first_entry = None;
+
+    if let Some(first) = tokio::time::timeout(Duration::from_secs(1), stream.try_next())
+        .await
+        .ok()
+        .transpose()?
+        .flatten()
+    {
+        let (cursor, entry) = first.log_entry()?;
+        start_cursor = Some(cursor);
+        first_entry = Some(entry);
+    }
+
+    let guid = RequestGuid::new();
+    ctx.add_continuation(
+        guid.clone(),
+        RpcContinuation::ws(
+            Box::new(move |ws_fut| ws_handler(first_entry, stream, ws_fut).boxed()),
+            Duration::from_secs(30),
+        ),
+    )
+    .await;
+    Ok(LogFollowResponse { start_cursor, guid })
+}
+
+// #[tokio::test]
+// pub async fn test_logs() {
+//     let response = fetch_logs(
+//         // change `tor.service` to an actual journald unit on your machine
+//         // LogSource::Service("tor.service"),
+//         // first run `docker run --name=hello-world.embassy --log-driver=journald hello-world`
+//         LogSource::Container("hello-world".parse().unwrap()),
+//         // Some(5),
+//         None,
+//         None,
+//         // Some("s=1b8c418e28534400856c27b211dd94fd;i=5a7;b=97571c13a1284f87bc0639b5cff5acbe;m=740e916;t=5ca073eea3445;x=f45bc233ca328348".to_owned()),
+//         false,
+//         true,
+//     )
+//     .await
+//     .unwrap();
+//     let serialized = serde_json::to_string_pretty(&response).unwrap();
+//     println!("{}", serialized);
+// }
+
 #[tokio::test]
 pub async fn test_logs() {
-    let response = fetch_logs(
-        // change `tor.service` to an actual journald unit on your machine
-        // LogSource::Service("tor.service"),
-        // first run `docker run --name=hello-world.embassy --log-driver=journald hello-world`
-        LogSource::Container("hello-world".parse().unwrap()),
-        // Some(5),
-        None,
-        None,
-        // Some("s=1b8c418e28534400856c27b211dd94fd;i=5a7;b=97571c13a1284f87bc0639b5cff5acbe;m=740e916;t=5ca073eea3445;x=f45bc233ca328348".to_owned()),
-        false,
-    )
-    .await
-    .unwrap();
-    let serialized = serde_json::to_string_pretty(&response).unwrap();
-    println!("{}", serialized);
+    let mut cmd = Command::new("journalctl");
+    cmd.kill_on_drop(true);
+
+    cmd.arg("-f");
+    cmd.arg("CONTAINER_NAME=hello-world.embassy");
+
+    let mut child = cmd.stdout(Stdio::piped()).spawn().unwrap();
+    let out = BufReader::new(
+        child
+            .stdout
+            .take()
+            .ok_or_else(|| Error::new(eyre!("No stdout available"), crate::ErrorKind::Journald))
+            .unwrap(),
+    );
+
+    let mut journalctl_entries = LinesStream::new(out.lines());
+
+    while let Some(line) = journalctl_entries.try_next().await.unwrap() {
+        dbg!(line);
+    }
 }