Add event logs for proc mesh from HostMesh::spawn (#1883)

Summary:
Pull Request resolved: #1883

The `HostMesh::spawn` method spawn a new proc mesh. This diff logs the status changes of this new proc mesh.

Reviewed By: mariusae

Differential Revision: D86931476

fbshipit-source-id: 27119c7445ab6cc66b05aba270a8884a03270155

Author: pzhan9

hyperactor/src/proc.rs

@@ -152,6 +152,19 @@ struct ProcState {
     clock: ClockKind,
 }
 
+impl Drop for ProcState {
+    fn drop(&mut self) {
+        // We only want log ProcStatus::Dropped when ProcState is dropped,
+        // rather than Proc is dropped. This is because we need to wait for
+        // Proc::inner's ref count becomes 0.
+        tracing::info!(
+            proc_id = %self.proc_id,
+            name = "ProcStatus",
+            status = "Dropped"
+        );
+    }
+}
+
 /// A snapshot view of the proc's actor ledger.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct ActorLedgerSnapshot {
@@ -337,7 +350,6 @@ impl Proc {
     }
 
     /// Create a new direct-addressed proc.
-    #[tracing::instrument]
     pub async fn direct(addr: ChannelAddr, name: String) -> Result<Self, ChannelError> {
         let (addr, rx) = channel::serve(addr)?;
         let proc_id = ProcId::Direct(addr, name);
@@ -347,7 +359,6 @@ impl Proc {
     }
 
     /// Create a new direct-addressed proc with a default sender for the forwarder.
-    #[tracing::instrument(skip(default))]
     pub fn direct_with_default(
         addr: ChannelAddr,
         name: String,
@@ -369,6 +380,11 @@ impl Proc {
         forwarder: BoxedMailboxSender,
         clock: ClockKind,
     ) -> Self {
+        tracing::info!(
+            proc_id = %proc_id,
+            name = "ProcStatus",
+            status = "Created"
+        );
         Self {
             inner: Arc::new(ProcState {
                 proc_id,

hyperactor_mesh/src/v1/host_mesh.rs

@@ -670,6 +670,16 @@ impl HostMeshRef {
         cx: &impl context::Actor,
         name: &str,
         per_host: Extent,
+    ) -> v1::Result<ProcMesh> {
+        self.spawn_inner(cx, Name::new(name), per_host).await
+    }
+
+    #[hyperactor::instrument(fields(mesh_name=mesh_name.to_string()))]
+    async fn spawn_inner(
+        &self,
+        cx: &impl context::Actor,
+        mesh_name: Name,
+        per_host: Extent,
     ) -> v1::Result<ProcMesh> {
         let per_host_labels = per_host.labels().iter().collect::<HashSet<_>>();
         let host_labels = self.region.labels().iter().collect::<HashSet<_>>();
@@ -690,7 +700,13 @@ impl HostMeshRef {
             .map_err(|err| v1::Error::ConfigurationError(err.into()))?;
 
         let region: Region = extent.clone().into();
-        let mesh_name = Name::new(name);
+
+        tracing::info!(
+            name = "ProcMeshStatus",
+            status = "Spawn::Attempt",
+            %region,
+            "spawning proc mesh"
+        );
 
         let mut procs = Vec::new();
         let num_ranks = region.num_ranks();
@@ -714,7 +730,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!("{}_{}", name, per_host_rank));
+                let proc_name = Name::new(format!("{}_{}", mesh_name.name(), per_host_rank));
                 proc_names.push(proc_name.clone());
                 host.mesh_agent()
                     .create_or_update(
@@ -743,8 +759,15 @@ impl HostMeshRef {
                             format!("failed while querying proc status: {}", e),
                         )
                     })?;
+                let proc_id = host.named_proc(&proc_name);
+                tracing::info!(
+                    name = "ProcMeshStatus",
+                    status = "Spawn::CreatingProc",
+                    %proc_id,
+                    rank = create_rank,
+                );
                 procs.push(ProcRef::new(
-                    host.named_proc(&proc_name),
+                    proc_id,
                     create_rank,
                     // TODO: specify or retrieve from state instead, to avoid attestation.
                     ActorRef::attest(host.named_proc(&proc_name).actor_id("agent", 0)),
@@ -803,6 +826,15 @@ impl HostMeshRef {
                         },
                     };
 
+                    tracing::error!(
+                        name = "ProcMeshStatus",
+                        status = "Spawn::GetRankStatus",
+                        rank = host_rank,
+                        "rank {} is terminating with state: {}",
+                        host_rank,
+                        state
+                    );
+
                     return Err(v1::Error::ProcCreationError {
                         state,
                         host_rank,
@@ -811,6 +843,12 @@ impl HostMeshRef {
                 }
             }
             Err(complete) => {
+                tracing::error!(
+                    name = "ProcMeshStatus",
+                    status = "Spawn::GetRankStatus",
+                    "timeout after {:?} when waiting for procs being created",
+                    config::global::get(PROC_SPAWN_MAX_IDLE),
+                );
                 // Fill remaining ranks with a timeout status via the
                 // legacy shim.
                 let legacy = mesh_to_rankedvalues_with_default(
@@ -823,7 +861,10 @@ impl HostMeshRef {
             }
         }
 
-        ProcMesh::create_owned_unchecked(cx, mesh_name, extent, self.clone(), procs).await
+        let mesh =
+            ProcMesh::create_owned_unchecked(cx, mesh_name, extent, self.clone(), procs).await;
+        tracing::info!(name = "ProcMeshStatus", status = "Spawn::Created",);
+        mesh
     }
 
     /// The name of the referenced host mesh.