More logging for drivers and coordinator

Author: joshua-spacetime

tools/tpcc-runner/src/coordinator.rs

@@ -7,6 +7,7 @@ use std::collections::BTreeMap;
 use std::fs;
 use std::path::Path;
 use std::sync::Arc;
+use std::time::Duration;
 
 use crate::config::CoordinatorConfig;
 use crate::metrics_module_bindings::reset_reducer::reset;
@@ -15,7 +16,9 @@ use crate::protocol::{
     DriverAssignment, RegisterDriverRequest, RegisterDriverResponse, RunSchedule, ScheduleResponse,
     SubmitSummaryRequest,
 };
-use crate::summary::{aggregate_summaries, now_millis, write_json, AggregateSummary, DriverSummary};
+use crate::summary::{
+    aggregate_summaries, log_aggregate_summary, now_millis, write_json, AggregateSummary, DriverSummary,
+};
 
 #[derive(Clone)]
 struct AppState {
@@ -66,8 +69,8 @@ async fn register_driver(
     Json(request): Json<RegisterDriverRequest>,
 ) -> Json<RegisterDriverResponse> {
     let mut inner = state.inner.lock();
-    let assignment = match inner.registrations.get(&request.driver_id) {
-        Some(existing) => existing.assignment.clone(),
+    let (assignment, is_new_registration) = match inner.registrations.get(&request.driver_id) {
+        Some(existing) => (existing.assignment.clone(), false),
         None => {
             if inner.registration_order.len() >= inner.config.expected_drivers {
                 return Json(RegisterDriverResponse {
@@ -84,10 +87,30 @@ async fn register_driver(
                     assignment: assignment.clone(),
                 },
             );
-            assignment
+            (assignment, true)
         }
     };
     maybe_create_schedule(&mut inner);
+    let registered = inner.registrations.len();
+    let warehouse_end = assignment_end(&assignment);
+    if is_new_registration {
+        log::info!(
+            "driver {} registered and ready ({}/{}): warehouses {}..={} ({} warehouse(s))",
+            request.driver_id,
+            registered,
+            inner.config.expected_drivers,
+            assignment.warehouse_start,
+            warehouse_end,
+            assignment.driver_warehouse_count
+        );
+    } else {
+        log::info!(
+            "driver {} re-registered and remains ready: warehouses {}..={}",
+            request.driver_id,
+            assignment.warehouse_start,
+            warehouse_end
+        );
+    }
     Json(RegisterDriverResponse {
         accepted: true,
         assignment: Some(assignment),
@@ -108,16 +131,32 @@ async fn submit_summary(
 ) -> Result<Json<AggregateSummary>, axum::http::StatusCode> {
     let aggregate = {
         let mut inner = state.inner.lock();
-        inner
+        let replaced = inner
             .summaries
-            .insert(request.summary.driver_id.clone(), request.summary.clone());
-        if inner.summaries.len() == inner.config.expected_drivers {
+            .insert(request.summary.driver_id.clone(), request.summary.clone())
+            .is_some();
+        let received = inner.summaries.len();
+        log::info!(
+            "received summary from driver {} ({}/{}{})",
+            request.summary.driver_id,
+            received,
+            inner.config.expected_drivers,
+            if replaced { ", replaced existing summary" } else { "" }
+        );
+        if received == inner.config.expected_drivers {
             let summaries: Vec<_> = inner.summaries.values().cloned().collect();
             let aggregate = aggregate_summaries(inner.config.run_id.clone(), &summaries);
+            let summary_path = aggregate_summary_path(&inner.config.output_dir, &aggregate);
             if let Err(err) = write_aggregate(&inner.config.output_dir, &aggregate) {
                 log::error!("failed to write aggregate summary: {err:#}");
                 return Err(axum::http::StatusCode::INTERNAL_SERVER_ERROR);
             }
+            log::info!(
+                "received all {} summary(s) for run {}",
+                inner.config.expected_drivers,
+                inner.config.run_id
+            );
+            log_aggregate_summary(&aggregate, &summary_path);
             aggregate
         } else {
             aggregate_summaries(
@@ -142,18 +181,23 @@ fn maybe_create_schedule(inner: &mut CoordinatorState) {
         .reducers
         .reset(inner.config.warmup_secs * 1000, measure_start_ms, measure_end_ms);
 
-    inner.schedule = Some(RunSchedule {
+    let schedule = RunSchedule {
         run_id: inner.config.run_id.clone(),
         warmup_start_ms,
         measure_start_ms,
         measure_end_ms,
         stop_ms: measure_end_ms,
-    });
+    };
+    inner.schedule = Some(schedule.clone());
     log::info!(
-        "all {} driver(s) registered; schedule ready for run {}",
+        "all {} driver(s) registered; schedule ready for run {} (warmup_start_ms={} measure_start_ms={} measure_end_ms={})",
         inner.config.expected_drivers,
-        inner.config.run_id
+        inner.config.run_id,
+        warmup_start_ms,
+        measure_start_ms,
+        measure_end_ms
     );
+    tokio::spawn(log_schedule_events(schedule));
 }
 
 fn assignment_for_index(config: &CoordinatorConfig, index: usize) -> DriverAssignment {
@@ -173,7 +217,35 @@ fn assignment_for_index(config: &CoordinatorConfig, index: usize) -> DriverAssig
 }
 
 fn write_aggregate(output_dir: &Path, aggregate: &AggregateSummary) -> Result<()> {
-    let run_dir = output_dir.join(&aggregate.run_id);
-    fs::create_dir_all(&run_dir).with_context(|| format!("failed to create {}", run_dir.display()))?;
-    write_json(&run_dir.join("summary.json"), aggregate)
+    let summary_path = aggregate_summary_path(output_dir, aggregate);
+    if let Some(run_dir) = summary_path.parent() {
+        fs::create_dir_all(run_dir).with_context(|| format!("failed to create {}", run_dir.display()))?;
+    }
+    write_json(&summary_path, aggregate)
+}
+
+fn aggregate_summary_path(output_dir: &Path, aggregate: &AggregateSummary) -> std::path::PathBuf {
+    output_dir.join(&aggregate.run_id).join("summary.json")
+}
+
+fn assignment_end(assignment: &DriverAssignment) -> u32 {
+    assignment
+        .warehouse_start
+        .saturating_add(assignment.driver_warehouse_count.saturating_sub(1))
+}
+
+async fn log_schedule_events(schedule: RunSchedule) {
+    sleep_until_ms(schedule.warmup_start_ms).await;
+    log::info!("run {} warmup started", schedule.run_id);
+    sleep_until_ms(schedule.measure_start_ms).await;
+    log::info!("run {} measurement started", schedule.run_id);
+    sleep_until_ms(schedule.measure_end_ms).await;
+    log::info!("run {} measurement ended", schedule.run_id);
+}
+
+async fn sleep_until_ms(target_ms: u64) {
+    let now_ms = now_millis();
+    if target_ms > now_ms {
+        tokio::time::sleep(Duration::from_millis(target_ms - now_ms)).await;
+    }
 }

tools/tpcc-runner/src/driver.rs

@@ -1,7 +1,7 @@
 use anyhow::{anyhow, bail, Context, Result};
 use rand::{rngs::StdRng, Rng, SeedableRng};
 use std::fs;
-use std::path::{Path, PathBuf};
+use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -15,14 +15,18 @@ use crate::module_bindings::*;
 use crate::protocol::{
     RegisterDriverRequest, RegisterDriverResponse, RunSchedule, ScheduleResponse, SubmitSummaryRequest,
 };
-use crate::summary::{write_json, DriverSummary, DriverSummaryMeta, SharedMetrics, TransactionKind, TransactionRecord};
+use crate::summary::{
+    log_driver_summary, write_json, DriverSummary, DriverSummaryMeta, SharedMetrics, TransactionKind,
+    TransactionRecord,
+};
 use crate::topology::DatabaseTopology;
 use crate::tpcc::*;
 
 struct TerminalRuntime {
     config: DriverConfig,
     metrics: SharedMetrics,
     abort: Arc<AtomicBool>,
+    start_logged: Arc<AtomicBool>,
     request_ids: Arc<AtomicU64>,
     schedule: RunSchedule,
     run_constants: RunConstants,
@@ -60,9 +64,22 @@ pub async fn run(config: DriverConfig) -> Result<()> {
     };
 
     let abort = Arc::new(AtomicBool::new(false));
+    let start_logged = Arc::new(AtomicBool::new(false));
     let request_ids = Arc::new(AtomicU64::new(1));
     let mut tasks = JoinSet::new();
 
+    log::info!(
+        "driver {} ready for run {}: warehouses {}..={} terminals={} warmup_start_ms={} measure_start_ms={} measure_end_ms={}",
+        config.driver_id,
+        run_id,
+        config.warehouse_start,
+        config.warehouse_end(),
+        config.terminals(),
+        schedule.warmup_start_ms,
+        schedule.measure_start_ms,
+        schedule.measure_end_ms
+    );
+
     for warehouse_id in config.warehouse_start..=config.warehouse_end() {
         let database_identity = topology.identity_for_warehouse(warehouse_id)?;
         for district_id in 1..=DISTRICTS_PER_WAREHOUSE {
@@ -75,13 +92,15 @@ pub async fn run(config: DriverConfig) -> Result<()> {
             let terminal_config = config.clone();
             let terminal_metrics = metrics.clone();
             let terminal_abort = abort.clone();
+            let terminal_start_logged = start_logged.clone();
             let terminal_constants = run_constants.clone();
             let terminal_schedule = schedule.clone();
             let terminal_request_ids = request_ids.clone();
             let runtime = TerminalRuntime {
                 config: terminal_config,
                 metrics: terminal_metrics,
                 abort: terminal_abort,
+                start_logged: terminal_start_logged,
                 request_ids: terminal_request_ids,
                 schedule: terminal_schedule,
                 run_constants: terminal_constants,
@@ -131,7 +150,7 @@ pub async fn run(config: DriverConfig) -> Result<()> {
         measure_end_ms: schedule.measure_end_ms,
     })?;
     write_json(&summary_path, &summary)?;
-    print_summary(&summary, &summary_path, &events_path);
+    log_driver_summary(&summary, &summary_path, &events_path);
 
     if let Some(coordinator_url) = &config.coordinator_url {
         submit_summary(coordinator_url, summary).await?;
@@ -145,6 +164,7 @@ async fn run_terminal(runtime: TerminalRuntime) -> Result<()> {
         config,
         metrics,
         abort,
+        start_logged,
         request_ids,
         schedule,
         run_constants,
@@ -154,7 +174,23 @@ async fn run_terminal(runtime: TerminalRuntime) -> Result<()> {
     } = runtime;
     let client = ModuleClient::connect_async(&config.connection, database_identity).await?;
     let metrics_client = connect_metrics_module_async(&config.connection).await?;
+    log::info!(
+        "driver {} terminal {} connected to {} for warehouse {} district {}",
+        config.driver_id,
+        assignment.terminal_id,
+        database_identity,
+        assignment.warehouse_id,
+        assignment.district_id
+    );
     sleep_until_ms_async(schedule.warmup_start_ms).await;
+    if !start_logged.swap(true, Ordering::Relaxed) {
+        log::info!(
+            "driver {} starting run {} with {} terminal(s)",
+            config.driver_id,
+            schedule.run_id,
+            config.terminals()
+        );
+    }
 
     let mut rng = StdRng::seed_from_u64(seed);
     while !abort.load(Ordering::Relaxed) {
@@ -683,32 +719,6 @@ fn resolve_output_dir(config: &DriverConfig, run_id: &str) -> PathBuf {
     }
 }
 
-fn print_summary(summary: &DriverSummary, summary_path: &Path, events_path: &Path) {
-    log::info!("run_id={}", summary.run_id);
-    log::info!("driver_id={}", summary.driver_id);
-    log::info!("tpmc_like={:.2}", summary.tpmc_like);
-    log::info!("total_transactions={}", summary.total_transactions);
-    for (name, txn) in &summary.transactions {
-        log::info!(
-            "{} count={} success={} failure={} p95_ms={} p99_ms={}",
-            name,
-            txn.count,
-            txn.success,
-            txn.failure,
-            txn.p95_latency_ms,
-            txn.p99_latency_ms
-        );
-    }
-    log::info!(
-        "delivery queued={} completed={} pending={}",
-        summary.delivery.queued,
-        summary.delivery.completed,
-        summary.delivery.pending
-    );
-    log::info!("summary={}", summary_path.display());
-    log::info!("events={}", events_path.display());
-}
-
 async fn sleep_until_ms_async(target_ms: u64) {
     let now_ms = crate::summary::now_millis();
     if target_ms > now_ms {

tools/tpcc-runner/src/summary.rs

@@ -572,6 +572,57 @@ pub fn aggregate_summaries(run_id: String, summaries: &[DriverSummary]) -> Aggre
     }
 }
 
+pub fn log_driver_summary(summary: &DriverSummary, summary_path: &Path, events_path: &Path) {
+    log::info!("run_id={}", summary.run_id);
+    log::info!("driver_id={}", summary.driver_id);
+    log::info!("tpmc_like={:.2}", summary.tpmc_like);
+    log::info!("total_transactions={}", summary.total_transactions);
+    for (name, txn) in &summary.transactions {
+        log::info!(
+            "{} count={} success={} failure={} p95_ms={} p99_ms={}",
+            name,
+            txn.count,
+            txn.success,
+            txn.failure,
+            txn.p95_latency_ms,
+            txn.p99_latency_ms
+        );
+    }
+    log::info!(
+        "delivery queued={} completed={} pending={}",
+        summary.delivery.queued,
+        summary.delivery.completed,
+        summary.delivery.pending
+    );
+    log::info!("summary={}", summary_path.display());
+    log::info!("events={}", events_path.display());
+}
+
+pub fn log_aggregate_summary(summary: &AggregateSummary, summary_path: &Path) {
+    log::info!("run_id={}", summary.run_id);
+    log::info!("driver_count={}", summary.driver_count);
+    log::info!("tpmc_like={:.2}", summary.tpmc_like);
+    log::info!("total_transactions={}", summary.total_transactions);
+    for (name, txn) in &summary.transactions {
+        log::info!(
+            "{} count={} success={} failure={} p95_ms={} p99_ms={}",
+            name,
+            txn.count,
+            txn.success,
+            txn.failure,
+            txn.p95_latency_ms,
+            txn.p99_latency_ms
+        );
+    }
+    log::info!(
+        "delivery queued={} completed={} pending={}",
+        summary.delivery.queued,
+        summary.delivery.completed,
+        summary.delivery.pending
+    );
+    log::info!("summary={}", summary_path.display());
+}
+
 pub fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<()> {
     let file = File::create(path).with_context(|| format!("failed to create {}", path.display()))?;
     serde_json::to_writer_pretty(file, value).with_context(|| format!("failed to write {}", path.display()))