Support log snapshot (#52)

The log snapshot is basically printing a copy of the current
logs buffer (then terminate) instead of the default `follow`
behavior.

by providing a `-s` flag to the log command line, the logs will
print the current buffer, then exits.

running `zinit log` will still automatically follow the logs as
the default behavior

Author: muhamadazmy

src/app/api.rs

@@ -65,7 +65,10 @@ impl Api {
         };
 
         let response = match Self::process(cmd, &mut stream, zinit).await {
-            Ok(body) => Response {
+            // When process returns None means we can terminate without
+            // writing any result to the socket.
+            Ok(None) => return,
+            Ok(Some(body)) => Response {
                 body,
                 state: State::Ok,
             },
@@ -94,7 +97,7 @@ impl Api {
         cmd: String,
         stream: &mut BufStream<UnixStream>,
         zinit: ZInit,
-    ) -> Result<Value> {
+    ) -> Result<Option<Value>> {
         let parts = match shlex::split(&cmd) {
             Some(parts) => parts,
             None => bail!("invalid command syntax"),
@@ -104,23 +107,34 @@ impl Api {
             bail!("unknown command");
         }
 
-        match parts[0].as_ref() {
+        if &parts[0] == "log" {
+            match parts.len() {
+                1 => Self::log(stream, zinit, true).await,
+                2 if parts[1] == "snapshot" => Self::log(stream, zinit, false).await,
+                _ => bail!("invalid log command arguments"),
+            }?;
+
+            return Ok(None);
+        }
+
+        let value = match parts[0].as_ref() {
             "list" => Self::list(zinit).await,
             "shutdown" => Self::shutdown(zinit).await,
             "reboot" => Self::reboot(zinit).await,
-            "log" => Self::log(stream, zinit).await,
             "start" if parts.len() == 2 => Self::start(&parts[1], zinit).await,
             "stop" if parts.len() == 2 => Self::stop(&parts[1], zinit).await,
             "kill" if parts.len() == 3 => Self::kill(&parts[1], &parts[2], zinit).await,
             "status" if parts.len() == 2 => Self::status(&parts[1], zinit).await,
             "forget" if parts.len() == 2 => Self::forget(&parts[1], zinit).await,
             "monitor" if parts.len() == 2 => Self::monitor(&parts[1], zinit).await,
             _ => bail!("unknown command '{}' or wrong arguments count", parts[0]),
-        }
+        }?;
+
+        Ok(Some(value))
     }
 
-    async fn log(stream: &mut BufStream<UnixStream>, zinit: ZInit) -> Result<Value> {
-        let mut logs = zinit.logs().await?;
+    async fn log(stream: &mut BufStream<UnixStream>, zinit: ZInit, follow: bool) -> Result<Value> {
+        let mut logs = zinit.logs(follow).await;
 
         while let Some(line) = logs.recv().await {
             stream.write_all(line.as_bytes()).await?;
@@ -256,9 +270,18 @@ impl Client {
         &self,
         mut out: O,
         filter: Option<S>,
+        follow: bool,
     ) -> Result<()> {
         let mut con = self.connect().await?;
-        con.write_all(b"log\n").await?;
+        if follow {
+            // default behavior of log with no extra arguments
+            // is to stream all logs
+            con.write_all(b"log\n").await?;
+        } else {
+            // adding a snapshot subcmd will make it auto terminate
+            // immediate after
+            con.write_all(b"log snapshot\n").await?;
+        }
         con.flush().await?;
         match filter {
             None => tokio::io::copy(&mut con, &mut out).await?,

src/app/mod.rs

@@ -146,7 +146,10 @@ pub async fn kill(socket: &str, name: &str, signal: &str) -> Result<()> {
     client.kill(name, signal).await?;
     Ok(())
 }
-pub async fn logs(socket: &str, filter: Option<&str>) -> Result<()> {
+pub async fn logs(socket: &str, filter: Option<&str>, follow: bool) -> Result<()> {
     let client = api::Client::new(socket);
-    client.logs(tokio::io::stdout(), filter).await
+    if let Some(filter) = filter {
+        client.status(filter).await?;
+    }
+    client.logs(tokio::io::stdout(), filter, follow).await
 }

src/main.rs

@@ -107,6 +107,13 @@ async fn main() -> Result<()> {
         )
         .subcommand(
             SubCommand::with_name("log")
+                .arg(
+                    Arg::with_name("snapshot")
+                        .short("s")
+                        .long("snapshot")
+                        .required(false)
+                        .help("if set log prints current buffer without following")
+                    )
                 .arg(
                     Arg::with_name("filter")
                         .value_name("FILTER")
@@ -171,14 +178,21 @@ async fn main() -> Result<()> {
             )
             .await
         }
-        ("log", Some(matches)) => app::logs(socket, matches.value_of("filter")).await,
+        ("log", Some(matches)) => {
+            app::logs(
+                socket,
+                matches.value_of("filter"),
+                !matches.is_present("snapshot"),
+            )
+            .await
+        }
         _ => app::list(socket).await, // default command
     };
 
     match result {
         Ok(_) => Ok(()),
         Err(e) => {
-            eprintln!("{}", e);
+            eprintln!("{:#}", e);
             std::process::exit(1);
         }
     }

src/manager/buffer.rs

@@ -97,7 +97,12 @@ impl Ring {
         Ok(())
     }
 
-    pub async fn stream(&self) -> Result<Logs> {
+    /// stream returns a continues stream that first receive
+    /// a snapshot of the current buffer.
+    /// then if follow is true the logs stream will remain
+    /// open and fed each received line forever until the
+    /// received closed the channel from its end.
+    pub async fn stream(&self, follow: bool) -> Logs {
         let (tx, stream) = mpsc::channel::<Arc<String>>(100);
         let mut rx = self.sender.subscribe();
         let buffer = self
@@ -112,6 +117,10 @@ impl Ring {
                 let _ = tx.send(Arc::clone(&item)).await;
             }
 
+            if !follow {
+                return;
+            }
+
             loop {
                 let line = match rx.recv().await {
                     Ok(line) => line,
@@ -128,6 +137,6 @@ impl Ring {
             }
         });
 
-        Ok(stream)
+        stream
     }
 }

src/manager/mod.rs

@@ -141,8 +141,8 @@ impl ProcessManager {
         });
     }
 
-    pub async fn stream(&self) -> Result<Logs> {
-        self.ring.stream().await
+    pub async fn stream(&self, follow: bool) -> Logs {
+        self.ring.stream(follow).await
     }
 
     pub fn signal(&self, pid: Pid, sig: signal::Signal) -> Result<()> {

src/zinit/config.rs

@@ -23,19 +23,15 @@ impl Default for Signal {
     }
 }
 
-#[derive(Clone, Debug, Deserialize)]
+#[derive(Default, Clone, Debug, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum Log {
     None,
+    #[default]
     Ring,
     Stdout,
 }
 
-impl Default for Log {
-    fn default() -> Self {
-        Log::Ring
-    }
-}
 fn default_shutdown_timeout_fn() -> u64 {
     DEFAULT_SHUTDOWN_TIMEOUT
 }

src/zinit/mod.rs

@@ -194,8 +194,8 @@ impl ZInit {
         let _ = self.shutdown().await;
     }
 
-    pub async fn logs(&self) -> Result<Logs> {
-        self.pm.stream().await
+    pub async fn logs(&self, follow: bool) -> Logs {
+        self.pm.stream(follow).await
     }
 
     pub async fn monitor<S: Into<String>>(&self, name: S, service: config::Service) -> Result<()> {
@@ -668,7 +668,7 @@ impl ZInit {
             drop(service);
             if config.one_shot {
                 // we don't need to restart the service anymore
-                let _ = self.notify.notify_waiters();
+                self.notify.notify_waiters();
                 break;
             }
             // we trying again in 2 seconds