imp(orchestrator): optimize heartbeat & metrics storage (#545)

* optimize heartbeat & metrics store

Author: JannikSt

crates/orchestrator/src/api/routes/heartbeat.rs

@@ -95,46 +95,54 @@ async fn heartbeat(
         error!("Heartbeat Error: {}", e);
     }
     if let Some(metrics) = heartbeat.metrics.clone() {
-        // Get all previously reported metrics for this node
-        let previous_metrics = match app_state
+        // Get current metric keys for this node efficiently using HKEYS
+        let previous_metric_keys = match app_state
             .store_context
             .metrics_store
-            .get_metrics_for_node(node_address)
+            .get_metric_keys_for_node(node_address)
             .await
         {
-            Ok(metrics) => metrics,
+            Ok(keys) => keys,
             Err(e) => {
-                error!("Error getting metrics for node: {}", e);
-                Default::default()
+                error!("Error getting metric keys for node: {}", e);
+                Vec::new()
             }
         };
 
-        // Create a HashSet of new metrics for efficient lookup
-        let new_metrics_set: HashSet<_> = metrics
+        // Create a HashSet of new metric keys for efficient lookup
+        let new_metrics_set: HashSet<String> = metrics
             .iter()
-            .map(|metric| (&metric.key.task_id, &metric.key.label))
+            .map(|metric| {
+                let task_id = if metric.key.task_id.is_empty() {
+                    "manual".to_string()
+                } else {
+                    metric.key.task_id.clone()
+                };
+                format!("{}:{}", task_id, metric.key.label.replace(':', ""))
+            })
             .collect();
 
-        // Clean up stale metrics from Redis only
-        // The sync service will handle all Prometheus updates
-        for (task_id, task_metrics) in previous_metrics {
-            for (label, _value) in task_metrics {
-                let prev_key = (&task_id, &label);
-                if !new_metrics_set.contains(&prev_key) {
-                    // Remove from Redis metrics store
-                    if let Err(e) = app_state
-                        .store_context
-                        .metrics_store
-                        .delete_metric(&task_id, &label, &node_address.to_string())
-                        .await
-                    {
-                        error!("Error deleting metric: {}", e);
-                    }
+        // Find stale metrics to delete
+        let stale_metrics: Vec<String> = previous_metric_keys
+            .into_iter()
+            .filter(|key| !new_metrics_set.contains(key))
+            .collect();
+
+        // Delete stale metrics efficiently
+        for metric_key in stale_metrics {
+            if let Some((task_id, label)) = metric_key.split_once(':') {
+                if let Err(e) = app_state
+                    .store_context
+                    .metrics_store
+                    .delete_metric(task_id, label, &node_address.to_string())
+                    .await
+                {
+                    error!("Error deleting metric: {}", e);
                 }
             }
         }
 
-        // Store new metrics in Redis only
+        // Store new metrics in Redis
         if let Err(e) = app_state
             .store_context
             .metrics_store

crates/orchestrator/src/main.rs

@@ -126,6 +126,51 @@ struct Args {
     max_healthy_nodes_with_same_endpoint: u32,
 }
 
+async fn run_inactive_node_metric_migration(store_context: Arc<StoreContext>) -> Result<()> {
+    info!("Starting migration of inactive node metrics to new data model");
+
+    let node_addresses = match store_context.node_store.get_nodes().await {
+        Ok(nodes) => {
+            let addresses: Vec<_> = nodes.into_iter().map(|node| node.address).collect();
+            info!("Found {} nodes to migrate", addresses.len());
+            addresses
+        }
+        Err(e) => {
+            error!("Error getting all nodes for migration: {}", e);
+            return Ok(()); // Don't fail startup if migration can't get nodes
+        }
+    };
+
+    let mut migrated_count = 0;
+    let mut error_count = 0;
+
+    for node_address in node_addresses {
+        match store_context
+            .metrics_store
+            .migrate_node_metrics_if_needed(node_address)
+            .await
+        {
+            Ok(()) => {
+                migrated_count += 1;
+                if migrated_count % 100 == 0 {
+                    info!("Migrated {} nodes so far...", migrated_count);
+                }
+            }
+            Err(e) => {
+                error!("Error migrating metrics for node {}: {}", node_address, e);
+                error_count += 1;
+                // Continue with other nodes even if one fails
+            }
+        }
+    }
+
+    info!(
+        "Migration completed. Successfully migrated {} nodes, {} errors",
+        migrated_count, error_count
+    );
+    Ok(())
+}
+
 #[tokio::main]
 async fn main() -> Result<()> {
     let args = Args::parse();
@@ -176,6 +221,10 @@ async fn main() -> Result<()> {
     let store = Arc::new(RedisStore::new(&args.redis_store_url));
     let store_context = Arc::new(StoreContext::new(store.clone()));
 
+    // Run one-time migration for inactive nodes
+    let migration_store_context = store_context.clone();
+    run_inactive_node_metric_migration(migration_store_context).await?;
+
     let p2p_client = Arc::new(P2PClient::new(wallet.clone()).await.unwrap());
 
     let contracts = ContractBuilder::new(wallet.provider())