UI feedback for docker pull

kamu-data · Nov 17, 2020 · e7c6a0e · e7c6a0e
1 parent 8741ec9
commit e7c6a0e
Show file tree

Hide file tree

Showing 19 changed files with 244 additions and 65 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,7 +4,7 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## Unreleased
+## [0.36.0] - 2020-11-16
 ### Changed
 - Engine errors will not list all relevant log files
 - UI improvements

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/kamu-cli/Cargo.toml b/kamu-cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "kamu-cli"
-version = "0.35.0"
+version = "0.36.0"
 description = "Decentralized data management tool"
 authors = ["Sergii Mikhtoniuk <mikhtoniuk@gmail.com>"]
 license = "MPL-2.0"

diff --git a/kamu-cli/src/commands/notebook_command.rs b/kamu-cli/src/commands/notebook_command.rs
@@ -1,5 +1,6 @@
 use super::{Command, Error};
 use crate::output::OutputConfig;
+use kamu::domain::PullImageListener;
 use kamu::infra::explore::*;
 use kamu::infra::*;
 
@@ -65,6 +66,9 @@ impl Command for NotebookCommand {
             .collect::<Result<Vec<_>, _>>()?;
 
         let spinner = if self.output_config.verbosity_level == 0 {
+            let mut pull_progress = PullImageProgress { progress_bar: None };
+            NotebookServerImpl::ensure_images(&mut pull_progress);
+
             let s = indicatif::ProgressBar::new_spinner();
             s.set_style(
                 indicatif::ProgressStyle::default_spinner().template("{spinner:.cyan} {msg}"),
@@ -99,3 +103,21 @@ impl Command for NotebookCommand {
         Ok(())
     }
 }
+
+struct PullImageProgress {
+    progress_bar: Option<indicatif::ProgressBar>,
+}
+
+impl PullImageListener for PullImageProgress {
+    fn begin(&mut self, image: &str) {
+        let s = indicatif::ProgressBar::new_spinner();
+        s.set_style(indicatif::ProgressStyle::default_spinner().template("{spinner:.cyan} {msg}"));
+        s.set_message(&format!("Pulling docker image {}", image));
+        s.enable_steady_tick(100);
+        self.progress_bar = Some(s);
+    }
+
+    fn success(&mut self) {
+        self.progress_bar = None;
+    }
+}
diff --git a/kamu-cli/src/commands/pull_command.rs b/kamu-cli/src/commands/pull_command.rs
@@ -397,13 +397,33 @@ impl IngestListener for PrettyIngestProgress {
                 console::style("Failed to update root dataset").red(),
             ));
     }
+
+    fn get_pull_image_listener(&mut self) -> Option<&mut dyn PullImageListener> {
+        Some(self)
+    }
+}
+
+impl PullImageListener for PrettyIngestProgress {
+    fn begin(&mut self, image: &str) {
+        // This currently happens during the Read stage
+        self.curr_progress.set_message(&Self::spinner_message(
+            &self.dataset_id,
+            IngestStage::Read as u32,
+            format!("Pulling engine image {}", image),
+        ));
+    }
+
+    fn success(&mut self) {
+        self.curr_progress.finish();
+        self.on_stage_progress(self.stage, 0, 0);
+    }
 }
 
 ///////////////////////////////////////////////////////////////////////////////
 
 struct PrettyTransformProgress {
     dataset_id: DatasetIDBuf,
-    //multi_progress: Arc<indicatif::MultiProgress>,
+    multi_progress: Arc<indicatif::MultiProgress>,
     curr_progress: indicatif::ProgressBar,
 }
 
@@ -416,7 +436,7 @@ impl PrettyTransformProgress {
                 0,
                 "Applying derivative transformations",
             ))),
-            //multi_progress: multi_progress,
+            multi_progress: multi_progress,
         }
     }
 
@@ -462,4 +482,29 @@ impl TransformListener for PrettyTransformProgress {
                 console::style("Failed to update derivative dataset").red(),
             ));
     }
+
+    fn get_pull_image_listener(&mut self) -> Option<&mut dyn PullImageListener> {
+        Some(self)
+    }
+}
+
+impl PullImageListener for PrettyTransformProgress {
+    fn begin(&mut self, image: &str) {
+        self.curr_progress.set_message(&Self::spinner_message(
+            &self.dataset_id,
+            0,
+            format!("Pulling engine image {}", image),
+        ));
+    }
+
+    fn success(&mut self) {
+        self.curr_progress.finish();
+        self.curr_progress = self
+            .multi_progress
+            .add(Self::new_spinner(&Self::spinner_message(
+                &self.dataset_id,
+                0,
+                "Applying derivative transformations",
+            )));
+    }
 }
diff --git a/kamu-cli/src/commands/sql_shell_command.rs b/kamu-cli/src/commands/sql_shell_command.rs
@@ -1,5 +1,6 @@
 use super::{Command, Error};
 use crate::output::*;
+use kamu::domain::PullImageListener;
 use kamu::infra::explore::*;
 use kamu::infra::*;
 
@@ -34,6 +35,9 @@ impl SqlShellCommand {
 impl Command for SqlShellCommand {
     fn run(&mut self) -> Result<(), Error> {
         let spinner = if self.output_config.verbosity_level == 0 {
+            let mut pull_progress = PullImageProgress { progress_bar: None };
+            SqlShellImpl::ensure_images(&mut pull_progress);
+
             let s = indicatif::ProgressBar::new_spinner();
             s.set_style(
                 indicatif::ProgressStyle::default_spinner().template("{spinner:.cyan} {msg}"),
@@ -64,3 +68,21 @@ impl Command for SqlShellCommand {
         Ok(())
     }
 }
+
+struct PullImageProgress {
+    progress_bar: Option<indicatif::ProgressBar>,
+}
+
+impl PullImageListener for PullImageProgress {
+    fn begin(&mut self, image: &str) {
+        let s = indicatif::ProgressBar::new_spinner();
+        s.set_style(indicatif::ProgressStyle::default_spinner().template("{spinner:.cyan} {msg}"));
+        s.set_message(&format!("Pulling docker image {}", image));
+        s.enable_steady_tick(100);
+        self.progress_bar = Some(s);
+    }
+
+    fn success(&mut self) {
+        self.progress_bar = None;
+    }
+}
diff --git a/kamu-core-test/Cargo.toml b/kamu-core-test/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "kamu-test"
-version = "0.35.0"
+version = "0.36.0"
 authors = ["Sergii Mikhtoniuk <mikhtoniuk@gmail.com>"]
 edition = "2018"
 

diff --git a/kamu-core/Cargo.toml b/kamu-core/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "kamu"
-version = "0.35.0"
+version = "0.36.0"
 authors = ["Sergii Mikhtoniuk <mikhtoniuk@gmail.com>"]
 edition = "2018"
 

diff --git a/kamu-core/src/domain/engine.rs b/kamu-core/src/domain/engine.rs
@@ -16,6 +16,14 @@ pub trait Engine {
     fn transform(&self, request: ExecuteQueryRequest) -> Result<ExecuteQueryResponse, EngineError>;
 }
 
+pub trait PullImageListener {
+    fn begin(&mut self, _image: &str) {}
+    fn success(&mut self) {}
+}
+
+pub struct NullPullImageListener;
+impl PullImageListener for NullPullImageListener {}
+
 ///////////////////////////////////////////////////////////////////////////////
 // Request / Response DTOs
 ///////////////////////////////////////////////////////////////////////////////

diff --git a/kamu-core/src/domain/ingest_service.rs b/kamu-core/src/domain/ingest_service.rs
@@ -1,4 +1,4 @@
-use super::EngineError;
+use super::{EngineError, PullImageListener};
 use opendatafabric::{DatasetID, DatasetIDBuf, Sha3_256};
 
 use std::backtrace::Backtrace;
@@ -75,6 +75,10 @@ pub trait IngestListener: Send {
     fn success(&mut self, _result: &IngestResult) {}
     fn uncacheable(&mut self) {}
     fn error(&mut self, _error: &IngestError) {}
+
+    fn get_pull_image_listener(&mut self) -> Option<&mut dyn PullImageListener> {
+        None
+    }
 }
 
 pub struct NullIngestListener;

diff --git a/kamu-core/src/domain/transform_service.rs b/kamu-core/src/domain/transform_service.rs
@@ -1,4 +1,4 @@
-use super::EngineError;
+use super::{EngineError, PullImageListener};
 use opendatafabric::{DatasetID, DatasetIDBuf, Sha3_256};
 
 use std::backtrace::Backtrace;
@@ -37,6 +37,10 @@ pub trait TransformListener: Send {
     fn begin(&mut self) {}
     fn success(&mut self, _result: &TransformResult) {}
     fn error(&mut self, _error: &TransformError) {}
+
+    fn get_pull_image_listener(&mut self) -> Option<&mut dyn PullImageListener> {
+        None
+    }
 }
 
 pub struct NullTransformListener;

diff --git a/kamu-core/src/infra/engine/engine_factory.rs b/kamu-core/src/infra/engine/engine_factory.rs
@@ -1,16 +1,21 @@
 use crate::domain::*;
+use crate::infra::utils::docker_client::DockerClient;
 use crate::infra::utils::docker_images;
 use crate::infra::*;
 
 use super::engine_flink::*;
 use super::engine_spark::*;
 
 use slog::{o, Logger};
+use std::collections::HashSet;
+use std::process::Stdio;
 use std::sync::{Arc, Mutex};
 
 pub struct EngineFactory {
     spark_engine: Arc<Mutex<SparkEngine>>,
     flink_engine: Arc<Mutex<FlinkEngine>>,
+    docker_client: DockerClient,
+    known_images: HashSet<String>,
 }
 
 impl EngineFactory {
@@ -26,14 +31,47 @@ impl EngineFactory {
                 workspace_layout,
                 logger.new(o!("engine" => "flink")),
             ))),
+            docker_client: DockerClient::new(),
+            known_images: HashSet::new(),
         }
     }
 
-    pub fn get_engine(&mut self, engine_id: &str) -> Result<Arc<Mutex<dyn Engine>>, EngineError> {
-        match engine_id {
-            "spark" => Ok(self.spark_engine.clone()),
-            "flink" => Ok(self.flink_engine.clone()),
+    pub fn get_engine(
+        &mut self,
+        engine_id: &str,
+        maybe_listener: Option<&mut dyn PullImageListener>,
+    ) -> Result<Arc<Mutex<dyn Engine>>, EngineError> {
+        let (engine, image) = match engine_id {
+            "spark" => Ok((
+                self.spark_engine.clone() as Arc<Mutex<dyn Engine>>,
+                docker_images::SPARK,
+            )),
+            "flink" => Ok((
+                self.flink_engine.clone() as Arc<Mutex<dyn Engine>>,
+                docker_images::FLINK,
+            )),
             _ => Err(EngineError::not_found(engine_id)),
+        }?;
+
+        if !self.known_images.contains(image) {
+            if !self.docker_client.has_image(image) {
+                let mut null_listener = NullPullImageListener;
+                let listener = maybe_listener.unwrap_or(&mut null_listener);
+
+                listener.begin(image);
+
+                // TODO: Return better errors
+                self.docker_client
+                    .pull_cmd(image)
+                    .stdout(Stdio::null())
+                    .stderr(Stdio::null())
+                    .status()?;
+
+                listener.success();
+            }
+            self.known_images.insert(image.to_owned());
         }
+
+        Ok(engine)
     }
 }