Replace field mesh_name with host_mesh and proc_mesh (#1937)

Summary:
Pull Request resolved: #1937

Right now I use `mesh_name` for both proc and host mesh. I find it is not adequate, since for many logs, I could have both fields.

This diff deprecates `mesh_name`, and add two new columns: `proc_mesh` and `host_mesh`.

Reviewed By: vidhyav, shayne-fletcher

Differential Revision: D87374168

fbshipit-source-id: fccb57ac7d022df2c9670a25fb89ef187bb77cee

Author: pzhan9

hyperactor_mesh/src/v1/host_mesh.rs

@@ -371,7 +371,7 @@ impl HostMesh {
     }
 
     // Use allocate_inner to set field mesh_name in span
-    #[hyperactor::instrument(fields(mesh_name=name.to_string()))]
+    #[hyperactor::instrument(fields(host_mesh=name.to_string()))]
     async fn allocate_inner(
         cx: &impl context::Actor,
         alloc: Box<dyn Alloc + Send + Sync>,
@@ -475,7 +475,7 @@ impl HostMesh {
     /// `BootstrapProcManager`. On drop, the manager walks its PID
     /// table and sends SIGKILL to any procs it spawned—tying proc
     /// lifetimes to their hosts and preventing leaks.
-    #[hyperactor::instrument(fields(mesh_name=self.name.to_string()))]
+    #[hyperactor::instrument(fields(host_mesh=self.name.to_string()))]
     pub async fn shutdown(&self, cx: &impl hyperactor::context::Actor) -> anyhow::Result<()> {
         tracing::info!(name = "HostMeshStatus", status = "Shutdown::Attempt");
         let mut failed_hosts = vec![];
@@ -534,7 +534,7 @@ impl Drop for HostMesh {
     fn drop(&mut self) {
         tracing::info!(
             name = "HostMeshStatus",
-            mesh_name = %self.name,
+            host_mesh = %self.name,
             status = "Dropping",
         );
         // Snapshot the owned hosts we're responsible for.
@@ -556,7 +556,7 @@ impl Drop for HostMesh {
             handle.spawn(async move {
                 let span = tracing::info_span!(
                     "hostmesh_drop_cleanup",
-                    %mesh_name,
+                    host_mesh = %mesh_name,
                     allocation = %allocation_label,
                     hosts = hosts.len(),
                 );
@@ -619,15 +619,15 @@ impl Drop for HostMesh {
             // No runtime here; PDEATHSIG and manager Drop remain the
             // last-resort safety net.
             tracing::warn!(
-                mesh_name = %self.name,
+                host_mesh = %self.name,
                 hosts = hosts.len(),
                 "HostMesh dropped without a tokio runtime; skipping best-effort shutdown"
             );
         }
 
         tracing::info!(
             name = "HostMeshStatus",
-            mesh_name = %self.name,
+            host_mesh = %self.name,
             status = "Dropped",
         );
     }
@@ -725,37 +725,36 @@ impl HostMeshRef {
         name: &str,
         per_host: Extent,
     ) -> v1::Result<ProcMesh> {
-        let proc_mesh_name = Name::new(name);
-        tracing::info!(
-            name = "HostMeshStatus",
-            status = "ProcMesh::Spawn::Attempt",
-            mesh_name = %self.name,
-            "spawning proc mesh {}", proc_mesh_name
-        );
-        let result = self.spawn_inner(cx, proc_mesh_name.clone(), per_host).await;
-        if result.is_ok() {
-            tracing::info!(
-                name = "HostMeshStatus",
-                status = "ProcMesh::Spawn::Success",
-                mesh_name = %self.name,
-                "spawned proc mesh {}", proc_mesh_name
-            );
-        } else {
-            tracing::error!(
-                name = "HostMeshStatus",
-                status = "ProcMesh::Spawn::Failed",
-                mesh_name = %self.name,
-                "failed to spawn proc mesh {}", proc_mesh_name
-            );
+        self.spawn_inner(cx, Name::new(name), per_host).await
+    }
+
+    #[hyperactor::instrument(fields(host_mesh=self.name.to_string(), proc_mesh=proc_mesh_name.to_string()))]
+    async fn spawn_inner(
+        &self,
+        cx: &impl context::Actor,
+        proc_mesh_name: Name,
+        per_host: Extent,
+    ) -> v1::Result<ProcMesh> {
+        tracing::info!(name = "HostMeshStatus", status = "ProcMesh::Spawn::Attempt");
+        tracing::info!(name = "ProcMeshStatus", status = "Spawn::Attempt",);
+        let result = self.spawn_inner_inner(cx, proc_mesh_name, per_host).await;
+        match &result {
+            Ok(_) => {
+                tracing::info!(name = "HostMeshStatus", status = "ProcMesh::Spawn::Success");
+                tracing::info!(name = "ProcMeshStatus", status = "Spawn::Success");
+            }
+            Err(error) => {
+                tracing::error!(name = "HostMeshStatus", status = "ProcMesh::Spawn::Failed", %error);
+                tracing::error!(name = "ProcMeshStatus", status = "Spawn::Failed", %error);
+            }
         }
         result
     }
 
-    #[hyperactor::instrument(fields(mesh_name=mesh_name.to_string()))]
-    async fn spawn_inner(
+    async fn spawn_inner_inner(
         &self,
         cx: &impl context::Actor,
-        mesh_name: Name,
+        proc_mesh_name: Name,
         per_host: Extent,
     ) -> v1::Result<ProcMesh> {
         let per_host_labels = per_host.labels().iter().collect::<HashSet<_>>();
@@ -807,7 +806,7 @@ impl HostMeshRef {
         for (host_rank, host) in self.ranks.iter().enumerate() {
             for per_host_rank in 0..per_host.num_ranks() {
                 let create_rank = per_host.num_ranks() * host_rank + per_host_rank;
-                let proc_name = Name::new(format!("{}_{}", mesh_name.name(), per_host_rank));
+                let proc_name = Name::new(format!("{}_{}", proc_mesh_name.name(), per_host_rank));
                 proc_names.push(proc_name.clone());
                 host.mesh_agent()
                     .create_or_update(
@@ -939,8 +938,7 @@ impl HostMeshRef {
         }
 
         let mesh =
-            ProcMesh::create_owned_unchecked(cx, mesh_name, extent, self.clone(), procs).await;
-        tracing::info!(name = "ProcMeshStatus", status = "Spawn::Created",);
+            ProcMesh::create_owned_unchecked(cx, proc_mesh_name, extent, self.clone(), procs).await;
         if let Ok(ref mesh) = mesh {
             // Spawn a unique mesh controller for each proc mesh, so the type of the
             // mesh can be preserved.
@@ -957,6 +955,7 @@ impl HostMeshRef {
         &self.name
     }
 
+    #[hyperactor::instrument(fields(host_mesh=self.name.to_string(), proc_mesh=proc_mesh_name.to_string()))]
     pub(crate) async fn stop_proc_mesh(
         &self,
         cx: &impl hyperactor::context::Actor,
@@ -1004,19 +1003,20 @@ impl HostMeshRef {
 
             tracing::info!(
                 name = "ProcMeshStatus",
-                mesh_name = %proc_mesh_name,
                 %proc_id,
                 status = "Stop::Sent",
             );
         }
         tracing::info!(
-            mesh_name = %self.name,
             name = "HostMeshStatus",
             status = "ProcMesh::Stop::Sent",
-            "sending Stop to proc mesh {} for {} procs: {}",
-            proc_mesh_name,
+            "sending Stop to proc mesh for {} procs: {}",
             proc_names.len(),
-            proc_names.iter().map(|n| n.to_string()).collect::<Vec<_>>().join(", ")
+            proc_names
+                .iter()
+                .map(|n| n.to_string())
+                .collect::<Vec<_>>()
+                .join(", ")
         );
 
         let start_time = RealClock.now();
@@ -1034,7 +1034,6 @@ impl HostMeshRef {
                 if !all_stopped {
                     tracing::error!(
                         name = "ProcMeshStatus",
-                        mesh_name = %proc_mesh_name,
                         status = "FailedToStop",
                         "failed to terminate proc mesh: {:?}",
                         statuses,
@@ -1044,11 +1043,7 @@ impl HostMeshRef {
                         statuses,
                     ));
                 }
-                tracing::info!(
-                    name = "ProcMeshStatus",
-                    mesh_name = %proc_mesh_name,
-                    status = "Stopped",
-                );
+                tracing::info!(name = "ProcMeshStatus", status = "Stopped");
             }
             Err(complete) => {
                 // Fill remaining ranks with a timeout status via the
@@ -1061,7 +1056,6 @@ impl HostMeshRef {
                 );
                 tracing::error!(
                     name = "ProcMeshStatus",
-                    mesh_name = %proc_mesh_name,
                     status = "StoppingTimeout",
                     "failed to terminate proc mesh before timeout: {:?}",
                     legacy,

hyperactor_mesh/src/v1/proc_mesh.rs

@@ -306,7 +306,7 @@ impl ProcMesh {
     }
 
     // Use allocate_inner to set field mesh_name in span
-    #[hyperactor::instrument(fields(mesh_name=name.to_string()))]
+    #[hyperactor::instrument(fields(proc_mesh=name.to_string()))]
     async fn allocate_inner(
         cx: &impl context::Actor,
         mut alloc: Box<dyn Alloc + Send + Sync + 'static>,
@@ -328,7 +328,7 @@ impl ProcMesh {
             .instrument(tracing::info_span!(
                 "ProcMeshStatus::Allocate::Initialize",
                 alloc_id,
-                mesh_name = name.to_string()
+                proc_mesh = %name
             ))
             .await?;
 
@@ -348,7 +348,7 @@ impl ProcMesh {
             tracing::info!(
                 name = "ProcMeshStatus",
                 status = "Allocate::ChannelServe",
-                mesh_name = name.to_string(),
+                proc_mesh = %name,
                 %addr,
                 "proc started listening on addr: {addr}"
             );
@@ -516,7 +516,7 @@ impl ProcMesh {
                 stop.notify_one();
                 tracing::info!(
                     name = "ProcMeshStatus",
-                    mesh_name = %self.name,
+                    proc_mesh = %self.name,
                     alloc_name,
                     status = "StoppingAlloc",
                     "sending stop to alloc {alloc_name}; check its log for stop status",
@@ -551,7 +551,7 @@ impl Drop for ProcMesh {
     fn drop(&mut self) {
         tracing::info!(
             name = "ProcMeshStatus",
-            mesh_name = %self.name,
+            proc_mesh = %self.name,
             status = "Dropped",
         );
     }