Merge pull request #182 from firstbatchxyz/erhant/steps-log

feat: add steps info to diagnostics

Author: erhant

Cargo.lock

@@ -2,14 +2,12 @@
 resolver = "2"
 members = ["compute", "p2p", "workflows", "utils"]
 
-# FIXME: removing this breaks the workflows
-# compute node is the default member, until Oracle comes in
-# then, a Launcher will be the default member
+# removing this breaks the workflows
 default-members = ["compute"]
 
 [workspace.package]
 edition = "2021"
-version = "0.3.7"
+version = "0.3.8"
 license = "Apache-2.0"
 readme = "README.md"
 

Cargo.toml

@@ -39,7 +39,7 @@ Compute nodes can technically do any arbitrary task, from computing the square r
 
 ### Running a Node
 
-Refer to [node guide](./docs/NODE_GUIDE.md) to quickly get started and run your own node!
+Use the [Dria Compute Launcher](https://github.com/firstbatchxyz/dkn-compute-launcher/) to run a compute node with many more features!
 
 ## Releases
 

README.md

@@ -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

@@ -18,7 +18,7 @@ pub struct DriaComputeNodeConfig {
     pub secret_key: SecretKey,
     /// Wallet public key, derived from the secret key.
     pub public_key: PublicKey,
-    /// Wallet address in hex, derived from the public key.
+    /// Wallet address in hex without `0x` prefix, derived from the public key.
     pub address: String,
     /// Peer ID of the node.
     pub peer_id: PeerId,

compute/src/config.rs

@@ -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,8 +1,9 @@
+use colored::Colorize;
 use dkn_p2p::libp2p::multiaddr::Protocol;
 use std::time::Duration;
 use tokio::time::Instant;
 
-use crate::{refresh_dria_nodes, DriaComputeNode, DRIA_COMPUTE_NODE_VERSION};
+use crate::{refresh_dria_nodes, utils::get_steps, DriaComputeNode, DRIA_COMPUTE_NODE_VERSION};
 
 /// Number of seconds such that if the last ping is older than this, the node is considered unreachable.
 const PING_LIVENESS_SECS: u64 = 150;
@@ -21,20 +22,32 @@ 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 tasks count
-        let [single, batch] = self.get_pending_task_count();
-        diagnostics.push(format!(
-            "Pending Tasks (single/batch): {} / {}",
-            single, batch
-        ));
+        // print steps
+        if let Ok(steps) = get_steps(&self.config.address).await {
+            let earned = steps.score - self.initial_steps;
+            diagnostics.push(format!(
+                "Steps: {} total, {} earned in this run, within top {}%",
+                steps.score, earned, steps.percentile
+            ));
+        }
 
         // completed tasks count is printed as well in debug
         if log::log_enabled!(log::Level::Debug) {
@@ -60,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
@@ -72,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

@@ -13,7 +13,7 @@ use tokio::{sync::mpsc, time::Instant};
 use crate::{
     config::*,
     gossipsub::*,
-    utils::{crypto::secret_to_keypair, refresh_dria_nodes, SpecCollector},
+    utils::{crypto::secret_to_keypair, get_steps, refresh_dria_nodes, SpecCollector},
     workers::task::{TaskWorker, TaskWorkerInput, TaskWorkerMetadata, TaskWorkerOutput},
 };
 
@@ -56,6 +56,8 @@ pub struct DriaComputeNode {
     completed_tasks_batch: usize,
     /// Specifications collector.
     spec_collector: SpecCollector,
+    /// Initial steps count.
+    initial_steps: u64,
 }
 
 impl DriaComputeNode {
@@ -114,6 +116,12 @@ impl DriaComputeNode {
         };
 
         let model_names = config.workflows.get_model_names();
+
+        let initial_steps = get_steps(&config.address)
+            .await
+            .map(|s| s.score)
+            .unwrap_or_default();
+
         Ok((
             DriaComputeNode {
                 config,
@@ -132,6 +140,7 @@ impl DriaComputeNode {
                 completed_tasks_single: 0,
                 completed_tasks_batch: 0,
                 // others
+                initial_steps,
                 spec_collector: SpecCollector::new(model_names),
                 last_pinged_at: Instant::now(),
             },

compute/src/node/mod.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?;