Merge pull request #241 from AdExNetwork/issue-105-validator_worker-logging

Issue #105 validator worker logging

Author: samparsky

.travis.yml

@@ -1,6 +1,7 @@
 language: rust
 rust:
   - stable
+os: linux
 # add nodejs for ganche-cli
 node_js: "12.12.0"
 # Need to cache the whole `.cargo` directory to keep .crates.toml for
@@ -13,15 +14,14 @@ env:
     - CARGO_MAKE_RUN_CHECK_FORMAT="true"
     - CARGO_MAKE_RUN_CLIPPY="true"
 services:
-  - redis-server
-matrix:
-  fast_finish: true
+  - redis
 
 stages:
   - test
   - deploy
 
 jobs:
+  fast_finish: true
   include:
     - stage: test
       script:

Cargo.lock

@@ -14,6 +14,7 @@ Components:
 ## Local & Testing setup
 
 #### Linux
+- `build-essentials` is required to build the project (error: `linker ``cc`` not found`)
 - The crate `openssl-sys` requires `libssl-dev` and `pkg-config` for Ubuntu.
 
 ### Run Postgres

README.md

@@ -45,6 +45,12 @@ impl FromHex for ChannelId {
     }
 }
 
+impl fmt::Display for ChannelId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "0x{}", hex::encode(self.0))
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct Channel {

primitives/src/channel.rs

@@ -8,12 +8,11 @@ use hyper::service::{make_service_fn, service_fn};
 use hyper::{Error, Server};
 use primitives::adapter::{Adapter, DummyAdapterOptions, KeystoreOptions};
 use primitives::config::configuration;
-use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator};
 use primitives::util::tests::prep_db::{AUTH, IDS};
 use primitives::ValidatorId;
 use sentry::db::{postgres_connection, redis_connection, setup_migrations};
 use sentry::Application;
-use slog::{error, info, o, Drain, Logger};
+use slog::{error, info, Logger};
 use std::convert::TryFrom;
 
 const DEFAULT_PORT: u16 = 8005;
@@ -146,11 +145,14 @@ async fn run<A: Adapter + 'static>(app: Application<A>, _clustered: bool, port:
     let server = Server::bind(&addr).serve(make_service);
 
     if let Err(e) = server.await {
-        error!(&logger, "server error: {}", e);
+        error!(&logger, "server error: {}", e; "main" => "run");
     }
 }
 
 fn logger() -> Logger {
+    use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator};
+    use slog::{o, Drain};
+
     let decorator = TermDecorator::new().build();
     let drain = PrefixedCompactFormat::new("sentry", decorator).fuse();
     let drain = Async::new(drain).build().fuse();

sentry/src/main.rs

@@ -111,6 +111,6 @@ async fn cache(
         .query_async::<_, ()>(&mut redis.clone())
         .await
     {
-        error!(&logger, "server error: {}", err);
+        error!(&logger, "Server error: {}", err; "module" => "analytics-cache");
     }
 }

sentry/src/routes/analytics.rs

@@ -18,6 +18,8 @@ num-traits = "0.2.0"
 # To/From Hex
 hex = "0.3.2"
 byteorder = "1.3"
+# Logging
+slog = { version = "^2.5.2" , features = ["max_level_trace"] }
 # Futures
 futures = "0.3"
 # Concurrency

validator_worker/Cargo.toml

@@ -47,6 +47,7 @@ mod test {
     use primitives::config::configuration;
     use primitives::util::tests::prep_db::{AUTH, DUMMY_CHANNEL, IDS};
     use primitives::{BalancesMap, Channel};
+    use slog::{o, Discard, Logger};
 
     fn setup_iface(channel: &Channel) -> SentryApi<DummyAdapter> {
         let adapter_options = DummyAdapterOptions {
@@ -56,8 +57,9 @@ mod test {
         };
         let config = configuration("development", None).expect("Dev config should be available");
         let dummy_adapter = DummyAdapter::init(adapter_options, &config);
+        let logger = Logger::root(Discard, o!());
 
-        SentryApi::init(dummy_adapter, &channel, &config, false).expect("should succeed")
+        SentryApi::init(dummy_adapter, &channel, &config, logger).expect("should succeed")
     }
 
     #[test]

validator_worker/src/lib.rs

@@ -15,6 +15,7 @@ use primitives::adapter::{Adapter, DummyAdapterOptions, KeystoreOptions};
 use primitives::config::{configuration, Config};
 use primitives::util::tests::prep_db::{AUTH, IDS};
 use primitives::{Channel, SpecValidator, ValidatorId};
+use slog::{error, Logger};
 use validator_worker::error::ValidatorWorker as ValidatorWorkerError;
 use validator_worker::{all_channels, follower, leader, SentryApi};
 
@@ -105,12 +106,14 @@ fn main() -> Result<(), Box<dyn Error>> {
         _ => panic!("We don't have any other adapters implemented yet!"),
     };
 
+    let logger = logger();
+
     match adapter {
         AdapterTypes::EthereumAdapter(ethadapter) => {
-            run(is_single_tick, &sentry_url, &config, *ethadapter)
+            run(is_single_tick, &sentry_url, &config, *ethadapter, &logger)
         }
         AdapterTypes::DummyAdapter(dummyadapter) => {
-            run(is_single_tick, &sentry_url, &config, *dummyadapter)
+            run(is_single_tick, &sentry_url, &config, *dummyadapter, &logger)
         }
     }
 }
@@ -120,6 +123,7 @@ fn run<A: Adapter + 'static>(
     sentry_url: &str,
     config: &Config,
     mut adapter: A,
+    logger: &Logger,
 ) -> Result<(), Box<dyn Error>> {
     // unlock adapter
     adapter.unlock()?;
@@ -134,29 +138,29 @@ fn run<A: Adapter + 'static>(
     let mut rt = Runtime::new()?;
 
     if is_single_tick {
-        rt.block_on(iterate_channels(args));
+        rt.block_on(iterate_channels(args, &logger));
     } else {
-        rt.block_on(infinite(args));
+        rt.block_on(infinite(args, &logger));
     }
 
     Ok(())
 }
 
-async fn infinite<A: Adapter + 'static>(args: Args<A>) {
+async fn infinite<A: Adapter + 'static>(args: Args<A>, logger: &Logger) {
     loop {
         let arg = args.clone();
         let delay_future = delay_for(Duration::from_secs(arg.config.wait_time as u64));
-        let _result = join(iterate_channels(arg), delay_future).await;
+        let _result = join(iterate_channels(arg, logger), delay_future).await;
     }
 }
 
-async fn iterate_channels<A: Adapter + 'static>(args: Args<A>) {
+async fn iterate_channels<A: Adapter + 'static>(args: Args<A>, logger: &Logger) {
     let result = all_channels(&args.sentry_url, args.adapter.whoami()).await;
 
     let channels = match result {
         Ok(channels) => channels,
         Err(e) => {
-            eprintln!("Failed to get channels {}", e);
+            error!(logger, "Failed to get channels {}", &e; "main" => "iterate_channels");
             return;
         }
     };
@@ -166,29 +170,28 @@ async fn iterate_channels<A: Adapter + 'static>(args: Args<A>) {
     let tick = try_join_all(
         channels
             .into_iter()
-            .map(|channel| validator_tick(args.adapter.clone(), channel, &args.config)),
+            .map(|channel| validator_tick(args.adapter.clone(), channel, &args.config, logger)),
     )
     .await;
 
     if let Err(e) = tick {
-        eprintln!("An occurred while processing channels {}", e);
+        error!(logger, "An occurred while processing channels {}", &e; "main" => "iterate_channels");
     }
 
     if channels_size >= args.config.max_channels as usize {
-        eprintln!(
-            "WARNING: channel limit cfg.MAX_CHANNELS={} reached",
-            args.config.max_channels
-        )
+        error!(logger, "WARNING: channel limit cfg.MAX_CHANNELS={} reached", &args.config.max_channels; "main" => "iterate_channels");
     }
 }
 
 async fn validator_tick<A: Adapter + 'static>(
     adapter: A,
     channel: Channel,
     config: &Config,
+    logger: &Logger,
 ) -> Result<(), ValidatorWorkerError> {
     let whoami = adapter.whoami().clone();
-    let sentry = SentryApi::init(adapter, &channel, &config, true)?;
+    // Cloning the `Logger` is cheap, see documentation for more info
+    let sentry = SentryApi::init(adapter, &channel, &config, logger.clone())?;
     let duration = Duration::from_secs(config.validator_tick_timeout as u64);
 
     match channel.spec.validators.find(&whoami) {
@@ -210,3 +213,14 @@ async fn validator_tick<A: Adapter + 'static>(
     };
     Ok(())
 }
+
+fn logger() -> Logger {
+    use primitives::util::logging::{Async, PrefixedCompactFormat, TermDecorator};
+    use slog::{o, Drain};
+
+    let decorator = TermDecorator::new().build();
+    let drain = PrefixedCompactFormat::new("validator_worker", decorator).fuse();
+    let drain = Async::new(drain).build().fuse();
+
+    Logger::root(drain, o!())
+}

validator_worker/src/main.rs

@@ -8,6 +8,7 @@ use primitives::BalancesMap;
 
 use crate::core::events::merge_aggrs;
 use crate::sentry_interface::SentryApi;
+use slog::info;
 
 pub type Result = std::result::Result<(BalancesMap, Option<Accounting>), Box<dyn Error>>;
 
@@ -28,7 +29,12 @@ pub async fn tick<A: Adapter + 'static>(iface: &SentryApi<A>) -> Result {
         .await?;
 
     if !aggrs.events.is_empty() {
-        // TODO: Log the merge
+        info!(
+            iface.logger,
+            "channel {}: processing {} event aggregates",
+            iface.channel.id,
+            aggrs.events.len()
+        );
         let (balances, new_accounting) = merge_aggrs(&accounting, &aggrs.events, &iface.channel)?;
 
         let message_types = MessageTypes::Accounting(new_accounting.clone());