Reduce transport/sync metric cardinality

Author: gorbak25

crates/corro-agent/src/agent/handlers.rs

@@ -299,9 +299,10 @@ pub async fn handle_gossip_to_send(
                     error!("could not write datagram {addr}: {e}");
                     return;
                 }
-                counter!("corro.peer.datagram.sent.total", "actor_id" => actor_id.to_string())
+                counter!("corro.peer.datagram.sent.total", "traffic" => "foca")
                     .increment(1);
-                counter!("corro.peer.datagram.bytes.sent.total").increment(len as u64);
+                counter!("corro.peer.datagram.bytes.sent.total", "traffic" => "foca")
+                    .increment(len as u64);
             }
             .instrument(debug_span!("send_swim_payload", %addr, %actor_id, buf_size = len)),
         );
@@ -340,10 +341,16 @@ pub async fn handle_notifications(
                     MemberAddedResult::NewMember | MemberAddedResult::Removed => {
                         if matches!(member_added_res, MemberAddedResult::Removed) {
                             debug!("Member Removed {actor:?} due to member id mismatch");
-                            counter!("corro.gossip.member.removed", "id" => actor.id().0.to_string(), "addr" => actor.addr().to_string()).increment(1);
+                            counter!(
+                                "corro.gossip.member.removed",
+                                "traffic" => "foca",
+                                "kind" => "member_id_mismatch"
+                            )
+                            .increment(1);
                         } else {
                             debug!("Member Added {actor:?}");
-                            counter!("corro.gossip.member.added", "id" => actor.id().0.to_string(), "addr" => actor.addr().to_string()).increment(1);
+                            counter!("corro.gossip.member.added", "traffic" => "foca")
+                                .increment(1);
                         }
 
                         let members_len = { agent.members().read().states.len() as u32 };
@@ -386,7 +393,8 @@ pub async fn handle_notifications(
                 info!("Member Down {actor:?} (removed: {removed})");
                 if removed {
                     debug!("Member Down {actor:?}");
-                    counter!("corro.gossip.member.removed", "id" => actor.id().0.to_string(), "addr" => actor.addr().to_string()).increment(1);
+                    counter!("corro.gossip.member.removed", "traffic" => "foca", "kind" => "member_down")
+                        .increment(1);
                     // actually removed a member
                     // notify of new cluster size
                     let member_len = { agent.members().read().states.len() as u32 };
@@ -1065,13 +1073,20 @@ pub async fn handle_sync(
 ) -> Result<(), SyncClientError> {
     let sync_state = generate_sync(bookie, agent.actor_id()).await;
 
-    for (actor_id, needed) in sync_state.need.iter() {
-        gauge!("corro.sync.client.needed", "actor_id" => actor_id.to_string())
-            .set(needed.len() as f64);
-    }
-    for (actor_id, version) in sync_state.heads.iter() {
-        gauge!("corro.sync.client.head", "actor_id" => actor_id.to_string()).set(version.0 as f64);
-    }
+    let needed_total = sync_state
+        .need
+        .values()
+        .map(|needed| needed.len() as u64)
+        .sum::<u64>();
+    gauge!("corro.sync.client.needed.v2", "traffic" => "sync").set(needed_total as f64);
+
+    let head_max = sync_state
+        .heads
+        .values()
+        .map(|version| version.0)
+        .max()
+        .unwrap_or_default();
+    gauge!("corro.sync.client.head.v2", "traffic" => "sync").set(head_max as f64);
 
     let chosen: Vec<(ActorId, SocketAddr)> = {
         let candidates = {

crates/corro-agent/src/api/peer/mod.rs

@@ -1088,7 +1088,7 @@ pub async fn parallel_sync(
                     }
                     trace!(%actor_id, self_actor_id = %agent.actor_id(), "read clock payload");
 
-                    counter!("corro.sync.client.member", "id" => actor_id.to_string(), "addr" => addr.to_string()).increment(1);
+                    counter!("corro.sync.client.member", "traffic" => "sync").increment(1);
 
                     let needs = our_sync_state.compute_available_needs(&their_sync_state);
 
@@ -1117,9 +1117,8 @@ pub async fn parallel_sync(
             Err(e) => {
                 counter!(
                     "corro.sync.client.handshake.errors",
-                    "actor_id" => actor_id.to_string(),
-                    "addr" => addr.to_string(),
-                    "error" => e.to_string()
+                    "traffic" => "sync",
+                    "kind" => "handshake"
                 )
                 .increment(1);
                 match agg {
@@ -1294,7 +1293,8 @@ pub async fn parallel_sync(
                         continue 'servers;
                     }
 
-                    counter!("corro.sync.client.req.sent", "actor_id" => server_actor_id.to_string()).increment(req_len as u64);
+                    counter!("corro.sync.client.req.sent", "traffic" => "sync")
+                        .increment(req_len as u64);
                 }
 
                 if !send_buf.is_empty() {
@@ -1344,7 +1344,7 @@ pub async fn parallel_sync(
                             let changes_len = cmp::max(change.len(), 1);
                             // tracing::Span::current().record("changes_len", changes_len);
                             count += changes_len;
-                            counter!("corro.sync.changes.recv", "actor_id" => actor_id.to_string())
+                            counter!("corro.sync.changes.recv", "traffic" => "sync")
                                 .increment(changes_len as u64);
 
                             debug!(
@@ -1598,7 +1598,7 @@ pub async fn serve_sync(
 
             debug!(actor_id = %agent.actor_id(), "done writing sync messages (count: {count})");
 
-            counter!("corro.sync.changes.sent", "actor_id" => their_actor_id.to_string()).increment(count as u64);
+            counter!("corro.sync.changes.sent", "traffic" => "sync").increment(count as u64);
 
             Ok::<_, SyncError>(count)
         }.instrument(info_span!("process_versions_to_send")),
@@ -1649,7 +1649,7 @@ pub async fn serve_sync(
 
             debug!(actor_id = %agent.actor_id(), "done reading sync messages");
 
-            counter!("corro.sync.requests.recv", "actor_id" => their_actor_id.to_string()).increment(count as u64);
+            counter!("corro.sync.requests.recv", "traffic" => "sync").increment(count as u64);
 
             Ok(count)
         }.instrument(info_span!("process_version_requests"))

crates/corro-agent/src/broadcast/mod.rs

@@ -1102,7 +1102,7 @@ fn try_transmit_broadcast(
                 error!("could not write to uni stream to {addr}: {e}");
             }
             Ok(Ok(_)) => {
-                counter!("corro.peer.stream.bytes.sent.total", "type" => "uni")
+                counter!("corro.peer.stream.bytes.sent.total", "traffic" => "broadcast")
                     .increment(len as u64);
             }
         }