Merge pull request dmnd-pool#137 from Priceless-P/feat/worker-activity

Log worker activity

Author: jbesraa

src/monitor/mod.rs

@@ -4,18 +4,19 @@ use tracing::{debug, error};
 
 use crate::{
     config::Configuration,
-    monitor::{logs::ProxyLog, shares::ShareInfo},
+    monitor::{logs::ProxyLog, shares::ShareInfo, worker_activity::WorkerActivity},
     shared::error::Error,
     LOCAL_URL, PRODUCTION_URL, STAGING_URL, TESTNET3_URL,
 };
 
 pub mod logs;
 pub mod shares;
-
+pub mod worker_activity;
 pub struct MonitorAPI {
     pub url: Url,
     pub client: reqwest::Client,
 }
+
 fn proxy_log_server_endpoint() -> String {
     match Configuration::environment().as_str() {
         "staging" => format!("{}/api/proxy/logs", STAGING_URL),
@@ -36,6 +37,17 @@ fn shares_server_endpoint() -> String {
     }
 }
 
+fn worker_activity_server_endpoint() -> String {
+    // Determine the monitoring server URL based on the environment
+    match Configuration::environment().as_str() {
+        "staging" => format!("{}/api/worker/activity", LOCAL_URL),
+        "testnet3" => format!("{}/api/worker/activity", TESTNET3_URL),
+        "local" => format!("{}/api/worker/activity", LOCAL_URL),
+        "production" => format!("{}/api/worker/activity", PRODUCTION_URL),
+        _ => unreachable!(),
+    }
+}
+
 impl MonitorAPI {
     pub fn new(url: String) -> Self {
         let client = reqwest::Client::new();
@@ -86,4 +98,24 @@ impl MonitorAPI {
             }
         }
     }
+
+    /// Sends a worker activity log to the monitoring server.
+    pub async fn send_worker_activity(&self, activity: WorkerActivity) -> Result<(), Error> {
+        let token = crate::config::Configuration::token().expect("Token is not set");
+        debug!("Sending worker activity to API: {:?}", activity);
+        let response = self
+            .client
+            .post(worker_activity_server_endpoint())
+            .json(&json!({ "data": activity, "token": token }))
+            .send()
+            .await?;
+
+        match response.error_for_status() {
+            Ok(_) => Ok(()),
+            Err(err) => {
+                error!("Failed to send worker activity: {}", err);
+                Err(err.into())
+            }
+        }
+    }
 }

src/monitor/worker_activity.rs

@@ -0,0 +1,28 @@
+use crate::monitor::{worker_activity_server_endpoint, MonitorAPI};
+
+#[derive(serde::Serialize, Debug)]
+pub enum WorkerActivityType {
+    Connected,
+    Disconnected,
+}
+
+#[derive(serde::Serialize, Debug)]
+pub struct WorkerActivity {
+    user_agent: String,
+    worker_name: String,
+    activity: WorkerActivityType,
+}
+
+impl WorkerActivity {
+    pub fn new(user_agent: String, worker_name: String, activity: WorkerActivityType) -> Self {
+        WorkerActivity {
+            user_agent,
+            worker_name,
+            activity,
+        }
+    }
+
+    pub fn monitor_api(&self) -> MonitorAPI {
+        MonitorAPI::new(worker_activity_server_endpoint())
+    }
+}

src/translator/downstream/downstream.rs

@@ -1,6 +1,9 @@
 use crate::{
     api::stats::StatsSender,
-    monitor::shares::{RejectionReason, ShareInfo, SharesMonitor},
+    monitor::{
+        shares::{RejectionReason, ShareInfo, SharesMonitor},
+        worker_activity::{WorkerActivity, WorkerActivityType},
+    },
     proxy_state::{DownstreamType, ProxyState},
     shared::utils::AbortOnDrop,
     translator::{error::Error, utils::validate_share},
@@ -117,6 +120,7 @@ pub struct Downstream {
     pub recent_jobs: RecentJobs,
     pub first_job: Notify<'static>,
     pub share_monitor: SharesMonitor,
+    pub user_agent: std::cell::RefCell<String>, // RefCell is used here because `handle_subscribe` and `handle_authorize` take &self not &mut self and we need to mutate user_agent
 }
 
 impl Downstream {
@@ -197,6 +201,7 @@ impl Downstream {
             recent_jobs: RecentJobs::new(),
             first_job: last_notify.expect("we have an assertion at the beginning of this function"),
             share_monitor: SharesMonitor::new(),
+            user_agent: std::cell::RefCell::new(String::new()),
         }));
 
         if let Err(e) = start_receive_downstream(
@@ -399,6 +404,7 @@ impl Downstream {
             stats_sender,
             recent_jobs: RecentJobs::new(),
             share_monitor: SharesMonitor::new(),
+            user_agent: std::cell::RefCell::new(String::new()),
         }
     }
 }
@@ -446,12 +452,29 @@ impl IsServer<'static> for Downstream {
             "mining.notify".to_string(),
             "ae6812eb4cd7735a302a8a9dd95cf71f".to_string(),
         );
-
+        self.user_agent.replace(request.agent_signature.clone());
         vec![set_difficulty_sub, notify_sub]
     }
 
-    fn handle_authorize(&self, _request: &client_to_server::Authorize) -> bool {
+    fn handle_authorize(&self, request: &client_to_server::Authorize) -> bool {
         if self.authorized_names.is_empty() {
+            let user_agent = self.user_agent.borrow().clone();
+            let worker_activity = WorkerActivity::new(
+                user_agent,
+                request.name.clone(),
+                WorkerActivityType::Connected,
+            );
+
+            tokio::spawn(async move {
+                if let Err(e) = worker_activity
+                    .monitor_api()
+                    .send_worker_activity(worker_activity)
+                    .await
+                {
+                    error!("Failed to send worker activity: {}", e);
+                }
+            });
+
             true
         } else {
             // when downstream is already authorized we do not want return an ok response otherwise

src/translator/downstream/receive_from_downstream.rs

@@ -1,5 +1,9 @@
 use super::{downstream::Downstream, task_manager::TaskManager};
-use crate::{proxy_state::ProxyState, translator::error::Error};
+use crate::{
+    monitor::worker_activity::{WorkerActivity, WorkerActivityType},
+    proxy_state::ProxyState,
+    translator::error::Error,
+};
 use roles_logic_sv2::utils::Mutex;
 use std::sync::Arc;
 use sv1_api::{client_to_server::Submit, json_rpc};
@@ -58,6 +62,31 @@ pub async fn start_receive_downstream(
             if let Err(e) = Downstream::remove_downstream_hashrate_from_channel(&downstream) {
                 error!("Failed to remove downstream hashrate from channel: {}", e)
             };
+
+            let (worker_name, user_agent) = downstream
+                .safe_lock(|d| {
+                    (
+                        d.authorized_names.first().cloned().unwrap_or_default(),
+                        d.user_agent.borrow().clone(),
+                    )
+                })
+                .unwrap_or_else(|e| {
+                    error!("Failed to lock downstream: {:?}", e);
+                    ProxyState::update_inconsistency(Some(1));
+                    ("unknown".to_string(), "unknown".to_string())
+                });
+
+            let worker_activity =
+                WorkerActivity::new(user_agent, worker_name, WorkerActivityType::Disconnected);
+
+            worker_activity
+                .monitor_api()
+                .send_worker_activity(worker_activity)
+                .await
+                .unwrap_or_else(|e| {
+                    error!("Failed to send worker activity: {}", e);
+                });
+
             // Apparently there is no way to make the compiler happy without unwrapping here. But
             // is not an issue since:
             // 1. the mutex should never get poisioned and if it does will be very very rare