Logs wip

Cargo.toml

@@ -34,3 +34,4 @@ fs2 = "0.4.3"
 log = "0.4.20"
 handlebars = "5.1.2"
 cron = "0.12.1"
+russh = "0.38.0"

README.md

@@ -155,11 +155,12 @@ sudo apt-get install -y gcc make libssl-dev pkg-config
         - [ ] Remove
         - [x] List
         - [ ] Store manifest in store so CNI plugin can get access
-        - [ ] Fix pod naming to avoid collisions
+        - [x] Fix pod naming to avoid collisions
     - Deployments
         - [x] Apply
         - [x] Remove
         - [x] List
+        - [ ] Logs
         - [x] Output matches kubectl
     - Daemonsets
         - [x] Apply

src/lib.rs

@@ -16,6 +16,7 @@ mod executor;
 mod describe;
 mod filestore;
 mod cron;
+mod logs;
 
 pub use skate::skate;
 pub use skatelet::skatelet;

src/logs.rs

@@ -0,0 +1,69 @@
+use std::error::Error;
+use std::process;
+use std::process::Stdio;
+use anyhow::anyhow;
+use clap::Args;
+use futures::stream::FuturesUnordered;
+use crate::config::Config;
+use crate::create::CreateCommands;
+use crate::skate::ConfigFileArgs;
+use crate::ssh;
+use futures::StreamExt;
+
+#[derive(Debug, Args)]
+#[command(arg_required_else_help(true))]
+pub struct LogArgs {
+    #[command(flatten)]
+    config: ConfigFileArgs,
+    #[arg(short, long, long_help = "Specify if the logs should be streamed.")]
+    pub follow: bool,
+    #[arg(
+        short, default_value_t = - 1, long, long_help = "Lines of recent log file to display. Defaults to -1."
+    )]
+    pub tail: i32,
+    #[arg(long, short, long_help = "Filter by resource namespace")]
+    namespace: Option<String>,
+    #[arg(trailing_var_arg = true, name = "POD | TYPE/NAME")]
+    var_args: Vec<String>,
+}
+
+pub async fn logs(args: LogArgs) -> Result<(), Box<dyn Error>> {
+
+    let config = Config::load(Some(args.config.skateconfig.clone()))?;
+    let (conns, errors) = ssh::cluster_connections(config.current_cluster()?).await;
+
+    if errors.is_some() {
+        eprintln!("{}", errors.as_ref().unwrap())
+    }
+
+    if conns.is_none() {
+        if errors.is_some() {
+            return Err(anyhow!(errors.unwrap().to_string()).into());
+        }
+        println!("No connections found");
+        return Ok(());
+    }
+
+    let conns = conns.unwrap();
+
+    let name = args.var_args.first();
+    if name.is_none() {
+        return Err("No resource name provided".into());
+    }
+
+    let name = name.unwrap();
+    let ns = args.namespace.unwrap_or("default".to_string());
+
+    let cmd = format!("sudo podman logs {}", name);
+    let fut: FuturesUnordered<_> = conns.clients.iter().map(|c| c.execute_stdout(&cmd)).collect();
+
+    let result: Vec<_> = fut.collect().await;
+
+    for res in result {
+        match res {
+            Err(e) => eprintln!("{}", e),
+            _ => {}
+        }
+    }
+    Ok(())
+}

src/skate.rs

@@ -35,6 +35,7 @@ use crate::create::{create, CreateArgs};
 use crate::delete::{delete, DeleteArgs};
 use crate::get::{get, GetArgs};
 use crate::describe::{DescribeArgs, describe};
+use crate::logs::{LogArgs, logs};
 use crate::skate::Distribution::{Debian, Raspbian, Ubuntu, Unknown};
 use crate::skate::Os::{Darwin, Linux};
 use crate::ssh::SshClient;
@@ -57,6 +58,7 @@ enum Commands {
     Refresh(RefreshArgs),
     Get(GetArgs),
     Describe(DescribeArgs),
+    Logs(LogArgs)
 }
 
 #[derive(Debug, Clone, Args)]
@@ -78,6 +80,7 @@ pub async fn skate() -> Result<(), Box<dyn Error>> {
         Commands::Refresh(args) => refresh(args).await,
         Commands::Get(args) => get(args).await,
         Commands::Describe(args) => describe(args).await,
+        Commands::Logs(args) => logs(args).await,
         _ => Ok(())
     }
 }

src/ssh.rs

@@ -1,6 +1,8 @@
 use std::error::Error;
+use russh;
 use std::fmt;
 use std::fmt::{Debug, Formatter};
+use std::process::Stdio;
 use std::time::Duration;
 use anyhow::anyhow;
 use async_ssh2_tokio::{AuthMethod, ServerCheckMethod};
@@ -242,6 +244,41 @@ echo ovs=$(cat /tmp/ovs-$$);
         }
     }
 
+    pub async fn execute_stdout(self: &SshClient, cmd: &str) -> Result<(), Box<dyn Error>> {
+        let mut ch = self.client.get_channel().await?;
+        let _ = ch.exec(true, cmd).await?;
+
+        let mut result: Option<_> = None;
+
+        while let Some(msg) = ch.wait().await {
+            //dbg!(&msg);
+            match msg {
+                // If we get data, add it to the buffer
+                russh::ChannelMsg::Data { ref data } => print!("{}", &String::from_utf8_lossy(&data.to_vec())),
+                russh::ChannelMsg::ExtendedData { ref data, ext } => {
+                    if ext == 1 {
+                        eprint!("{}", &String::from_utf8_lossy(&data.to_vec()))
+                    }
+                }
+                // If we get an exit code report, store it, but crucially don't
+                // assume this message means end of communications. The data might
+                // not be finished yet!
+                russh::ChannelMsg::ExitStatus { exit_status } => result = Some(exit_status),
+
+                // We SHOULD get this EOF messagge, but 4254 sec 5.3 also permits
+                // the channel to close without it being sent. And sometimes this
+                // message can even precede the Data message, so don't handle it
+                // russh::ChannelMsg::Eof => break,
+                _ => {}
+            }
+        }
+
+        if result.is_none() || result == Some(0) {
+            return Ok(());
+        }
+        Err(anyhow!("exit status {}", result.unwrap()).into())
+    }
+
     pub async fn execute(self: &SshClient, cmd: &str) -> Result<String, Box<dyn Error>> {
         cmd.lines().for_each(|l| println!("{} | > {}", self.node_name, l.green()));
         let result = self.client.execute(cmd).await.