Telemetry

.dockerignore

@@ -1,6 +1,5 @@
 .dockerignore
 Dockerfile
-.git/
 .idea/
 
 docs/

Dockerfile

@@ -3,6 +3,7 @@
 FROM phusion/baseimage:0.11
 
 RUN apt-get update -qq && apt-get install -y \
+    git \
     cmake \
     g++ \
     protobuf-compiler \
@@ -24,7 +25,10 @@ WORKDIR /near
 COPY . .
 
 ENV CARGO_TARGET_DIR=/tmp/target
-RUN cargo build -p near --release && \
+RUN --mount=type=cache,target=/tmp/target \
+    --mount=type=cache,target=/usr/local/cargo/git \
+    --mount=type=cache,target=/usr/local/cargo/registry \
+    cargo build -p near --release && \
     cp /tmp/target/release/near /usr/local/bin/
 
 EXPOSE 3030 24567

chain/client/Cargo.toml

@@ -14,9 +14,13 @@ log = "0.4"
 rand = "0.6.5"
 serde_derive = "1.0"
 serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+
+sysinfo = "0.9.0"
 
 near-primitives = { path = "../../core/primitives" }
 near-store = { path = "../../core/store" }
 near-chain = { path = "../chain" }
 near-network = { path = "../network" }
 near-pool = { path = "../pool" }
+near-telemetry = { path = "../telemetry" }

chain/client/src/client.rs

@@ -8,15 +8,15 @@ use std::thread;
 use std::time::{Duration, Instant};
 
 use actix::{
-    Actor, ActorFuture, AsyncContext, Context, ContextFutureSpawner, Handler, Recipient, WrapFuture,
+    Actor, ActorFuture, Addr, AsyncContext, Context, ContextFutureSpawner, Handler, Recipient,
+    WrapFuture,
 };
-use ansi_term::Color::{Cyan, Green, White, Yellow};
 use chrono::{DateTime, Utc};
 use log::{debug, error, info, warn};
 
 use near_chain::{
     Block, BlockApproval, BlockHeader, BlockStatus, Chain, ErrorKind, Provenance, RuntimeAdapter,
-    Tip, ValidTransaction,
+    ValidTransaction,
 };
 use near_network::types::{PeerId, ReasonForBan};
 use near_network::{
@@ -29,7 +29,9 @@ use near_primitives::transaction::{ReceiptTransaction, SignedTransaction};
 use near_primitives::types::{AccountId, BlockIndex, ShardId};
 use near_primitives::unwrap_or_return;
 use near_store::Store;
+use near_telemetry::TelemetryActor;
 
+use crate::info::InfoHelper;
 use crate::sync::{most_weight_peer, BlockSync, HeaderSync, StateSync};
 use crate::types::{
     BlockProducer, ClientConfig, Error, NetworkInfo, ShardSyncStatus, Status, StatusSyncInfo,
@@ -42,10 +44,13 @@ pub struct ClientActor {
     sync_status: SyncStatus,
     chain: Chain,
     runtime_adapter: Arc<dyn RuntimeAdapter>,
+    block_producer: Option<BlockProducer>,
     tx_pool: TransactionPool,
     network_actor: Recipient<NetworkRequests>,
-    block_producer: Option<BlockProducer>,
     network_info: NetworkInfo,
+    /// Identity that represents this Client at the network level.
+    /// It is used as part of the messages that identify this client.
+    node_id: PeerId,
     /// Set of approvals for the next block.
     approvals: HashMap<usize, Signature>,
     /// Timestamp when last block was received / processed. Used to timeout block production.
@@ -56,12 +61,8 @@ pub struct ClientActor {
     block_sync: BlockSync,
     /// Keeps track of syncing state.
     state_sync: StateSync,
-    /// Timestamp when client was started.
-    started: Instant,
-    /// Total number of blocks processed.
-    num_blocks_processed: u64,
-    /// Total number of transactions processed.
-    num_tx_processed: u64,
+    /// Info helper.
+    info_helper: InfoHelper,
 }
 
 fn wait_until_genesis(genesis_time: &DateTime<Utc>) {
@@ -82,8 +83,10 @@ impl ClientActor {
         store: Arc<Store>,
         genesis_time: DateTime<Utc>,
         runtime_adapter: Arc<dyn RuntimeAdapter>,
+        node_id: PeerId,
         network_actor: Recipient<NetworkRequests>,
         block_producer: Option<BlockProducer>,
+        telemtetry_actor: Addr<TelemetryActor>,
     ) -> Result<Self, Error> {
         wait_until_genesis(&genesis_time);
         let chain = Chain::new(store, runtime_adapter.clone(), genesis_time)?;
@@ -95,13 +98,15 @@ impl ClientActor {
         if let Some(bp) = &block_producer {
             info!(target: "client", "Starting validator node: {}", bp.account_id);
         }
+        let info_helper = InfoHelper::new(telemtetry_actor, block_producer.clone());
         Ok(ClientActor {
             config,
             sync_status,
             chain,
             runtime_adapter,
             tx_pool,
             network_actor,
+            node_id,
             block_producer,
             network_info: NetworkInfo {
                 num_active_peers: 0,
@@ -115,9 +120,7 @@ impl ClientActor {
             header_sync,
             block_sync,
             state_sync,
-            started: Instant::now(),
-            num_blocks_processed: 0,
-            num_tx_processed: 0,
+            info_helper,
         })
     }
 }
@@ -292,8 +295,7 @@ impl ClientActor {
         self.last_block_processed = Instant::now();
 
         // Count blocks and transactions processed both in SYNC and regular modes.
-        self.num_blocks_processed += 1;
-        self.num_tx_processed += block.transactions.len() as u64;
+        self.info_helper.block_processed(block.transactions.len() as u64);
 
         if provenance != Provenance::SYNC {
             // If we produced the block, then we want to broadcast it.
@@ -774,7 +776,6 @@ impl ClientActor {
 
         if !needs_syncing {
             if currently_syncing {
-                self.started = Instant::now();
                 self.last_block_processed = Instant::now();
                 self.sync_status = SyncStatus::NoSync;
 
@@ -865,7 +866,6 @@ impl ClientActor {
     /// Periodically log summary.
     fn log_summary(&self, ctx: &mut Context<Self>) {
         ctx.run_later(self.config.log_summary_period, move |act, ctx| {
-            // TODO: collect traffic, tx, blocks.
             let head = unwrap_or_return!(act.chain.head(), ());
             let validators = unwrap_or_return!(act.get_epoch_block_proposers(head.epoch_hash), ());
             let num_validators = validators.len();
@@ -874,19 +874,15 @@ impl ClientActor {
             } else {
                 false
             };
-            // Block#, Block Hash, is validator/# validators, active/max peers.
-            let avg_bls = (act.num_blocks_processed as f64) / (act.started.elapsed().as_millis() as f64) * 1000.0;
-            let avg_tps = (act.num_tx_processed as f64) / (act.started.elapsed().as_millis() as f64) * 1000.0;
-            info!(target: "info", "{} {} {} {} {}",
-                  Yellow.bold().paint(display_sync_status(&act.sync_status, &head)),
-                  White.bold().paint(format!("{}/{}", if is_validator { "V" } else { "-" }, num_validators)),
-                  Cyan.bold().paint(format!("{:2}/{:?}/{:2} peers", act.network_info.num_active_peers, act.network_info.most_weight_peers.len(), act.network_info.peer_max_count)),
-                  Cyan.bold().paint(format!("⬇ {} ⬆ {}", pretty_bytes_per_sec(act.network_info.received_bytes_per_sec), pretty_bytes_per_sec(act.network_info.sent_bytes_per_sec))),
-                  Green.bold().paint(format!("{:.2} bls {:.2} tps", avg_bls, avg_tps))
+
+            act.info_helper.info(
+                &head,
+                &act.sync_status,
+                &act.node_id,
+                &act.network_info,
+                is_validator,
+                num_validators,
             );
-            act.started = Instant::now();
-            act.num_blocks_processed = 0;
-            act.num_tx_processed = 0;
 
             act.log_summary(ctx);
         });
@@ -943,58 +939,3 @@ impl ClientActor {
         Ok((payload, receipts))
     }
 }
-
-fn display_sync_status(sync_status: &SyncStatus, head: &Tip) -> String {
-    match sync_status {
-        SyncStatus::AwaitingPeers => format!("#{:>8} Waiting for peers", head.height),
-        SyncStatus::NoSync => format!("#{:>8} {}", head.height, head.last_block_hash),
-        SyncStatus::HeaderSync { current_height, highest_height } => {
-            let percent =
-                if *highest_height == 0 { 0 } else { current_height * 100 / highest_height };
-            format!("#{:>8} Downloading headers {}%", head.height, percent)
-        }
-        SyncStatus::BodySync { current_height, highest_height } => {
-            let percent =
-                if *highest_height == 0 { 0 } else { current_height * 100 / highest_height };
-            format!("#{:>8} Downloading blocks {}%", head.height, percent)
-        }
-        SyncStatus::StateSync(_sync_hash, shard_statuses) => {
-            let mut res = String::from("State ");
-            for (shard_id, shard_status) in shard_statuses {
-                res = res
-                    + format!(
-                        "{}: {}",
-                        shard_id,
-                        match shard_status {
-                            ShardSyncStatus::StateDownload {
-                                start_time: _,
-                                prev_update_time: _,
-                                prev_downloaded_size: _,
-                                downloaded_size: _,
-                                total_size: _,
-                            } => format!("download"),
-                            ShardSyncStatus::StateValidation => format!("validation"),
-                            ShardSyncStatus::StateDone => format!("done"),
-                            ShardSyncStatus::Error(error) => format!("error {}", error),
-                        }
-                    )
-                    .as_str();
-            }
-            res
-        }
-        SyncStatus::StateSyncDone => format!("State sync donee"),
-    }
-}
-
-/// Format bytes per second in a nice way.
-fn pretty_bytes_per_sec(num: u64) -> String {
-    if num < 100 {
-        // Under 0.1 kiB, display in bytes.
-        format!("{} B/s", num)
-    } else if num < 1024 * 1024 {
-        // Under 1.0 MiB/sec display in kiB/sec.
-        format!("{:.1}kiB/s", num as f64 / 1024.0)
-    } else {
-        format!("{:.1}MiB/s", num as f64 / (1024.0 * 1024.0))
-    }
-}