add colors for better visibility

Author: erhant

Cargo.lock

@@ -34,6 +34,7 @@ rand.workspace = true
 env_logger.workspace = true
 log.workspace = true
 eyre.workspace = true
+colored = "3.0.0"
 
 # encryption (ecies) & signatures (ecdsa) & hashing & bloom-filters
 ecies = { version = "0.2", default-features = false, features = ["pure"] }

compute/Cargo.toml

@@ -1,3 +1,4 @@
+use colored::Colorize;
 use dkn_p2p::libp2p::gossipsub::MessageAcceptance;
 use dkn_utils::get_current_time_nanos;
 use dkn_workflows::{Model, ModelProvider};
@@ -62,7 +63,7 @@ impl PingpongHandler {
             return Ok(MessageAcceptance::Ignore);
         }
 
-        log::info!("Received a ping for: {}", pingpong.uuid);
+        log::info!("Received a {} for: {}", "ping".blue(), pingpong.uuid);
 
         // record ping moment
         node.last_pinged_at = Instant::now();

compute/src/gossipsub/pingpong.rs

@@ -1,3 +1,4 @@
+use colored::Colorize;
 use dkn_p2p::libp2p::multiaddr::Protocol;
 use std::time::Duration;
 use tokio::time::Instant;
@@ -21,19 +22,29 @@ impl DriaComputeNode {
     pub(crate) async fn handle_diagnostic_refresh(&self) {
         let mut diagnostics = vec![format!("Diagnostics (v{}):", DRIA_COMPUTE_NODE_VERSION)];
 
+        // if we have not received pings for a while, we are considered offline
+        let is_offline =
+            self.last_pinged_at < Instant::now() - Duration::from_secs(PING_LIVENESS_SECS);
+
         // print peer counts
         match self.p2p.peer_counts().await {
-            Ok((mesh, all)) => {
-                diagnostics.push(format!("Peer Count (mesh/all): {} / {}", mesh, all))
-            }
+            Ok((mesh, all)) => diagnostics.push(format!(
+                "Peer Count (mesh/all): {} / {}",
+                if mesh == 0 {
+                    "0".red()
+                } else {
+                    mesh.to_string().white()
+                },
+                all
+            )),
             Err(e) => log::error!("Error getting peer counts: {:?}", e),
         }
 
         // print steps
         if let Ok(steps) = get_steps(&self.config.address).await {
             let earned = steps.score - self.initial_steps;
             diagnostics.push(format!(
-                "Steps: {} total (+{} this run), within top {}%",
+                "Steps: {} total, {} earned in this run, within top {}%",
                 steps.score, earned, steps.percentile
             ));
         }
@@ -62,10 +73,20 @@ impl DriaComputeNode {
                 .join(", ")
         ));
 
+        // add network status as well
+        diagnostics.push(format!(
+            "Node Status: {}",
+            if is_offline {
+                "OFFLINE".bold().red()
+            } else {
+                "ONLINE".bold().green()
+            }
+        ));
+
         log::info!("{}", diagnostics.join("\n  "));
 
-        // check liveness of the node w.r.t last ping-pong time
-        if self.last_pinged_at < Instant::now() - Duration::from_secs(PING_LIVENESS_SECS) {
+        // if offline, print this error message as well
+        if is_offline {
             log::error!(
                 "Node has not received any pings for at least {} seconds & it may be unreachable!\nPlease restart your node!",
                 PING_LIVENESS_SECS
@@ -74,7 +95,7 @@ impl DriaComputeNode {
 
         // added rpc nodes check, sometimes this happens when API is down / bugs for some reason
         if self.dria_nodes.rpc_peerids.is_empty() {
-            log::error!("No RPC peerids were found to be available, please restart your node!",);
+            log::error!("No RPC peer IDs were found to be available, please restart your node!",);
         }
     }
 

compute/src/node/diagnostic.rs

@@ -1,3 +1,4 @@
+use colored::Colorize;
 use dkn_p2p::libp2p::gossipsub::{Message, MessageAcceptance, MessageId};
 use dkn_p2p::libp2p::PeerId;
 use eyre::Result;
@@ -40,7 +41,11 @@ impl DriaComputeNode {
     pub async fn publish(&mut self, message: DriaMessage) -> Result<()> {
         let message_bytes = serde_json::to_vec(&message)?;
         let message_id = self.p2p.publish(&message.topic, message_bytes).await?;
-        log::info!("Published {} message ({})", message.topic, message_id);
+        log::info!(
+            "Published {} message ({})",
+            message.topic.blue(),
+            message_id
+        );
         Ok(())
     }
 
@@ -80,11 +85,10 @@ impl DriaComputeNode {
 
                 // parse the raw gossipsub message to a prepared DKN message
                 // the received message is expected to use IdentHash for the topic, so we can see the name of the topic immediately.
-                log::debug!("Parsing {} message.", gossipsub_message.topic.as_str());
                 let message: DriaMessage = match serde_json::from_slice(&gossipsub_message.data) {
                     Ok(message) => message,
                     Err(e) => {
-                        log::error!("Error parsing message: {:?}", e);
+                        log::error!("Error parsing {} message: {:?}", gossipsub_message.topic, e);
                         log::debug!(
                             "Message: {}",
                             String::from_utf8_lossy(&gossipsub_message.data)

compute/src/node/gossipsub.rs

@@ -1,3 +1,4 @@
+use colored::Colorize;
 use dkn_p2p::libp2p::{request_response::ResponseChannel, PeerId};
 use eyre::{eyre, Result};
 
@@ -46,7 +47,8 @@ impl DriaComputeNode {
         spec_request: <SpecResponder as IsResponder>::Request,
     ) -> Result<()> {
         log::info!(
-            "Got a spec request from peer {} with id {}",
+            "Got a {} request from peer {} with id {}",
+            "spec".green(),
             peer_id,
             spec_request.request_id
         );
@@ -55,7 +57,8 @@ impl DriaComputeNode {
         let response_data = serde_json::to_vec(&response)?;
 
         log::info!(
-            "Responding to spec request from peer {} with id {}",
+            "Responding to {} request from peer {} with id {}",
+            "spec".green(),
             peer_id,
             response.request_id
         );
@@ -75,7 +78,7 @@ impl DriaComputeNode {
         channel: ResponseChannel<Vec<u8>>,
         task_request: <TaskResponder as IsResponder>::Request,
     ) -> Result<()> {
-        log::info!("Received a task request from {}", peer_id);
+        log::info!("Received a {} request from {}", "task".yellow(), peer_id);
 
         let (task_input, task_metadata) =
             TaskResponder::prepare_worker_input(self, &task_request, channel).await?;

compute/src/node/reqres.rs

@@ -26,7 +26,7 @@ mod tests {
 
     use super::*;
 
-    // TODO: remove this test
+    // TODO: remove this test when its done
     #[test]
     fn test_enum_serialization() {
         use serde::Deserialize;

compute/src/reqres/mod.rs

@@ -1,5 +1,6 @@
 #![allow(unused)]
 
+use colored::Colorize;
 use dkn_p2p::libp2p::request_response::ResponseChannel;
 use dkn_utils::get_current_time_nanos;
 use dkn_workflows::{Entry, Executor, ModelProvider, Workflow};
@@ -122,7 +123,11 @@ impl TaskResponder {
         let response = match task_output.result {
             Ok(result) => {
                 // prepare signed and encrypted payload
-                log::info!("Publishing result for task {}", task_output.task_id);
+                log::info!(
+                    "Publishing {} result for {}",
+                    "task".green(),
+                    task_output.task_id
+                );
                 let payload = TaskResponsePayload::new(
                     result,
                     &task_output.task_id,