Add initial support for operating with workflows

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "gvltctl"
-version = "0.1.2"
+version = "0.1.3"
 edition = "2021"
 authors = ["Gevulot Team"]
 license = "MIT OR Apache-2.0"
@@ -11,7 +11,7 @@ description = "Gevulot Control CLI"
 # TODO: change rev to tag when available
 # NOTE: this revision is aligned with `mia-installer` dependency.
 # Be careful changing it.
-gevulot-rs = { git = "https://github.com/gevulotnetwork/gevulot-rs.git", rev = "f6657f57771b1c549a83cfe6fc822c3db687707e" }
+gevulot-rs = { git = "https://github.com/gevulotnetwork/gevulot-rs.git", rev = "961a71a9ff933ee5767e7f96597767c809e2c36e" }
 
 bip32 = "0.5.1"
 clap = { version = "4", features = ["env", "cargo"] }
@@ -26,7 +26,7 @@ tokio = { version = "1", features = ["full"] }
 toml = "0.8.19"
 
 [target.'cfg(target_os = "linux")'.dependencies]
-mia-installer = { git = "https://github.com/gevulotnetwork/mia.git", tag = "mia-installer-0.2.2" }
+mia-installer = { git = "https://github.com/gevulotnetwork/mia.git", rev = "4434218127c7aa98a74a4abb9fca72eadfc04c40" }
 
 anyhow = "1"
 log = "0.4.22"

src/builders/skopeo_builder.rs

@@ -1,6 +1,6 @@
 use anyhow::{Context, Result};
 use log::debug;
-use gevulot_rs::runtime_config::{self, RuntimeConfig};
+use mia_installer::runtime_config::{self, RuntimeConfig};
 use oci_spec::image::{ImageConfiguration, ImageManifest};
 use std::io::{self, BufRead, BufReader, Write};
 use std::{env, fs, path::Path, process::Command};

src/commands/mod.rs

@@ -2,5 +2,6 @@
 pub mod build;
 pub mod pins;
 pub mod tasks;
+pub mod workflows;
 pub mod workers;
 pub mod sudo;

src/commands/pins.rs

@@ -30,8 +30,8 @@ pub async fn list_pins(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::err
 /// * `_sub_m` - A reference to the ArgMatches struct containing parsed command-line arguments.
 ///              This is used to access the CID of the pin to retrieve and any additional options.
 pub async fn get_pin(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = connect_to_gevulot(_sub_m).await?;
     if let Some(pin_cid) = _sub_m.get_one::<String>("cid") {
+        let mut client = connect_to_gevulot(_sub_m).await?;
         let pin = client.pins.get(pin_cid).await?;
         // Convert the pin to the gevulot_rs::models::Pin type
         let pin: gevulot_rs::models::Pin = pin.into();

src/commands/tasks.rs

@@ -33,8 +33,8 @@ pub async fn list_tasks(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::er
 ///
 /// A Result indicating success or an error if the task retrieval fails.
 pub async fn get_task(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = crate::connect_to_gevulot(_sub_m).await?;
     if let Some(task_id) = _sub_m.get_one::<String>("id") {
+        let mut client = crate::connect_to_gevulot(_sub_m).await?;
         let task = client.tasks.get(task_id).await?;
         let task: gevulot_rs::models::Task = task.into();
         print_object(_sub_m, &task)?;

src/commands/workers.rs

@@ -31,8 +31,8 @@ pub async fn list_workers(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::
 ///
 /// A Result containing () if successful, or a Box<dyn std::error::Error> if an error occurs.
 pub async fn get_worker(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = connect_to_gevulot(_sub_m).await?;
     if let Some(worker_id) = _sub_m.get_one::<String>("id") {
+        let mut client = connect_to_gevulot(_sub_m).await?;
         let worker = client.workers.get(worker_id).await?;
         let worker: gevulot_rs::models::Worker = worker.into();
         print_object(_sub_m, &worker)?;

src/commands/workflows.rs

@@ -0,0 +1,97 @@
+use gevulot_rs::proto::gevulot::gevulot::{workflow_spec::Stage, InputContext, MsgCreateWorkflow, MsgDeleteWorkflow, OutputContext, TaskEnv, TaskSpec, WorkflowSpec};
+
+use crate::{connect_to_gevulot, print_object, read_file};
+
+
+
+
+pub async fn list_workflows(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
+    let mut client = crate::connect_to_gevulot(_sub_m).await?;
+    let workflows = client.workflows.list().await?;
+    let workflows: Vec<gevulot_rs::models::Workflow> = workflows.into_iter().map(Into::into).collect();
+    print_object(_sub_m, &workflows)?;
+    Ok(())
+}
+
+pub async fn get_workflow(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
+    if let Some(workflow_id) = _sub_m.get_one::<String>("id") {
+        let mut client = crate::connect_to_gevulot(_sub_m).await?;
+        let workflow = client.workflows.get(workflow_id).await?;
+        let workflow: gevulot_rs::models::Workflow = workflow.into();
+        print_object(_sub_m, &workflow)?;
+    } else {
+        println!("Workflow ID is required");
+    }
+    Ok(())
+}
+
+pub async fn create_workflow(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
+    let workflow: gevulot_rs::models::WorkflowSpec = read_file(_sub_m).await?;
+    let mut client = connect_to_gevulot(_sub_m).await?;
+    let me = client
+        .base_client
+        .write()
+        .await
+        .address
+        .clone()
+        .ok_or("No address found, did you set a mnemonic?")?;
+
+    let resp = client
+        .workflows
+        .create(MsgCreateWorkflow{
+            creator: me,
+            spec: Some(WorkflowSpec{
+                stages: workflow.stages.iter().map(|s| Stage{
+                    tasks: s.tasks.iter().map(|t| TaskSpec{
+                        image: t.image.clone(),
+                        command: t.command.clone(),
+                        args: t.args.clone(),
+                        env: t.env.iter().map(|e| TaskEnv{
+                            name: e.name.clone(),
+                            value: e.value.clone()
+                        }).collect(),
+                        input_contexts: t.input_contexts.iter().map(|ic| InputContext{
+                            source: ic.source.clone(),
+                            target: ic.target.clone()
+                        }).collect(),
+                        output_contexts: t.output_contexts.iter().map(|oc| OutputContext{
+                            source: oc.source.clone(),
+                            retention_period: oc.retention_period as u64
+                        }).collect(),
+                        cpus: t.resources.cpus as u64,
+                        gpus: t.resources.gpus as u64,
+                        memory: t.resources.memory as u64,
+                        time: t.resources.time as u64,
+                        store_stdout: t.store_stdout.unwrap_or(false),
+                        store_stderr: t.store_stderr.unwrap_or(false),
+                        workflow_ref: "".to_string(),
+                    }).collect::<Vec<TaskSpec>>(),
+                }).collect::<Vec<Stage>>(),
+            }),
+        }).await?;
+
+    println!("Created workflow with ID: {}", resp.id);
+    Ok(())
+}
+
+pub async fn delete_workflow(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
+    if let Some(workflow_id) = _sub_m.get_one::<String>("id") {
+        let mut client = crate::connect_to_gevulot(_sub_m).await?;
+        let me = client
+            .base_client
+            .write()
+            .await
+            .address
+            .clone()
+            .ok_or("No address found, did you set a mnemonic?")?;
+
+        client.workflows.delete(MsgDeleteWorkflow{
+            creator: me,
+            id: workflow_id.clone(),
+        }).await?;
+        println!("ok");
+    } else {
+        println!("Workflow ID is required");
+    }
+    Ok(())
+}

src/main.rs

@@ -17,7 +17,7 @@ mod commands;
 
 #[cfg(target_os = "linux")]
 use commands::build::*;
-use commands::{pins::*, sudo::*, tasks::*, workers::*};
+use commands::{pins::*, sudo::*, tasks::*, workflows::*,workers::*};
 
 /// Main entry point for the Gevulot Control CLI application.
 ///
@@ -718,22 +718,3 @@ async fn generate_completion(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn st
     }
     Ok(())
 }
-async fn list_workflows(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    println!("Listing all workflows");
-    todo!();
-}
-
-async fn get_workflow(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    println!("Getting a specific workflow");
-    todo!();
-}
-
-async fn create_workflow(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    println!("Creating a new workflow");
-    todo!();
-}
-
-async fn delete_workflow(_sub_m: &clap::ArgMatches) -> Result<(), Box<dyn std::error::Error>> {
-    println!("Deleting a workflow");
-    todo!();
-}