Reworking sentry (#154)

* update toolchain to nightly and fix `cargo clippy` installation error

* fix clippy errors

* update to `.await` and fix cargo errors/warnings

* clean up some files

* Clean up some files and some not needed dependencies

* update .travis.yml to use latest nightly

* follower - merge fix for `.await`

* fix cargo fmt

* add rust-toolchain for sentry only, until `.await` is merged

* clean up `sentry`

* make sentry build again

* cargo-make 0.22.2 fixes the installation of clippy so remove the custom task override

* update `futures-preview` to `alpha.19`

* sentry - add the `adapter` along side `futures-preview`, `hyper`, `regex` and `tokio`

* sentry - add the cli options, load config and make a 404 Not found response.

* sentry - add `serde` & `serde_urlencoded`

* sentry - routes - channel - list channel WIP

* sentry - move `fn not_found` & `fn bad_request` to lib

* sentry - Cargo.toml - add `unstable-stream` feature for `hyper`

* sentry - more to channels list and channel create endpoints

* sentry - fix some issues and clean up code

* sentry - Cargo.toml - add `chrono`

* primitives - remove channel::ChannelListParams

* sentry - ChannelListQuery

* sentry - better error response handling & move channel list & create code to own modules

* sentry - Fix changes required because of the merge & `clippy` warnings

* sentry - remove `rust-toolchain`

* fix `rustfmt`

* improve `Adapter`, `EthereumAdapter` & `DummyAdapter`

* sentry - Application struct

* Fix `EthereumAdapter` and fix `clippy` warnings

* fix import

* primitives - big_num - fix `clippy` warnings about `to_owned()`

* primitives - add `slog` and impl own PrefixCompactFormat

* primitives - re-export `slog_async::Async` & `slog_term::TermDecorator`

* sentry - working logging for the `Application`:
- use the re-exported from `primitives` `slog` structs & traits
- Cargo.toml - remove `slog-term` & `slog-async` since we use the re-exported once

* sentry - use `error!` for server error & suppress `clustered` warning for not using it now

* Use patch instead of `git` url & branch in the `primitives` `Cargo.toml`

* sentry - add listening to port on `Application::run`

* sentry - add missing `info` import

* Fix some merge problems and re-add the `Send` trait bound for the `Adapter`

Author: elpiel

Cargo.lock

@@ -4,6 +4,9 @@ members = [
     "adapter",
     "primitives",
     "validator_worker",
-    # "sentry",
+    "sentry",
     "adview-manager",
-]
+]
+
+[patch.crates-io]
+slog-term = { git = "https://github.com/slog-rs/term", branch = "master"}

Cargo.toml

@@ -1,4 +1,4 @@
-use primitives::adapter::{Adapter, AdapterError, AdapterOptions, AdapterResult, Session};
+use primitives::adapter::{Adapter, AdapterError, AdapterResult, DummyAdapterOptions, Session};
 use primitives::channel_validator::ChannelValidator;
 use primitives::config::Config;
 use primitives::Channel;
@@ -18,31 +18,18 @@ pub struct DummyAdapter {
 // check if a channel is valid
 impl ChannelValidator for DummyAdapter {}
 
-impl Adapter for DummyAdapter {
-    type Output = DummyAdapter;
-
-    fn init(opts: AdapterOptions, config: &Config) -> AdapterResult<DummyAdapter> {
-        let (identity, authorization_tokens, session_tokens) = match opts {
-            AdapterOptions::DummAdapter {
-                dummy_identity,
-                dummy_auth,
-                dummy_auth_tokens,
-            } => (dummy_identity, dummy_auth, dummy_auth_tokens),
-            _ => {
-                return Err(AdapterError::Configuration(
-                    "dummy_identity, dummy_auth, dummy_auth_tokens required".to_string(),
-                ))
-            }
-        };
-
-        Ok(Self {
-            identity,
+impl DummyAdapter {
+    pub fn init(opts: DummyAdapterOptions, config: &Config) -> Self {
+        Self {
+            identity: opts.dummy_identity,
             config: config.to_owned(),
-            session_tokens,
-            authorization_tokens,
-        })
+            session_tokens: opts.dummy_auth_tokens,
+            authorization_tokens: opts.dummy_auth,
+        }
     }
+}
 
+impl Adapter for DummyAdapter {
     fn unlock(&mut self) -> AdapterResult<()> {
         Ok(())
     }

adapter/src/dummy.rs

@@ -4,7 +4,7 @@ use ethkey::{public_to_address, recover, verify_address, Address, Message, Passw
 use ethstore::SafeAccount;
 use lazy_static::lazy_static;
 use primitives::{
-    adapter::{Adapter, AdapterError, AdapterOptions, AdapterResult, Session},
+    adapter::{Adapter, AdapterError, AdapterResult, KeystoreOptions, Session},
     channel_validator::ChannelValidator,
     config::Config,
     Channel,
@@ -49,22 +49,9 @@ pub struct EthereumAdapter {
 // check if a channel is valid
 impl ChannelValidator for EthereumAdapter {}
 
-impl Adapter for EthereumAdapter {
-    type Output = EthereumAdapter;
-
-    fn init(opts: AdapterOptions, config: &Config) -> AdapterResult<EthereumAdapter> {
-        let (keystore_file, keystore_pwd) = match opts {
-            AdapterOptions::EthereumAdapter(keystore_opts) => {
-                (keystore_opts.keystore_file, keystore_opts.keystore_pwd)
-            }
-            _ => {
-                return Err(AdapterError::Configuration(
-                    "Missing keystore json file or password".to_string(),
-                ))
-            }
-        };
-
-        let keystore_contents = fs::read_to_string(&keystore_file)
+impl EthereumAdapter {
+    pub fn init(opts: KeystoreOptions, config: &Config) -> AdapterResult<EthereumAdapter> {
+        let keystore_contents = fs::read_to_string(&opts.keystore_file)
             .map_err(|_| map_error("Invalid keystore location provided"))?;
 
         let keystore_json: Value = serde_json::from_str(&keystore_contents)
@@ -82,14 +69,16 @@ impl Adapter for EthereumAdapter {
         Ok(Self {
             address,
             keystore_json,
-            keystore_pwd: keystore_pwd.into(),
+            keystore_pwd: opts.keystore_pwd.into(),
             session_tokens: HashMap::new(),
             authorization_tokens: HashMap::new(),
             wallet: None,
             config: config.to_owned(),
         })
     }
+}
 
+impl Adapter for EthereumAdapter {
     fn unlock(&mut self) -> AdapterResult<()> {
         let account = SafeAccount::from_file(
             serde_json::from_value(self.keystore_json.clone())
@@ -414,9 +403,8 @@ mod test {
             keystore_file: "./test/resources/keystore.json".to_string(),
             keystore_pwd: "adexvalidator".to_string(),
         };
-        let adapter_options = AdapterOptions::EthereumAdapter(keystore_options);
 
-        EthereumAdapter::init(adapter_options, &config).expect("should init ethereum adapter")
+        EthereumAdapter::init(keystore_options, &config).expect("should init ethereum adapter")
     }
 
     #[test]
@@ -473,8 +461,7 @@ mod test {
         let wallet = eth_adapter.wallet.clone();
         let response = ewt_sign(&wallet.unwrap(), &eth_adapter.keystore_pwd, &payload)
             .expect("failed to generate ewt signature");
-        let expected =
-            "eyJ0eXBlIjoiSldUIiwiYWxnIjoiRVRIIn0.eyJpZCI6ImF3ZXNvbWVWYWxpZGF0b3IiLCJlcmEiOjEwMDAwMCwiYWRkcmVzcyI6IjB4MmJEZUFGQUU1Mzk0MDY2OURhQTZGNTE5MzczZjY4NmMxZjNkMzM5MyJ9.gGw_sfnxirENdcX5KJQWaEt4FVRvfEjSLD4f3OiPrJIltRadeYP2zWy9T2GYcK5xxD96vnqAw4GebAW7rMlz4xw";
+        let expected = "eyJ0eXBlIjoiSldUIiwiYWxnIjoiRVRIIn0.eyJpZCI6ImF3ZXNvbWVWYWxpZGF0b3IiLCJlcmEiOjEwMDAwMCwiYWRkcmVzcyI6IjB4MmJEZUFGQUU1Mzk0MDY2OURhQTZGNTE5MzczZjY4NmMxZjNkMzM5MyJ9.gGw_sfnxirENdcX5KJQWaEt4FVRvfEjSLD4f3OiPrJIltRadeYP2zWy9T2GYcK5xxD96vnqAw4GebAW7rMlz4xw";
         assert_eq!(response, expected, "generated wrong ewt signature");
 
         let expected_verification_response = r#"VerifyPayload { from: "0x2bDeAFAE53940669DaA6F519373f686c1f3d3393", payload: Payload { id: "awesomeValidator", era: 100000, address: Some("0x2bDeAFAE53940669DaA6F519373f686c1f3d3393"), identity: None } }"#;

adapter/src/ethereum.rs

@@ -12,6 +12,10 @@ serde = { version = "^1.0", features = ['derive'] }
 serde_json = "1.0"
 serde-hex = "0.1.0"
 toml = "0.5"
+# Logging
+slog = { version = "^2.5.2" , features = ["max_level_trace"] }
+slog-term = "^2.4.1"
+slog-async = "^2.3.0"
 # Domain
 chrono = { version = "0.4", features = ["serde"] }
 time = "0.1.42"

primitives/Cargo.toml

@@ -1,5 +1,5 @@
 use crate::channel_validator::ChannelValidator;
-use crate::{Channel, Config};
+use crate::Channel;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
@@ -33,14 +33,10 @@ impl fmt::Display for AdapterError {
     }
 }
 
-#[derive(Debug, Clone)]
-pub enum AdapterOptions {
-    DummAdapter {
-        dummy_identity: String,
-        dummy_auth: HashMap<String, String>,
-        dummy_auth_tokens: HashMap<String, String>,
-    },
-    EthereumAdapter(KeystoreOptions),
+pub struct DummyAdapterOptions {
+    pub dummy_identity: String,
+    pub dummy_auth: HashMap<String, String>,
+    pub dummy_auth_tokens: HashMap<String, String>,
 }
 
 #[derive(Debug, Clone)]
@@ -55,12 +51,7 @@ pub struct Session {
     pub uid: String,
 }
 
-pub trait Adapter: ChannelValidator + Clone + Debug + Send {
-    type Output;
-
-    /// Initialize adapter
-    fn init(opts: AdapterOptions, config: &Config) -> AdapterResult<Self::Output>;
-
+pub trait Adapter: ChannelValidator + Send + Clone + Debug {
     /// Unlock adapter
     fn unlock(&mut self) -> AdapterResult<()>;
 

primitives/src/adapter.rs

@@ -129,49 +129,6 @@ impl<'a> IntoIterator for &'a SpecValidators {
     }
 }
 
-pub struct ChannelListParams {
-    /// page to show, should be >= 1
-    pub page: u64,
-    /// channels limit per page, should be >= 1
-    pub limit: u32,
-    /// filters `valid_until` to be `>= valid_until_ge`
-    pub valid_until_ge: DateTime<Utc>,
-    /// filters the channels containing a specific validator if provided
-    pub validator: Option<String>,
-    /// Ensures that this struct can only be created by calling `new()`
-    _secret: (),
-}
-
-impl ChannelListParams {
-    pub fn new(
-        valid_until_ge: DateTime<Utc>,
-        limit: u32,
-        page: u64,
-        validator: Option<&str>,
-    ) -> Result<Self, ChannelError> {
-        if page < 1 {
-            return Err(ChannelError::InvalidArgument(
-                "Page should be >= 1".to_string(),
-            ));
-        }
-
-        if limit < 1 {
-            return Err(ChannelError::InvalidArgument(
-                "Limit should be >= 1".to_string(),
-            ));
-        }
-
-        let validator = validator.and_then(|s| if s.is_empty() { None } else { Some(s.into()) });
-        Ok(Self {
-            valid_until_ge,
-            page,
-            limit,
-            validator,
-            _secret: (),
-        })
-    }
-}
-
 #[derive(Debug, PartialEq, Eq)]
 pub enum ChannelError {
     InvalidArgument(String),

primitives/src/channel.rs

@@ -66,3 +66,118 @@ pub mod serde {
         }
     }
 }
+
+pub mod logging {
+    use slog::{Drain, OwnedKVList, Record, KV};
+    use slog_term::{
+        timestamp_local, CompactFormatSerializer, CountingWriter, Decorator, RecordDecorator,
+        Serializer, ThreadSafeTimestampFn,
+    };
+    use std::cell::RefCell;
+    use std::{io, io::Write};
+
+    pub use slog_async::Async;
+    pub use slog_term::TermDecorator;
+
+    pub struct PrefixedCompactFormat<D>
+    where
+        D: Decorator,
+    {
+        decorator: D,
+        history: RefCell<Vec<(Vec<u8>, Vec<u8>)>>,
+        fn_timestamp: Box<dyn ThreadSafeTimestampFn<Output = io::Result<()>>>,
+        prefix: String,
+    }
+
+    impl<D> Drain for PrefixedCompactFormat<D>
+    where
+        D: Decorator,
+    {
+        type Ok = ();
+        type Err = io::Error;
+
+        fn log(&self, record: &Record<'_>, values: &OwnedKVList) -> Result<Self::Ok, Self::Err> {
+            self.format_compact(record, values)
+        }
+    }
+
+    impl<D> PrefixedCompactFormat<D>
+    where
+        D: Decorator,
+    {
+        pub fn new(prefix: &str, d: D) -> PrefixedCompactFormat<D> {
+            Self {
+                fn_timestamp: Box::new(timestamp_local),
+                decorator: d,
+                history: RefCell::new(vec![]),
+                prefix: prefix.to_owned(),
+            }
+        }
+
+        fn format_compact(&self, record: &Record<'_>, values: &OwnedKVList) -> io::Result<()> {
+            self.decorator.with_record(record, values, |decorator| {
+                let indent = {
+                    let mut history_ref = self.history.borrow_mut();
+                    let mut serializer = CompactFormatSerializer::new(decorator, &mut *history_ref);
+
+                    values.serialize(record, &mut serializer)?;
+
+                    serializer.finish()?
+                };
+
+                decorator.start_whitespace()?;
+
+                for _ in 0..indent {
+                    write!(decorator, " ")?;
+                }
+
+                let comma_needed =
+                    print_msg_header(&self.prefix, &*self.fn_timestamp, decorator, record)?;
+                {
+                    let mut serializer = Serializer::new(decorator, comma_needed, false);
+
+                    record.kv().serialize(record, &mut serializer)?;
+
+                    serializer.finish()?;
+                }
+
+                decorator.start_whitespace()?;
+                writeln!(decorator)?;
+
+                decorator.flush()?;
+
+                Ok(())
+            })
+        }
+    }
+
+    pub fn print_msg_header(
+        prefix: &str,
+        fn_timestamp: &dyn ThreadSafeTimestampFn<Output = io::Result<()>>,
+        mut rd: &mut dyn RecordDecorator,
+        record: &Record<'_>,
+    ) -> io::Result<bool> {
+        rd.start_timestamp()?;
+        fn_timestamp(&mut rd)?;
+
+        rd.start_whitespace()?;
+        write!(rd, " ")?;
+
+        rd.start_level()?;
+        write!(rd, "{}", record.level().as_short_str())?;
+
+        rd.start_whitespace()?;
+        write!(rd, " ")?;
+
+        rd.start_msg()?;
+        write!(rd, "{}:", prefix)?;
+
+        rd.start_whitespace()?;
+        write!(rd, " ")?;
+
+        rd.start_msg()?;
+        let mut count_rd = CountingWriter::new(&mut rd);
+        write!(count_rd, "{}", record.msg())?;
+        Ok(count_rd.count() != 0)
+    }
+}

primitives/src/util.rs

@@ -1,18 +1,25 @@
 [package]
 name = "sentry"
 version = "0.1.0"
-authors = ["Samparsky"]
+authors = ["Omidiora Samuel <[email protected]>"]
 edition = "2018"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
+# Futures
+futures-preview = { version = "=0.3.0-alpha.19", features = ["compat"]}
+# Primitives
 primitives = {path = "../primitives"}
+adapter = { version = "0.1", path = "../adapter" }
+chrono = { version = "0.4", features = ["serde"] }
 # CLI
 clap = "2.33.0"
+# Server
+tokio = "0.2.0-alpha.6"
+hyper = {version = "=0.13.0-alpha.4", features = ["unstable-stream"]}
+regex = "1"
 # Logger
 slog = { version = "^2.2.3" , features = ["max_level_trace"] }
-slog-term = "^2.4.0"
-slog-async = "^2.3.0"
-# Database
-diesel = { version = "1.3", features = ["postgres"] }
+# Serde
+serde = {version = "^1.0", features = ['derive']}
+serde_json = "^1.0"
+serde_urlencoded = "0.6.1"