add observability metadata to tracing

Author: gregl83

vigilo/src/cli/commands/coordinator.rs

@@ -308,11 +308,23 @@ async fn drain_dispatch_batch(
     debug!(coordinator_id = %coordinator_id, "draining dispatchable run-shard windows");
 
     let mut dispatched = 0usize;
+    let mut dispatched_by_alias = std::collections::BTreeMap::<String, usize>::new();
     for _ in 0..config.max_dispatch_per_cycle {
         let control_db = database.control().await?;
         let Some(route) = run_dispatch::select_next_dispatch_route(control_db).await? else {
             break;
         };
+        let database_alias = route.database_alias.clone();
+
+        debug!(
+            coordinator_id = %coordinator_id,
+            run_id = %route.run_id,
+            run_shard = route.run_shard,
+            database_alias = %database_alias,
+            placement_status = %route.placement_status,
+            routing_decision = "selected_dispatch_route",
+            "selected dispatch route"
+        );
 
         let Some(snapshot) = run_dispatch::prepare_dispatch_run_snapshot(
             control_db,
@@ -339,16 +351,22 @@ async fn drain_dispatch_batch(
         };
 
         dispatched += 1;
+        *dispatched_by_alias
+            .entry(database_alias.clone())
+            .or_default() += 1;
         debug!(
             run_id = %run.id,
             run_key = %run.run_key,
             run_shard = run.run_shard,
+            database_alias = %database_alias,
+            routing_decision = "dispatch_window_claimed",
             "claimed dispatchable run shard window"
         );
         info!(
             run_id = %run.id,
             run_key = %run.run_key,
             run_shard = run.run_shard,
+            database_alias = %database_alias,
             chunk_events_enqueued = run.chunk_events_enqueued,
             chunks_marked_dispatched = run.chunks_marked_dispatched,
             run_started_events_enqueued = run.run_started_events_enqueued,
@@ -364,6 +382,14 @@ async fn drain_dispatch_batch(
             dispatch_windows_prepared = dispatched,
             "completed coordinator dispatch drain pass"
         );
+        for (alias, alias_dispatched) in dispatched_by_alias {
+            info!(
+                coordinator_id = %coordinator_id,
+                database_alias = %alias,
+                dispatch_windows_prepared = alias_dispatched,
+                "completed coordinator dispatch drain pass for execution placement"
+            );
+        }
     }
 
     Ok(dispatched)
@@ -491,6 +517,7 @@ async fn collect_run_shard_summaries(
 ) -> anyhow::Result<Vec<run_shard_summary::RunShardSummary>> {
     let routes = database.execution_routes_for_run(run_id).await?;
     let mut summaries = Vec::with_capacity(routes.len());
+    let mut summaries_by_alias = std::collections::BTreeMap::<String, usize>::new();
 
     for (run_shard, alias, db) in routes {
         let Some(summary) =
@@ -514,8 +541,18 @@ async fn collect_run_shard_summaries(
             expected_execution_count = summary.expected_execution_count,
             "loaded run shard summary for finalization"
         );
+        *summaries_by_alias.entry(alias).or_default() += 1;
         summaries.push(summary);
     }
 
+    for (alias, count) in summaries_by_alias {
+        debug!(
+            run_id = %run_id,
+            database_alias = %alias,
+            shard_summaries_loaded = count,
+            "loaded routed shard summaries for finalization from execution placement"
+        );
+    }
+
     Ok(summaries)
 }

vigilo/src/cli/commands/shard.rs

@@ -40,6 +40,16 @@ pub(crate) enum SubCommand {
         command: PlacementSubCommand,
     },
 
+    /// Inspect the resolved route for one run shard
+    Route {
+        /// Run UUID
+        run_id: String,
+
+        /// Logical run shard
+        #[arg(value_parser = clap::value_parser!(i16).range(0..=127))]
+        run_shard: i16,
+    },
+
     /// Move one run shard to another database placement
     Move {
         /// Run UUID
@@ -160,6 +170,9 @@ impl Executable for Command {
                 .await
             }
             SubCommand::Placements { command } => exec_placement_command(context, command).await,
+            SubCommand::Route { run_id, run_shard } => {
+                exec_route_command(context, run_id, run_shard).await
+            }
         }
     }
 }
@@ -239,6 +252,28 @@ async fn exec_move_command(
     Ok(())
 }
 
+async fn exec_route_command(
+    context: Context,
+    run_id: String,
+    run_shard: i16,
+) -> anyhow::Result<()> {
+    let run_id = parse_run_id(&run_id)?;
+    let database = context.db().await?;
+    let out = context.out().await?;
+    let route = shard_admin::inspect_shard_route(database, run_id, run_shard).await?;
+
+    info!(
+        run_id = %run_id,
+        run_shard,
+        database_alias = %route.database_alias,
+        placement_status = %route.shard_placement_status,
+        routing_decision = route.routing_decision,
+        "inspected shard route"
+    );
+    out.write_value(&route_payload(&route))?;
+    Ok(())
+}
+
 async fn exec_placement_command(
     context: Context,
     command: PlacementSubCommand,
@@ -378,6 +413,24 @@ fn move_payload(outcome: &shard_admin::ShardMoveOutcome) -> Value {
     })
 }
 
+fn route_payload(route: &shard_admin::ShardRouteInspection) -> Value {
+    json!({
+        "data": {
+            "run_id": route.run_id,
+            "run_shard": route.run_shard,
+            "database_alias": route.database_alias,
+            "shard_placement_status": route.shard_placement_status,
+            "database_role": route.database_role,
+            "database_status": route.database_status,
+            "database_url_env": route.database_url_env,
+            "database_url_env_resolved": route.database_url_env_resolved,
+            "dispatchable": route.dispatchable,
+            "readable": route.readable,
+            "routing_decision": route.routing_decision,
+        }
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use chrono::Utc;
@@ -442,6 +495,23 @@ mod tests {
         assert_eq!(alias, "shard_001");
     }
 
+    #[test]
+    fn route_command_matches_argument_shape() {
+        let run_id = Uuid::now_v7().to_string();
+        let cli = TestCli::try_parse_from(["vigilo", "route", &run_id, "4"]).unwrap();
+
+        let SubCommand::Route {
+            run_id: parsed_run_id,
+            run_shard,
+        } = cli.command
+        else {
+            panic!("expected shard route command");
+        };
+
+        assert_eq!(parsed_run_id, run_id);
+        assert_eq!(run_shard, 4);
+    }
+
     #[test]
     fn database_list_payload_reports_count() {
         let placement = DatabasePlacement {
@@ -531,4 +601,39 @@ mod tests {
         assert_eq!(payload["meta"]["verified"], json!(true));
         assert_eq!(payload["data"]["tables"][0]["table"], json!("run_chunks"));
     }
+
+    #[test]
+    fn route_payload_excludes_secret_url_value() {
+        let run_id = Uuid::now_v7();
+        let route = shard_admin::ShardRouteInspection {
+            run_id,
+            run_shard: 4,
+            database_alias: "shard_001".to_string(),
+            shard_placement_status: "active".to_string(),
+            database_role: "shard".to_string(),
+            database_status: "active".to_string(),
+            database_url_env: "VIGILO_SHARD_001_DATABASE_URL".to_string(),
+            database_url_env_resolved: true,
+            dispatchable: true,
+            readable: true,
+            routing_decision: "dispatchable",
+        };
+
+        let payload = route_payload(&route);
+
+        assert_eq!(payload["data"]["run_id"], json!(run_id));
+        assert_eq!(payload["data"]["run_shard"], json!(4));
+        assert_eq!(payload["data"]["database_alias"], json!("shard_001"));
+        assert_eq!(
+            payload["data"]["database_url_env"],
+            json!("VIGILO_SHARD_001_DATABASE_URL")
+        );
+        assert!(
+            payload
+                .to_string()
+                .contains("VIGILO_SHARD_001_DATABASE_URL")
+        );
+        assert!(!payload.to_string().contains("postgres://"));
+        assert_eq!(payload["data"]["routing_decision"], json!("dispatchable"));
+    }
 }

vigilo/src/context/database.rs

@@ -202,6 +202,14 @@ impl Db {
         let key = ShardPlacementKey { run_id, run_shard };
         if let Some(placement) = self.shard_placement_cache.get(&key).await {
             self.validate_shard_placement_alias(&placement).await?;
+            debug!(
+                run_id = %run_id,
+                run_shard,
+                database_alias = %placement.database_alias,
+                placement_status = %placement.status,
+                routing_decision = "cached_execution_placement",
+                "resolved execution placement from cache"
+            );
             return Ok(placement);
         }
 
@@ -213,6 +221,14 @@ impl Db {
         };
 
         self.validate_shard_placement_alias(&placement).await?;
+        debug!(
+            run_id = %run_id,
+            run_shard,
+            database_alias = %placement.database_alias,
+            placement_status = %placement.status,
+            routing_decision = "control_lookup_execution_placement",
+            "resolved execution placement from control metadata"
+        );
         self.shard_placement_cache
             .insert(key, placement.clone())
             .await;
@@ -229,6 +245,14 @@ impl Db {
         let placement = self.execution_placement(run_id, run_shard).await?;
 
         if !placement.is_dispatchable() {
+            debug!(
+                run_id = %run_id,
+                run_shard,
+                database_alias = %placement.database_alias,
+                placement_status = %placement.status,
+                routing_decision = "blocked_non_dispatchable",
+                "execution placement is not dispatchable"
+            );
             return Err(ExecutionRouteError::NonDispatchableShardPlacement {
                 run_id,
                 run_shard,
@@ -237,6 +261,14 @@ impl Db {
             .into());
         }
 
+        debug!(
+            run_id = %run_id,
+            run_shard,
+            database_alias = %placement.database_alias,
+            placement_status = %placement.status,
+            routing_decision = "dispatchable_execution_pool",
+            "resolved dispatchable execution pool"
+        );
         self.placement(&placement.database_alias).await
     }
 
@@ -253,6 +285,14 @@ impl Db {
         for placement in placements {
             self.validate_shard_placement_alias(&placement).await?;
             if !placement.is_dispatchable() {
+                debug!(
+                    run_id = %run_id,
+                    run_shard = placement.run_shard,
+                    database_alias = %placement.database_alias,
+                    placement_status = %placement.status,
+                    routing_decision = "blocked_non_dispatchable",
+                    "execution route is not dispatchable"
+                );
                 anyhow::bail!(
                     "shard placement for run {} shard {} has status {}, which is not dispatchable",
                     run_id,
@@ -262,6 +302,14 @@ impl Db {
             }
 
             let pool = self.placement(&placement.database_alias).await?.clone();
+            debug!(
+                run_id = %run_id,
+                run_shard = placement.run_shard,
+                database_alias = %placement.database_alias,
+                placement_status = %placement.status,
+                routing_decision = "dispatchable_execution_route",
+                "resolved dispatchable execution route"
+            );
             routed.push((placement.run_shard, placement.database_alias, pool));
         }
 
@@ -284,6 +332,14 @@ impl Db {
         for placement in placements {
             self.validate_shard_placement_alias(&placement).await?;
             let pool = self.placement(&placement.database_alias).await?.clone();
+            debug!(
+                run_id = %run_id,
+                run_shard = placement.run_shard,
+                database_alias = %placement.database_alias,
+                placement_status = %placement.status,
+                routing_decision = "readable_execution_route",
+                "resolved readable execution route"
+            );
             routed.push((placement.run_shard, placement.database_alias, pool));
         }
 
@@ -364,6 +420,10 @@ impl Db {
         })
     }
 
+    pub(crate) fn database_url_env_is_resolved(&self, database_url_env: &str) -> bool {
+        database_url_env == DEFAULT_DATABASE_URL_ENV || std::env::var_os(database_url_env).is_some()
+    }
+
     #[allow(dead_code)]
     async fn placement_catalog(&self) -> anyhow::Result<&PlacementCatalog> {
         self.placement_catalog

vigilo/src/db/workflows/run_dispatch.rs

@@ -54,6 +54,8 @@ pub(crate) struct DispatchRunSnapshot {
 pub(crate) struct DispatchRoute {
     pub(crate) run_id: Uuid,
     pub(crate) run_shard: i16,
+    pub(crate) database_alias: String,
+    pub(crate) placement_status: String,
 }
 
 #[derive(Debug, Clone, sqlx::FromRow)]
@@ -279,7 +281,9 @@ pub(crate) async fn select_next_dispatch_route(
         r#"
         SELECT
             c.run_id,
-            c.run_shard
+            c.run_shard,
+            sp.database_alias,
+            sp.status AS placement_status
         FROM run_shard_dispatch_cursors c
         JOIN runs r
           ON r.id = c.run_id
@@ -1143,6 +1147,8 @@ mod tests {
 
         assert_eq!(route.run_id, run_id);
         assert_eq!(route.run_shard, 1);
+        assert_eq!(route.database_alias, "primary");
+        assert_eq!(route.placement_status, "active");
     }
 
     #[sqlx::test(migrations = "../migrations")]
@@ -1152,6 +1158,8 @@ mod tests {
         let route = DispatchRoute {
             run_id,
             run_shard: 1,
+            database_alias: "primary".to_string(),
+            placement_status: "active".to_string(),
         };
         let snapshot = prepare_dispatch_run_snapshot(&pool, &route, Uuid::now_v7(), 60)
             .await