upload data to buckets

Cargo.toml

@@ -23,5 +23,7 @@ tracing-subscriber = "0.3.18"
 openssl = { version = "0.10", features = ["vendored"] }
 
 termion = "3.0.0"
+walkdir = "2.4.0"
+reqwest = { version = "0.11.24", features = ["multipart"] }
 
 

src/agent.rs

@@ -1,15 +1,20 @@
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Agent {
     pub api_key: String,
+    pub robot_server_url: String,
 }
 
 pub struct AgentBuilder {
     api_key: Option<String>,
+    robot_server_url: Option<String>,
 }
 
 impl Default for AgentBuilder {
     fn default() -> Self {
-        Self { api_key: None }
+        Self {
+            api_key: None,
+            robot_server_url: None,
+        }
     }
 }
 
@@ -18,10 +23,14 @@ impl AgentBuilder {
         self.api_key = Some(api_key);
         self
     }
-
+    pub fn robot_server_url(mut self, robot_server_url: String) -> Self {
+        self.robot_server_url = Some(robot_server_url);
+        self
+    }
     pub fn build(self) -> Agent {
         Agent {
             api_key: self.api_key.unwrap(),
+            robot_server_url: self.robot_server_url.unwrap(),
         }
     }
 }

src/commands/docker.rs

@@ -14,16 +14,21 @@ use tracing::{error, info};
 
 use serde::{Deserialize, Serialize};
 
+use crate::agent;
 use crate::{
     commands::{RobotJob, RobotJobResult},
     store::{JobManager, Jobs},
+    utils::files::{
+        create_job_data_dir, get_files_in_directory_recursively, get_job_data_path,
+        get_merklebot_data_path, upload_content,
+    },
 };
 
-pub async fn execute_launch(socket: Client, robot_job: RobotJob, jobs: Jobs) {
+pub async fn execute_launch(socket: Client, robot_job: RobotJob, agent: agent::Agent, jobs: Jobs) {
     let args = serde_json::from_str::<DockerLaunchArgs>(&robot_job.args).unwrap();
     info!("launching docker job {:?}", args);
     let docker_launch = DockerLaunch { args };
-    let robot_job_result = match docker_launch.execute(robot_job.clone(), jobs).await {
+    let robot_job_result = match docker_launch.execute(robot_job.clone(), agent, jobs).await {
         Ok(result) => {
             info!("job successfully executed");
             result
@@ -48,6 +53,7 @@ pub struct DockerLaunchArgs {
     pub container_name: String,
     pub custom_cmd: Option<String>,
     pub save_logs: Option<bool>,
+    pub store_data: Option<bool>,
     pub network_mode: String,
     pub ports: Vec<DockerMap>,
     pub volumes: Vec<DockerMap>,
@@ -68,6 +74,7 @@ impl DockerLaunch {
     pub async fn execute(
         &self,
         robot_job: RobotJob,
+        agent: agent::Agent,
         jobs: Jobs,
     ) -> Result<RobotJobResult, bollard::errors::Error> {
         info!("launching docker with image {}", self.args.image);
@@ -92,6 +99,24 @@ impl DockerLaunch {
             volumes.push(format!("{}:{}", volume_pair.key, volume_pair.value))
         }
 
+        match self.args.store_data {
+            Some(true) => {
+                // 1. create folder for the job
+                let create_job_dir_res = create_job_data_dir(&robot_job.id);
+                match create_job_dir_res {
+                    Ok(path) => {
+                        info!("Sharing dir {}", path);
+                        // 2. Share folder as volume
+                        volumes.push(format!("{}:{}", path, "/merklebot/job_data/"));
+                    }
+                    _ => {
+                        error!("Couldn't create shared dir for job {}", robot_job.id);
+                    }
+                }
+            }
+            _ => {}
+        }
+
         let mut config = bollard::container::Config::<&str> {
             image: Some(&self.args.image),
             env: Some(self.args.env.iter().map(|s| s as &str).collect()),
@@ -222,11 +247,38 @@ impl DockerLaunch {
             .await?;
 
         let robot_job_result = RobotJobResult {
-            job_id: robot_job.id,
+            job_id: robot_job.id.clone(),
             status: String::from("done"),
             logs: concatenated_logs,
         };
+        let job_data_path = get_job_data_path(&robot_job.id);
 
+        match &self.args.store_data {
+            Some(true) => {
+                match get_files_in_directory_recursively(&job_data_path) {
+                    //TODO: change to path
+                    Ok(paths) => {
+                        info!("{:?}", paths);
+                        for path in paths {
+                            let path_str = path.as_path().display().to_string();
+                            let key = path_str.replace(&get_merklebot_data_path(), "");
+                            upload_content(
+                                agent.robot_server_url.clone(),
+                                path,
+                                key,
+                                robot_job.id.clone(),
+                                agent.api_key.clone(),
+                            )
+                            .await;
+                        }
+                    }
+                    _ => {
+                        error!("Can't get resulting paths");
+                    }
+                }
+            }
+            _ => {}
+        }
         Ok(robot_job_result)
     }
 }