Repository: golemfactory/yagna (language: python)

Issue:
save a copy of exe-unit logs in logs dir
Why:
- Maybe Provider would like to fix himself, if something doesn't work

What:
- Log explicit path to ExeUnit logs
- Increase logs verbosity
- Remove top level ExeUnit log


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 2859c455bc..b406bb2bc4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1967,6 +1967,23 @@ dependencies = [
  "yansi",
 ]
 
+[[package]]
+name = "flexi_logger"
+version = "0.19.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "35627d78fcea84d52181369fe48fcfce56a2a18e29443dba23d4c24b650fb107"
+dependencies = [
+ "ansi_term 0.12.1",
+ "atty",
+ "chrono",
+ "glob",
+ "lazy_static",
+ "log",
+ "regex",
+ "rustversion",
+ "thiserror",
+]
+
 [[package]]
 name = "fnv"
 version = "1.0.7"
@@ -5152,7 +5169,7 @@ version = "0.5.1"
 source = "git+https://github.com/tworec/serial_test.git?branch=actix_rt_test#6c848541e89262f90ca693d9e1a84c6f3df0953b"
 dependencies = [
  "lazy_static",
- "parking_lot 0.10.2",
+ "parking_lot 0.11.1",
  "serial_test_derive",
 ]
 
@@ -6793,7 +6810,7 @@ dependencies = [
  "derive_more",
  "dotenv",
  "env_logger",
- "flexi_logger",
+ "flexi_logger 0.19.4",
  "futures 0.3.13",
  "graphene-sgx",
  "hex",
@@ -6839,7 +6856,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "chrono",
- "flexi_logger",
+ "flexi_logger 0.17.1",
  "log",
  "yansi",
 ]
diff --git a/agent/provider/src/execution/exeunit_instance.rs b/agent/provider/src/execution/exeunit_instance.rs
index cbcd353291..9499e75bee 100644
--- a/agent/provider/src/execution/exeunit_instance.rs
+++ b/agent/provider/src/execution/exeunit_instance.rs
@@ -27,6 +27,11 @@ impl ExeUnitInstance {
         log::debug!("Spawning args: {:?}", args);
         log::debug!("Spawning in: {:?}", working_dir);
 
+        log::info!(
+            "Exeunit log directory: {}",
+            working_dir.join("logs").display()
+        );
+
         let binary_path = ya_utils_path::normalize_path(&binary_path)
             .map_err(|e| anyhow!("Failed to spawn [{}]: {}", binary_path.display(), e))?;
 
@@ -34,6 +39,9 @@ impl ExeUnitInstance {
         command
             .args(args)
             .current_dir(working_dir)
+            .stdin(Stdio::null())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null())
             // new_process_group is a no-op on non-Unix systems
             .new_process_group();
 
diff --git a/agent/provider/src/execution/task_runner.rs b/agent/provider/src/execution/task_runner.rs
index 4d56d77f57..91567e06e8 100644
--- a/agent/provider/src/execution/task_runner.rs
+++ b/agent/provider/src/execution/task_runner.rs
@@ -2,7 +2,7 @@ use actix::prelude::*;
 use anyhow::{anyhow, bail, Error, Result};
 use chrono::{DateTime, Utc};
 use derive_more::Display;
-use futures::future::join_all;
+use futures::future::{join_all, select, Either};
 use futures::{FutureExt, TryFutureExt};
 use humantime;
 use log_derive::{logfn, logfn_inputs};
@@ -17,7 +17,7 @@ use structopt::StructOpt;
 use ya_agreement_utils::{AgreementView, OfferTemplate};
 use ya_client::activity::ActivityProviderApi;
 use ya_client::model::activity::provider_event::ProviderEventType;
-use ya_client::model::activity::{ActivityState, ProviderEvent, State};
+use ya_client::model::activity::{ActivityState, ProviderEvent, State, StatePair};
 use ya_std_utils::LogErr;
 use ya_utils_actix::actix_handler::ResultTypeGetter;
 use ya_utils_actix::actix_signal::{Signal, SignalSlot};
@@ -281,6 +281,30 @@ impl TaskRunner {
         let process = task.exeunit.get_process_handle();
         self.tasks.push(task);
 
+        // Log ExeUnit initialization message
+        let activity_id = msg.activity_id.clone();
+        let api = self.api.clone();
+        let proc = process.clone();
+
+        Arbiter::spawn(async move {
+            let mut finished = Box::pin(proc.wait_until_finished());
+            let mut monitor = StateMonitor::default();
+
+            loop {
+                match select(Box::pin(api.get_activity_state(&activity_id)), finished).await {
+                    Either::Left((result, fut)) => {
+                        finished = fut;
+
+                        if let Ok(state) = result {
+                            monitor.update(state.state);
+                        }
+                        monitor.sleep().await;
+                    }
+                    Either::Right(_) => break,
+                }
+            }
+        });
+
         // We need to discover that ExeUnit process finished.
         // We can't be sure that Requestor will send DestroyActivity.
         let myself = ctx.address();
@@ -810,3 +834,53 @@ impl Handler<Shutdown> for TaskRunner {
         ActorResponse::r#async(fut.into_actor(self))
     }
 }
+
+struct StateMonitor {
+    state: StatePair,
+    interval: Duration,
+}
+
+impl Default for StateMonitor {
+    fn default() -> Self {
+        StateMonitor {
+            state: StatePair(State::New, None),
+            interval: Self::INITIAL_INTERVAL,
+        }
+    }
+}
+
+impl StateMonitor {
+    const INITIAL_INTERVAL: Duration = Duration::from_millis(750);
+    const INTERVAL: Duration = Duration::from_secs(5);
+
+    fn update(&mut self, state: StatePair) {
+        match state {
+            StatePair(State::Initialized, None)
+            | StatePair(State::Deployed, _)
+            | StatePair(State::Ready, _) => {
+                if self.state.0 == State::New {
+                    log::info!("ExeUnit initialized");
+                    self.interval = Self::INTERVAL;
+                }
+            }
+            StatePair(State::Unresponsive, _) => {
+                if self.state.0 != State::Unresponsive {
+                    log::warn!("ExeUnit is unresponsive");
+                }
+            }
+            _ => {}
+        }
+
+        if self.state.0 == State::Unresponsive {
+            if state.0 != State::Unresponsive {
+                log::warn!("ExeUnit is now responsive");
+            }
+        }
+
+        self.state = state;
+    }
+
+    fn sleep(&self) -> impl Future<Output = ()> {
+        tokio::time::delay_for(self.interval)
+    }
+}
diff --git a/exe-unit/Cargo.toml b/exe-unit/Cargo.toml
index b52f0c373b..5782f2319b 100644
--- a/exe-unit/Cargo.toml
+++ b/exe-unit/Cargo.toml
@@ -45,7 +45,7 @@ chrono = "0.4.10"
 derivative = "2.1"
 derive_more = "0.99"
 dotenv = "0.15.0"
-flexi_logger = { version = "0.17", features = ["colors"] }
+flexi_logger = { version = "0.19", features = ["colors"] }
 futures = "0.3"
 graphene-sgx = { version = "0.3", optional = true }
 hex = "0.4.2"
diff --git a/exe-unit/src/bin.rs b/exe-unit/src/bin.rs
index 724b9df502..7f2d67042c 100644
--- a/exe-unit/src/bin.rs
+++ b/exe-unit/src/bin.rs
@@ -1,12 +1,13 @@
 use actix::{Actor, Addr, Arbiter, System};
 use anyhow::bail;
-use flexi_logger::{DeferredNow, Record};
 use std::convert::TryFrom;
 use std::path::PathBuf;
 use structopt::{clap, StructOpt};
+
 use ya_client_model::activity::ExeScriptCommand;
 use ya_core_model::activity;
 use ya_exe_unit::agreement::Agreement;
+use ya_exe_unit::logger::*;
 use ya_exe_unit::message::{GetState, GetStateResponse, Register};
 use ya_exe_unit::runtime::process::RuntimeProcess;
 use ya_exe_unit::service::metrics::MetricsService;
@@ -289,35 +290,11 @@ fn run() -> anyhow::Result<()> {
     Ok(())
 }
 
-pub fn colored_stderr_exeunit_prefixed_format(
-    w: &mut dyn std::io::Write,
-    now: &mut DeferredNow,
-    record: &Record,
-) -> Result<(), std::io::Error> {
-    write!(w, "{}", yansi::Color::Fixed(92).paint("[ExeUnit] "))?;
-    flexi_logger::colored_opt_format(w, now, record)
-}
-
-fn configure_logger(logger: flexi_logger::Logger) -> flexi_logger::Logger {
-    logger
-        .format(flexi_logger::colored_opt_format)
-        .duplicate_to_stderr(flexi_logger::Duplicate::Debug)
-        .format_for_stderr(colored_stderr_exeunit_prefixed_format)
-}
-
 fn main() {
-    let default_log_level = "info";
-    if configure_logger(flexi_logger::Logger::with_env_or_str(default_log_level))
-        .log_to_file()
-        .directory("logs")
-        .start()
-        .is_err()
-    {
-        configure_logger(flexi_logger::Logger::with_env_or_str(default_log_level))
-            .start()
-            .expect("Failed to initialize logging");
-        log::warn!("Switched to fallback logging method");
-    }
+    if let Err(error) = start_file_logger() {
+        start_logger().expect("Failed to start logging");
+        log::warn!("Using fallback logging due to an error: {:?}", error);
+    };
 
     std::process::exit(match run() {
         Ok(_) => 0,
@@ -327,11 +304,3 @@ fn main() {
         }
     })
 }
-
-#[cfg(test)]
-mod test {
-    #[test]
-    fn test_paint() {
-        println!("Some: {}", yansi::Color::Fixed(92).paint("violet text!"));
-    }
-}
diff --git a/exe-unit/src/lib.rs b/exe-unit/src/lib.rs
index ccc5a23523..8f2873bfac 100644
--- a/exe-unit/src/lib.rs
+++ b/exe-unit/src/lib.rs
@@ -33,6 +33,7 @@ pub mod agreement;
 pub mod crypto;
 pub mod error;
 mod handlers;
+pub mod logger;
 pub mod message;
 pub mod metrics;
 mod network;
diff --git a/exe-unit/src/logger.rs b/exe-unit/src/logger.rs
new file mode 100644
index 0000000000..006ac0c2c3
--- /dev/null
+++ b/exe-unit/src/logger.rs
@@ -0,0 +1,77 @@
+use chrono::{DateTime, Local};
+use flexi_logger::{DeferredNow, Record};
+
+const ENV_VAR_LOG_DIR: &'static str = "EXE_UNIT_LOG_DIR";
+const ENV_VAR_FILE_LOG_LEVEL: &'static str = "EXE_UNIT_FILE_LOG_LEVEL";
+
+const DEFAULT_LOG_LEVEL: &'static str = "info";
+const DEFAULT_FILE_LOG_LEVEL: &'static str = "debug";
+const DEFAULT_LOG_DIR: &'static str = "logs";
+const DEFAULT_LOG_FORMAT: &'static str = "%Y-%m-%dT%H:%M:%S%.3f%z";
+
+pub fn start_file_logger() -> anyhow::Result<flexi_logger::LoggerHandle> {
+    let log_dir = std::env::var(ENV_VAR_LOG_DIR).unwrap_or(DEFAULT_LOG_DIR.to_string());
+    let log_level =
+        std::env::var(ENV_VAR_FILE_LOG_LEVEL).unwrap_or(DEFAULT_FILE_LOG_LEVEL.to_string());
+
+    Ok(build_logger(Some(log_level))?
+        .log_to_file(flexi_logger::FileSpec::default().directory(log_dir))
+        .duplicate_to_stderr(log_tty_dup_level()?)
+        .start()?)
+}
+
+pub fn start_logger() -> anyhow::Result<flexi_logger::LoggerHandle> {
+    Ok(build_logger(Option::<String>::None)?.start()?)
+}
+
+fn build_logger<S: ToString>(log_level: Option<S>) -> anyhow::Result<flexi_logger::Logger> {
+    let level = match log_level {
+        Some(level) => level.to_string(),
+        None => std::env::var("RUST_LOG").unwrap_or(DEFAULT_LOG_LEVEL.to_string()),
+    };
+
+    Ok(flexi_logger::Logger::try_with_str(level)?
+        .format(log_format)
+        .format_for_stderr(flexi_logger::colored_opt_format))
+}
+
+fn log_tty_dup_level() -> anyhow::Result<flexi_logger::Duplicate> {
+    use flexi_logger::Duplicate;
+    use log::LevelFilter;
+
+    let level_filter = flexi_logger::LogSpecification::env_or_parse(DEFAULT_LOG_LEVEL)?
+        .module_filters()
+        .iter()
+        .find(|f| f.module_name.is_none())
+        .map(|f| f.level_filter)
+        .unwrap_or(LevelFilter::Off);
+
+    Ok(match level_filter {
+        LevelFilter::Off => Duplicate::None,
+        LevelFilter::Trace => Duplicate::Trace,
+        LevelFilter::Debug => Duplicate::Debug,
+        LevelFilter::Info => Duplicate::Info,
+        LevelFilter::Warn => Duplicate::Warn,
+        LevelFilter::Error => Duplicate::Error,
+    })
+}
+
+fn log_format(
+    w: &mut dyn std::io::Write,
+    now: &mut DeferredNow,
+    record: &Record,
+) -> Result<(), std::io::Error> {
+    //use DateTime::<Local> instead of DateTime::<UTC> to obtain local date
+    let local_date = DateTime::<Local>::from(*now.now());
+    //format date as following: 2020-08-27T07:56:22.348+02:00 (local date + time zone with milliseconds precision)
+    let date_format = local_date.format(DEFAULT_LOG_FORMAT);
+
+    write!(
+        w,
+        "[{} {:5} {}] {}",
+        date_format,
+        record.level(),
+        record.module_path().unwrap_or("<unnamed>"),
+        record.args()
+    )
+}


Test patch:
diff --git a/goth_tests/helpers/probe.py b/goth_tests/helpers/probe.py
index 9995c06552..a17a4aabe9 100644
--- a/goth_tests/helpers/probe.py
+++ b/goth_tests/helpers/probe.py
@@ -26,7 +26,7 @@ async def wait_for_agreement_approved(self):
     async def wait_for_exeunit_started(self):
         """Wait until the provider agent starts the exe-unit."""
         await self.provider_agent.wait_for_log(
-            r"(.*)\[ExeUnit\](.+)Supervisor initialized$"
+            r"(.*)ExeUnit initialized.*"
         )
 
     @step()