feat: Add job command to start sessions as batch jobs

Cargo.toml

@@ -60,6 +60,7 @@ self_update = { version = "0.43.1", features = [
   "compression-flate2",
 ] }
 md5 = "0.8.0"
+ulid = "1.2.1"
 
 [features]
 default = ["reqwest/default-tls"] # link against system library

flake.nix

@@ -6,7 +6,6 @@
 
     crane = {
       url = "github:ipetkov/crane";
-      inputs.nixpkgs.follows = "nixpkgs";
     };
 
     fenix = {
@@ -42,12 +41,14 @@
       craneLib = crane.mkLib pkgs;
       src = craneLib.cleanCargoSource ./.;
       docSrc = ./docs;
+      version = "0.0.0";
 
       fenixToolChain = fenix.packages.${system}.complete;
 
       # Common arguments can be set here to avoid repeating them later
       commonArgs = {
         inherit src;
+        inherit version;
         strictDeps = true;
 
         nativeBuildInputs = [
@@ -128,17 +129,20 @@
         # Check formatting
         my-crate-fmt = craneLib.cargoFmt {
           inherit src;
+          inherit version;
         };
 
         # Audit dependencies
         my-crate-audit = craneLib.cargoAudit {
           inherit src advisory-db;
           cargoAuditExtraArgs = "--ignore RUSTSEC-2023-0071";
+          inherit version;
         };
 
         # Audit licenses
         my-crate-deny = craneLib.cargoDeny {
           inherit src;
+          inherit version;
         };
 
         # Run tests with cargo-nextest

src/cli.rs

@@ -27,6 +27,7 @@ pub async fn execute_cmd(opts: MainOpts) -> Result<(), CmdError> {
         SubCommand::UserDoc(input) => input.exec(ctx).await?,
 
         SubCommand::Dataset(input) => input.exec(ctx).await?,
+        SubCommand::Job(input) => input.exec(ctx).await?,
     };
     Ok(())
 }

src/cli/cmd.rs

@@ -1,4 +1,5 @@
 pub mod dataset;
+pub mod job;
 pub mod login;
 pub mod project;
 pub mod update;
@@ -113,6 +114,15 @@ pub enum CmdError {
 
     #[snafu(display("Dataset - {}", source))]
     Dataset { source: dataset::Error },
+
+    #[snafu(display("Job - {}", source))]
+    Job { source: job::Error },
+}
+
+impl From<job::Error> for CmdError {
+    fn from(source: job::Error) -> Self {
+        CmdError::Job { source }
+    }
 }
 
 impl From<version::Error> for CmdError {

src/cli/cmd/job.rs

@@ -0,0 +1,56 @@
+pub mod list;
+pub mod logs;
+pub mod start;
+pub mod stop;
+
+use super::Context;
+use clap::Parser;
+use snafu::{ResultExt, Snafu};
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Error starting job: {}", source))]
+    Start { source: start::Error },
+
+    #[snafu(display("Error stopping job: {}", source))]
+    Stop { source: stop::Error },
+
+    #[snafu(display("Error listing jobs: {}", source))]
+    List { source: list::Error },
+
+    #[snafu(display("Error getting logs: {}", source))]
+    Logs { source: logs::Error },
+}
+
+/// Sub command for managing projects
+#[derive(Parser, Debug)]
+pub struct Input {
+    #[command(subcommand)]
+    pub subcmd: JobCommand,
+}
+
+impl Input {
+    pub async fn exec(&self, ctx: Context) -> Result<(), Error> {
+        match &self.subcmd {
+            JobCommand::Start(input) => input.exec(ctx).await.context(StartSnafu),
+            JobCommand::Stop(input) => input.exec(ctx).await.context(StopSnafu),
+            JobCommand::List(input) => input.exec(ctx).await.context(ListSnafu),
+            JobCommand::Logs(input) => input.exec(ctx).await.context(LogsSnafu),
+        }
+    }
+}
+
+#[derive(Parser, Debug)]
+pub enum JobCommand {
+    #[command()]
+    Start(start::Input),
+
+    #[command()]
+    Stop(stop::Input),
+
+    #[command()]
+    List(list::Input),
+
+    #[command()]
+    Logs(logs::Input),
+}

src/cli/cmd/job/list.rs

@@ -0,0 +1,36 @@
+use super::Context;
+use crate::{
+    cli::sink::Error as SinkError,
+    httpclient::{self, data::SessionMode},
+};
+
+use clap::Parser;
+
+use snafu::{ResultExt, Snafu};
+
+/// Listing jobs.
+///
+/// List currently running jobs.
+#[derive(Parser, Debug)]
+pub struct Input {}
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Error writing data: {}", source))]
+    WriteResult { source: SinkError },
+
+    #[snafu(display("Http error: {}", source))]
+    HttpClient { source: httpclient::Error },
+}
+
+impl Input {
+    pub async fn exec(&self, ctx: Context) -> Result<(), Error> {
+        let result = ctx
+            .client
+            .list_sessions(Some(SessionMode::NonInteractive))
+            .await
+            .context(HttpClientSnafu)?;
+
+        ctx.write_result(&result).await.context(WriteResultSnafu)
+    }
+}

src/cli/cmd/job/logs.rs

@@ -0,0 +1,48 @@
+use super::Context;
+use crate::{cli::sink::Error as SinkError, data::simple_message::SimpleMessage, httpclient};
+
+use clap::{Parser, ValueHint};
+
+use snafu::{ResultExt, Snafu};
+
+/// Listing logs of a jobs.
+///
+/// List the logs of a job.
+#[derive(Parser, Debug)]
+pub struct Input {
+    #[arg(value_hint=ValueHint::Other)]
+    pub job_id: String,
+}
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Error writing data: {}", source))]
+    WriteResult { source: SinkError },
+
+    #[snafu(display("Http error: {}", source))]
+    HttpClient { source: httpclient::Error },
+}
+
+impl Input {
+    pub async fn exec(&self, ctx: Context) -> Result<(), Error> {
+        let result = ctx
+            .client
+            .session_logs(&self.job_id)
+            .await
+            .context(HttpClientSnafu)?;
+
+        if let Some(lines) = result.0.get("amalthea-session") {
+            ctx.write_result(&SimpleMessage {
+                message: lines.to_string(),
+            })
+            .await
+            .context(WriteResultSnafu)
+        } else {
+            ctx.write_result(&SimpleMessage {
+                message: "No logs available.".to_string(),
+            })
+            .await
+            .context(WriteResultSnafu)
+        }
+    }
+}

src/cli/cmd/job/start.rs

@@ -0,0 +1,44 @@
+use crate::httpclient::{self, data::SessionStartRequest};
+
+use super::Context;
+use crate::cli::sink::Error as SinkError;
+
+use clap::{Parser, ValueHint};
+use ulid::Ulid;
+
+use snafu::{ResultExt, Snafu};
+
+/// Start a job.
+///
+/// Starts a non-interactive session using a pre-configured session launcher.
+#[derive(Parser, Debug)]
+pub struct Input {
+    /// The launcher to use for launching the job.
+    #[arg(value_hint=ValueHint::Other)]
+    pub launcher: Ulid,
+}
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Error writing data: {}", source))]
+    WriteResult { source: SinkError },
+
+    #[snafu(display("Http error: {}", source))]
+    HttpClient { source: httpclient::Error },
+}
+
+impl Input {
+    pub async fn exec(&self, ctx: Context) -> Result<(), Error> {
+        let req = SessionStartRequest {
+            launcher_id: self.launcher.to_string(),
+            session_type: "non-interactive".into(),
+        };
+        let result = ctx
+            .client
+            .start_session(req)
+            .await
+            .context(HttpClientSnafu)?;
+
+        ctx.write_result(&result).await.context(WriteResultSnafu)
+    }
+}

src/cli/cmd/job/stop.rs

@@ -0,0 +1,41 @@
+use crate::{data::simple_message::SimpleMessage, httpclient};
+
+use super::Context;
+use crate::cli::sink::Error as SinkError;
+
+use clap::{Parser, ValueHint};
+
+use snafu::{ResultExt, Snafu};
+
+/// Stop a job.
+///
+/// Stop a running non-interactive session.
+#[derive(Parser, Debug)]
+pub struct Input {
+    /// The launcher to use for launching the job.
+    #[arg(value_hint=ValueHint::Other)]
+    pub job_id: String,
+}
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Error writing data: {}", source))]
+    WriteResult { source: SinkError },
+
+    #[snafu(display("Http error: {}", source))]
+    HttpClient { source: httpclient::Error },
+}
+
+impl Input {
+    pub async fn exec(&self, ctx: Context) -> Result<(), Error> {
+        ctx.client
+            .stop_session(&self.job_id)
+            .await
+            .context(HttpClientSnafu)?;
+        ctx.write_result(&SimpleMessage {
+            message: "Job is being removed.".into(),
+        })
+        .await
+        .context(WriteResultSnafu)
+    }
+}

src/cli/cmd/project/clone.rs

@@ -67,10 +67,7 @@ impl Input {
     pub async fn exec(&self, ctx: Context) -> Result<(), Error> {
         let opt_details = ctx
             .client
-            .get_project(
-                &self.project_ref,
-                ctx.opts.verbosity.log_level().unwrap_or(log::Level::Warn) > log::Level::Info,
-            )
+            .get_project(&self.project_ref)
             .await
             .context(HttpClientSnafu)?;
         if let Some(details) = opt_details {

src/cli/cmd/version.rs

@@ -36,13 +36,7 @@ impl Input {
             let vinfo = BuildInfo::default();
             ctx.write_result(&vinfo).await.context(WriteResultSnafu)?;
         } else {
-            let result = ctx
-                .client
-                .version(
-                    ctx.opts.verbosity.log_level().unwrap_or(log::Level::Warn) > log::Level::Info,
-                )
-                .await
-                .context(HttpClientSnafu)?;
+            let result = ctx.client.version().await.context(HttpClientSnafu)?;
             let urlstr = ctx.renku_url().as_str();
             let vinfo = Versions::create(result, urlstr);
             ctx.write_result(&vinfo).await.context(WriteResultSnafu)?;

src/cli/opts.rs

@@ -66,6 +66,9 @@ pub enum SubCommand {
 
     #[command()]
     Dataset(dataset::Input),
+
+    #[command()]
+    Job(job::Input),
 }
 
 /// This is the command line interface to the Renku platform. Main

src/cli/sink.rs

@@ -61,3 +61,6 @@ impl Sink for BuildInfo {}
 impl Sink for PathEntry {}
 impl Sink for UserCode {}
 impl Sink for Response {}
+impl Sink for SessionStartResponse {}
+impl Sink for SessionList {}
+impl Sink for SessionLogs {}