Merge pull request #174 from golemfactory/kek/progress

Reintroduce progress changes

model/Cargo.toml

@@ -17,6 +17,7 @@ sgx = ['secp256k1', 'openssl', 'hex', 'secp256k1/serde']
 bigdecimal = { version = "0.2", features = ["serde"]}
 chrono = { version = "0.4", features = ["serde"]}
 derive_more = "0.99"
+humantime-serde = "1.1"
 rand = "0.8"
 serde = { version = "1.0.146", features = ["derive"] }
 serde_with = { version = "3" }

model/src/activity.rs

@@ -21,7 +21,7 @@ pub use self::exe_script_command_result::{CommandOutput, CommandResult, ExeScrip
 pub use self::exe_script_command_state::ExeScriptCommandState;
 pub use self::exe_script_request::ExeScriptRequest;
 pub use self::provider_event::ProviderEvent;
-pub use self::runtime_event::{RuntimeEvent, RuntimeEventKind};
+pub use self::runtime_event::{CommandProgress, RuntimeEvent, RuntimeEventKind};
 #[cfg(feature = "sgx")]
 pub use self::sgx_credentials::SgxCredentials;
 

model/src/activity/exe_script_command.rs

@@ -11,6 +11,7 @@
 use crate::activity::ExeScriptCommandState;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::time::Duration;
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -21,6 +22,18 @@ pub enum ExeScriptCommand {
         net: Vec<Network>,
         #[serde(default)]
         hosts: HashMap<String, String>, // hostname -> IP
+
+        #[serde(default)]
+        hostname: Option<String>,
+
+        #[serde(default)]
+        volumes: Vec<String>,
+
+        #[serde(default)]
+        env: HashMap<String, String>,
+
+        #[serde(default)]
+        progress: Option<ProgressArgs>,
     },
     Start {
         #[serde(default)]
@@ -39,6 +52,8 @@ pub enum ExeScriptCommand {
         to: String,
         #[serde(flatten)]
         args: TransferArgs,
+        #[serde(default)]
+        progress: Option<ProgressArgs>,
     },
     Terminate {},
 }
@@ -107,6 +122,16 @@ pub struct TransferArgs {
     pub fileset: Option<FileSet>,
 }
 
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct ProgressArgs {
+    #[serde(default)]
+    #[serde(with = "humantime_serde")]
+    pub update_interval: Option<Duration>,
+    /// Number of bytes after which next progress event will be sent.
+    pub update_step: Option<usize>,
+}
+
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum FileSet {

model/src/activity/runtime_event.rs

@@ -47,6 +47,10 @@ impl RuntimeEvent {
     pub fn stderr(batch_id: String, idx: usize, out: CommandOutput) -> Self {
         Self::new(batch_id, idx, RuntimeEventKind::StdErr(out))
     }
+
+    pub fn progress(batch_id: String, idx: usize, progress: CommandProgress) -> Self {
+        Self::new(batch_id, idx, RuntimeEventKind::Progress(progress))
+    }
 }
 
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
@@ -61,4 +65,20 @@ pub enum RuntimeEventKind {
     },
     StdOut(CommandOutput),
     StdErr(CommandOutput),
+    Progress(CommandProgress),
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub struct CommandProgress {
+    /// Steps are counted starting from 0. That means that first step from 4-steps tasks
+    /// will report 0/4. Task is finished when counter reaches 4/4.
+    pub step: (usize, usize),
+    /// May contain additional arbitrary information about, what is happening with the task
+    /// like retrying transfer or that image was deployed from cache.
+    pub message: Option<String>,
+    /// Granular progress of currently executed step. The first element describes current
+    /// progress, the second the size of the whole task, which can be unknown.
+    pub progress: (u64, Option<u64>),
+    pub unit: Option<String>,
 }

specs/activity-api.yaml

@@ -486,7 +486,8 @@ components:
             "deploy": {
               "net": [{ "id": "id", "ip": "10.0.0.2", "mask": "255.255.0.0" }],
               "hosts": {"master": "10.0.0.1"},
-              "nodes": {"10.0.0.1": "0xdeadbeef"}
+              "nodes": {"10.0.0.1": "0xdeadbeef"},
+              "progress" : {"updateInterval": "300ms", "updateStep": null}
             }
         },
         {
@@ -500,7 +501,8 @@ components:
                 "to": "container:/input/file_in",
                 "format": "zip",
                 "depth": 2,
-                "fileset": [{"desc":"all images", "includes": ["*.jpg"], "excludes": ["db*.*"] }]
+                "fileset": [{"desc":"all images", "includes": ["*.jpg"], "excludes": ["db*.*"] }],
+                "progress" : {"updateInterval": null, "updateStep": 1048576}
             }
         },
         {
@@ -557,6 +559,8 @@ components:
           type: object
           additionalProperties:
             type: string
+        progress:
+          $ref: '#/components/schemas/ProgressArgs'
 
     DeployNetwork:
       type: object
@@ -572,6 +576,25 @@ components:
         mask:
           type: string
 
+    ProgressArgs:
+      type: object
+      description: Configuration of progress reporting.
+        Presence of this field in ExeUnitCommand indicates, that ExeUnit should send
+        '#/components/schemas/RuntimeEventKindProgress' events. If non of properties is set
+        ExeUnit will use default values.
+        Behavior when both properties are defined is ExeUnit specific.
+      properties:
+        update-interval:
+          type: string
+          description: Interval between progress reporting events expressed as
+            described in specification https://docs.rs/humantime/latest/humantime/fn.parse_duration.html
+        update-step:
+          type: number
+          format: int64
+          minimum: 1
+          description: Number of units (for example Bytes in case of transfer) after which next
+            progress event will be sent.
+
     StartCommand:
       allOf:
         - $ref: '#/components/schemas/ExeScriptCommand'
@@ -616,6 +639,8 @@ components:
           type: array
           items:
             $ref: '#/components/schemas/FileSet'
+        progress:
+          $ref: '#/components/schemas/ProgressArgs'
 
     FileSet:
       properties:
@@ -842,15 +867,65 @@ components:
               $ref: '#/components/schemas/CommandOutput'
 
     RuntimeEventKindFinishedBody:
-      type: object
-      required:
-        - returnCode
-      properties:
-        returnCode:
-          type: integer
-          format: int32
-        message:
-          type: string
+      allOf:
+        - $ref: '#/components/schemas/RuntimeEventKind'
+        - type: object
+          required:
+            - returnCode
+          properties:
+            returnCode:
+              type: integer
+              format: int32
+            message:
+              type: string
+
+    RuntimeEventKindProgress:
+      allOf:
+        - $ref: '#/components/schemas/RuntimeEventKind'
+        - type: object
+          description: Reports progress of currently executed command. This event will be sent only,
+            if `progress` field was set in `deploy` or `transfer` command.
+          required:
+            - step
+          properties:
+            step:
+              $ref: '#/components/schemas/ProgressStep'
+            message:
+              type: string
+              description: May contain additional arbitrary information, what is happening with the task,
+                for example "retrying transfer" or "image deployed from cache".
+                Content of this field is ExeUnit specific.
+            progress:
+              $ref: '#/components/schemas/ProgressDetails'
+            unit:
+              type: string
+              description: Units in which `progress` field is expressed. This should be human readable
+                string for displaying in UI. List of possible values is ExeUnit specific.
+
+    ProgressStep:
+      description: Can be used if single ExeUnit command is executing multiple steps. Number of steps is
+        Exeunit specific.
+
+        Steps are counted starting from 0. That means that first step from 4-steps task
+        will report 0/4. Task is finished when counter reaches 4/4.
+      type: array
+      items:
+        type: integer
+        format: int64
+      minItems: 2
+      maxItems: 2
+      minimum: 0
+
+    ProgressDetails:
+      description: Granular progress of currently executed step. The first element describes current
+        progress, the second the size of the whole task, which can be unknown.
+      type: array
+      items:
+        type: integer
+        format: int64
+      minItems: 1
+      maxItems: 2
+      minimum: 0
 
     CommandOutput:
       type: object

src/activity/requestor/control.rs

@@ -69,7 +69,12 @@ impl ActivityRequestorControlApi {
     /// Destroys given Activity.
     pub async fn destroy_activity(&self, activity_id: &str) -> Result<()> {
         let uri = url_format!("activity/{activity_id}");
-        self.client.delete(&uri).send().json().await?;
+
+        // Specify serialization target because of a rustc bug falsely
+        // emitting dependency_on_unit_never_type_fallback.
+        // If some time has passed try removing ::<()> and see if it still
+        // emits a warning.
+        self.client.delete(&uri).send().json::<()>().await?;
         Ok(())
     }