diff --git a/.gitea/workflows/deploy.yaml b/.gitea/workflows/deploy.yaml index 0651d53..a909925 100644 --- a/.gitea/workflows/deploy.yaml +++ b/.gitea/workflows/deploy.yaml @@ -307,6 +307,13 @@ jobs: else echo " ok connected; miner_gpu_devices ${devs%.*}" fi + # The build-info gauge is what ties every other metric to a commit + # (quantus/miner#9). In deploy mode it must carry this commit. + if grep -q "^miner_build_info{.*commit=\"${{ github.sha }}\"" <<<"$m"; then + echo " ok miner_build_info carries ${{ github.sha }}" + elif [ "${{ github.event.inputs.mode }}" != validate ]; then + echo " miner_build_info does not carry commit ${{ github.sha }}" >&2; fail=1 + fi deadline=$((SECONDS + 120)) h1=""; h2=""; ok=0 diff --git a/Cargo.lock b/Cargo.lock index 354935a..0537959 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -613,6 +613,7 @@ dependencies = [ "futures", "hex", "log", + "metrics", "pow-core", "primitive-types 0.13.1", "qp-plonky2", diff --git a/crates/engine-gpu/Cargo.toml b/crates/engine-gpu/Cargo.toml index 8132411..4adcd00 100644 --- a/crates/engine-gpu/Cargo.toml +++ b/crates/engine-gpu/Cargo.toml @@ -19,6 +19,7 @@ simd-poseidon2 = [] [dependencies] engine-cpu = { path = "../engine-cpu" } +metrics = { path = "../metrics" } # lair: per-device metrics (quantus/miner#9) pow-core = { path = "../pow-core" } primitive-types = { workspace = true } log = { workspace = true } diff --git a/crates/engine-gpu/src/lib.rs b/crates/engine-gpu/src/lib.rs index b436b93..3973750 100644 --- a/crates/engine-gpu/src/lib.rs +++ b/crates/engine-gpu/src/lib.rs @@ -20,6 +20,8 @@ use std::sync::{ /// Represents a single GPU device context. struct GpuContext { + // lair: per-device metric handles, labelled with the kernel id (quantus/miner#9). + metrics: metrics::DeviceMetrics, device: wgpu::Device, queue: wgpu::Queue, pipeline: wgpu::ComputePipeline, @@ -443,6 +445,7 @@ impl GpuEngine { initialized.push(InitializedGpu { context: Arc::new(GpuContext { + metrics: metrics::DeviceMetrics::new(i, kernel.id()), device, queue, pipeline, @@ -698,8 +701,9 @@ impl MinerEngine for GpuEngine { total_hashes += hash_count; } BatchResult::DeviceLost => { - // GPU device is lost/unresponsive - mark as permanently dead - // and clear resources to prevent "buffer already mapped" panics + gpu_ctx.metrics.record_device_lost(); // lair + // GPU device is lost/unresponsive - mark as permanently dead + // and clear resources to prevent "buffer already mapped" panics DEVICE_LOST.with(|lost| *lost.borrow_mut() = Some(self.engine_id)); WORKER_RESOURCES.with(|res| *res.borrow_mut() = None); @@ -793,7 +797,8 @@ fn run_single_batch( batch_start: U512, batch_size: u32, ) -> BatchResult { - // Calculate dispatch configuration for this batch + let batch_start_at = std::time::Instant::now(); // lair: batch phase timing + // Calculate dispatch configuration for this batch let threads_per_workgroup = 256u32; let limits = gpu_ctx.device.limits(); let max_workgroups = limits.max_compute_workgroups_per_dimension; @@ -865,6 +870,7 @@ fn run_single_batch( ); gpu_ctx.queue.submit(Some(encoder.finish())); + let submitted_at = std::time::Instant::now(); // lair // Wait for GPU to complete (blocking) let buffer_slice = resources.staging_buffer.slice(..); @@ -911,6 +917,12 @@ fn run_single_batch( // Only reach here if final_status == 1 (success), buffer is mapped debug_assert_eq!(final_status, 1); + // lair: gpu = submit to mapped; host = the rest of the batch so far plus + // the readback below, which is small and constant. + let gpu_time = submitted_at.elapsed(); + gpu_ctx + .metrics + .observe_batch(gpu_time, batch_start_at.elapsed() - gpu_time); // Read results let data = buffer_slice.get_mapped_range(); @@ -939,6 +951,8 @@ fn run_single_batch( drop(data); resources.staging_buffer.unmap(); + gpu_ctx.metrics.record_hashes(hashes_computed); // lair + gpu_ctx.metrics.record_solution(); // lair return BatchResult::Found { candidate: Candidate { nonce, work, hash }, hash_count: hashes_computed, @@ -948,6 +962,7 @@ fn run_single_batch( drop(data); resources.staging_buffer.unmap(); + gpu_ctx.metrics.record_hashes(dispatched_nonces); // lair BatchResult::NotFound { hash_count: dispatched_nonces, } diff --git a/crates/metrics/src/lair.rs b/crates/metrics/src/lair.rs new file mode 100644 index 0000000..790fb72 --- /dev/null +++ b/crates/metrics/src/lair.rs @@ -0,0 +1,340 @@ +//! lair: metrics that attribute performance to builds, devices and jobs +//! (quantus/miner#9). Additive to origin's metrics in `lib.rs`; nothing there +//! changes, so the fleet dashboard keeps working across origin merges. +//! +//! Naming: `miner_*` like origin's. Per-device series carry `device` (engine +//! index) and `kernel` (the kernel id the engine selected), so a silent +//! fallback from one kernel to another shows up as a label change rather than +//! as an unexplained hashrate drop. + +use crate::REGISTRY; +use once_cell::sync::Lazy; +use prometheus::{ + Counter, Histogram, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, + IntGaugeVec, Opts, +}; +use std::time::Duration; + +fn reg(m: M) -> M { + REGISTRY + .register(Box::new(m.clone())) + .expect("register lair metric"); + m +} + +// --------------------------------------------------------------------------- +// Identity: which build and which configuration produced the numbers +// --------------------------------------------------------------------------- + +static BUILD_INFO: Lazy = Lazy::new(|| { + reg(IntGaugeVec::new( + Opts::new( + "miner_build_info", + "Build identity of the running miner, always 1", + ), + &["version", "commit"], + ) + .expect("miner_build_info")) +}); + +static CONFIG_INFO: Lazy = Lazy::new(|| { + reg(IntGaugeVec::new( + Opts::new( + "miner_config_info", + "Effective mining configuration, always 1", + ), + &[ + "engine", + "gpu_batch_size", + "gpu_devices", + "cpu_workers", + "gpu_throttle_ms", + ], + ) + .expect("miner_config_info")) +}); + +/// Set once at startup. `commit` is the SHA embedded by build.rs. +pub fn set_build_info(version: &str, commit: &str) { + BUILD_INFO.with_label_values(&[version, commit]).set(1); +} + +/// Set once the engines are resolved. Two deploys of one commit with different +/// flags must not look like the same experiment. +pub fn set_config_info( + engine: &str, + gpu_batch_size: u32, + gpu_devices: usize, + cpu_workers: usize, + gpu_throttle_ms: u64, +) { + CONFIG_INFO + .with_label_values(&[ + engine, + &gpu_batch_size.to_string(), + &gpu_devices.to_string(), + &cpu_workers.to_string(), + &gpu_throttle_ms.to_string(), + ]) + .set(1); +} + +// --------------------------------------------------------------------------- +// Per device: throughput, outcomes, and where the time goes inside a batch +// --------------------------------------------------------------------------- + +static DEVICE_HASHES: Lazy = Lazy::new(|| { + reg(IntCounterVec::new( + Opts::new( + "miner_device_hashes_total", + "Hashes computed, per GPU device and kernel", + ), + &["device", "kernel"], + ) + .expect("miner_device_hashes_total")) +}); + +static DEVICE_SOLUTIONS: Lazy = Lazy::new(|| { + reg(IntCounterVec::new( + Opts::new( + "miner_device_solutions_total", + "Solutions found, per GPU device and kernel", + ), + &["device", "kernel"], + ) + .expect("miner_device_solutions_total")) +}); + +static DEVICE_LOST: Lazy = Lazy::new(|| { + reg(IntCounterVec::new( + Opts::new( + "miner_device_lost_total", + "Times a GPU device was lost or unresponsive and its worker stopped", + ), + &["device", "kernel"], + ) + .expect("miner_device_lost_total")) +}); + +static GPU_BATCH_SECONDS: Lazy = Lazy::new(|| { + reg(HistogramVec::new( + HistogramOpts::new( + "miner_gpu_batch_seconds", + "Per-batch time split into the GPU executing (phase=gpu) and the host \ + preparing, submitting and reading back (phase=host)", + ) + .buckets(vec![ + 0.0005, 0.001, 0.002, 0.004, 0.008, 0.016, 0.032, 0.064, 0.128, 0.256, 0.512, 1.0, 2.0, + ]), + &["device", "kernel", "phase"], + ) + .expect("miner_gpu_batch_seconds")) +}); + +/// Handles for one device, resolved once so the per-batch path does no label +/// lookups. Created by the engine when it initialises a device. +#[derive(Clone)] +pub struct DeviceMetrics { + hashes: IntCounter, + solutions: IntCounter, + lost: IntCounter, + batch_gpu: Histogram, + batch_host: Histogram, +} + +impl DeviceMetrics { + pub fn new(device: usize, kernel: &str) -> Self { + let d = device.to_string(); + Self { + hashes: DEVICE_HASHES.with_label_values(&[&d, kernel]), + solutions: DEVICE_SOLUTIONS.with_label_values(&[&d, kernel]), + lost: DEVICE_LOST.with_label_values(&[&d, kernel]), + batch_gpu: GPU_BATCH_SECONDS.with_label_values(&[&d, kernel, "gpu"]), + batch_host: GPU_BATCH_SECONDS.with_label_values(&[&d, kernel, "host"]), + } + } + + pub fn record_hashes(&self, n: u64) { + self.hashes.inc_by(n); + } + + pub fn record_solution(&self) { + self.solutions.inc(); + } + + pub fn record_device_lost(&self) { + self.lost.inc(); + } + + /// `gpu` is submit-to-completion on the device; `host` is everything else + /// in the batch (buffer writes, encoding, readback, bookkeeping). + pub fn observe_batch(&self, gpu: Duration, host: Duration) { + self.batch_gpu.observe(gpu.as_secs_f64()); + self.batch_host.observe(host.as_secs_f64()); + } +} + +// --------------------------------------------------------------------------- +// Jobs and results: the efficiency side of every throughput trade-off +// --------------------------------------------------------------------------- + +static JOBS_RECEIVED: Lazy = Lazy::new(|| { + reg( + IntCounter::new("miner_jobs_received_total", "Jobs received from the node") + .expect("miner_jobs_received_total"), + ) +}); + +static STALE_HASHES: Lazy = Lazy::new(|| { + reg(IntCounterVec::new( + Opts::new( + "miner_stale_hashes_total", + "Hashes reported for a job after a newer job had been issued (wasted work)", + ), + &["engine"], + ) + .expect("miner_stale_hashes_total")) +}); + +static JOB_PICKUP_SECONDS: Lazy = Lazy::new(|| { + reg(HistogramVec::new( + HistogramOpts::new( + "miner_job_pickup_seconds", + "Job issued to a worker starting on it. For a busy worker this is the \ + time to notice cancellation and finish the in-flight batch", + ) + .buckets(vec![ + 0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, + ]), + &["engine"], + ) + .expect("miner_job_pickup_seconds")) +}); + +static RESULTS_SUBMITTED: Lazy = Lazy::new(|| { + reg(IntCounter::new( + "miner_results_submitted_total", + "Solutions sent to the node", + ) + .expect("miner_results_submitted_total")) +}); + +static RESULTS_SEND_FAILED: Lazy = Lazy::new(|| { + reg(IntCounter::new( + "miner_results_send_failed_total", + "Solutions that could not be sent to the node", + ) + .expect("miner_results_send_failed_total")) +}); + +static SEAL_LATENCY: Lazy = Lazy::new(|| { + reg(Histogram::with_opts( + HistogramOpts::new( + "miner_seal_latency_seconds", + "Solution found on a worker to the result sent to the node", + ) + .buckets(vec![ + 0.001, 0.002, 0.005, 0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, + ]), + ) + .expect("miner_seal_latency_seconds")) +}); + +static JOB_IDLE_SECONDS: Lazy = Lazy::new(|| { + reg(Counter::new( + "miner_job_idle_seconds_total", + "Seconds spent between sending a result and receiving the next job (node-attributable idle)", + ) + .expect("miner_job_idle_seconds_total")) +}); + +pub fn record_job_received() { + JOBS_RECEIVED.inc(); +} + +pub fn record_stale_hashes(engine: &str, n: u64) { + STALE_HASHES.with_label_values(&[engine]).inc_by(n); +} + +pub fn observe_job_pickup(engine: &str, d: Duration) { + JOB_PICKUP_SECONDS + .with_label_values(&[engine]) + .observe(d.as_secs_f64()); +} + +pub fn record_result_submitted(found_to_sent: Option) { + RESULTS_SUBMITTED.inc(); + if let Some(d) = found_to_sent { + SEAL_LATENCY.observe(d.as_secs_f64()); + } +} + +pub fn record_result_send_failed() { + RESULTS_SEND_FAILED.inc(); +} + +pub fn record_job_idle(d: Duration) { + JOB_IDLE_SECONDS.inc_by(d.as_secs_f64()); +} + +// --------------------------------------------------------------------------- +// Connection to the node: is a hashrate drop the miner or the node? +// --------------------------------------------------------------------------- + +static CONNECTS: Lazy = Lazy::new(|| { + reg(IntCounter::new( + "miner_connects_total", + "Connections established to the node", + ) + .expect("miner_connects_total")) +}); + +static CONNECT_FAILURES: Lazy = Lazy::new(|| { + reg(IntCounter::new( + "miner_connect_failures_total", + "Connection attempts to the node that failed", + ) + .expect("miner_connect_failures_total")) +}); + +static DISCONNECTS: Lazy = Lazy::new(|| { + reg(IntCounter::new( + "miner_disconnects_total", + "Connections to the node that were lost", + ) + .expect("miner_disconnects_total")) +}); + +static CONNECTED: Lazy = Lazy::new(|| { + reg( + IntGauge::new("miner_connected", "1 while connected to the node, else 0") + .expect("miner_connected"), + ) +}); + +static DISCONNECTED_SECONDS: Lazy = Lazy::new(|| { + reg(Counter::new( + "miner_disconnected_seconds_total", + "Seconds spent without a node connection after the first successful connection", + ) + .expect("miner_disconnected_seconds_total")) +}); + +pub fn record_connected() { + CONNECTS.inc(); + CONNECTED.set(1); +} + +pub fn record_connect_failed() { + CONNECT_FAILURES.inc(); + CONNECTED.set(0); +} + +pub fn record_disconnected() { + DISCONNECTS.inc(); + CONNECTED.set(0); +} + +pub fn record_disconnected_time(d: Duration) { + DISCONNECTED_SECONDS.inc_by(d.as_secs_f64()); +} diff --git a/crates/metrics/src/lib.rs b/crates/metrics/src/lib.rs index d138f54..b263df5 100644 --- a/crates/metrics/src/lib.rs +++ b/crates/metrics/src/lib.rs @@ -39,6 +39,10 @@ use anyhow::Result; static REGISTRY: Lazy = Lazy::new(Registry::new); +// lair: build, device and job metrics (quantus/miner#9); see lair.rs. +mod lair; +pub use lair::*; + // --------------------------------------------------------------------------- // Hash Rate Metrics // --------------------------------------------------------------------------- diff --git a/crates/miner-cli/src/main.rs b/crates/miner-cli/src/main.rs index c8e9ffd..dc59517 100644 --- a/crates/miner-cli/src/main.rs +++ b/crates/miner-cli/src/main.rs @@ -144,6 +144,8 @@ struct Args { #[tokio::main] async fn main() { let args = Args::parse(); + // lair: identify this build in the metrics (quantus/miner#9). + metrics::set_build_info(env!("CARGO_PKG_VERSION"), env!("MINER_BUILD_SHA")); let Some(command) = args.command else { eprintln!("Error: No command provided. Use 'serve' to start mining (defaults to local node at 127.0.0.1:9833)."); diff --git a/crates/miner-service/src/lib.rs b/crates/miner-service/src/lib.rs index f1c91a7..c94677c 100644 --- a/crates/miner-service/src/lib.rs +++ b/crates/miner-service/src/lib.rs @@ -62,6 +62,8 @@ pub struct WorkerResult { pub hash_count: u64, /// Whether this worker has finished its range. pub completed: bool, + /// lair: when the candidate was found (seal latency, quantus/miner#9). + pub found_at: Option, } /// A successful mining candidate. @@ -90,6 +92,8 @@ pub struct MiningJob { pub ctx: pow_core::JobContext, /// Job ID to detect stale results after job transitions pub job_id: u64, + /// lair: when the job was issued (pickup latency, quantus/miner#9). + created_at: std::time::Instant, } /// Persistent worker thread pool that keeps threads alive between jobs. @@ -208,6 +212,7 @@ impl WorkerPool { let job = MiningJob { ctx, job_id: new_job_id, + created_at: std::time::Instant::now(), // lair }; // Dispatch job to all workers using bounded channels (capacity 16). @@ -340,6 +345,9 @@ fn worker_loop( log::debug!("[WORKER {type_str}-{thread_id}] Drained {skipped} stale jobs from queue"); } + // lair: issued-to-picked-up; for a busy worker this is the cancel latency. + metrics::observe_job_pickup(type_str, job.created_at.elapsed()); + // Capture the job's ID for later validation let job_id = job.job_id; log::debug!("[WORKER {type_str}-{thread_id}] Received job {job_id}"); @@ -398,6 +406,7 @@ fn worker_loop( candidate: None, // Discard the stale candidate hash_count, completed: true, + found_at: None, }); continue; } @@ -442,6 +451,7 @@ fn worker_loop( candidate: None, hash_count, completed: true, + found_at: None, }); break; // Exit the worker loop } @@ -452,6 +462,7 @@ fn worker_loop( }; // Send result (non-blocking to avoid deadlock if receiver is full) + let found_at = candidate.as_ref().map(|_| std::time::Instant::now()); // lair let _ = result_tx.try_send(WorkerResult { thread_id, engine_type, @@ -459,6 +470,7 @@ fn worker_loop( candidate, hash_count, completed: true, + found_at, }); } @@ -562,6 +574,14 @@ pub async fn run(config: ServiceConfig) -> anyhow::Result<()> { cpu_workers, gpu_devices ); + // lair: expose the effective configuration as labels (quantus/miner#9). + metrics::set_config_info( + gpu_engine.as_ref().map(|e| e.name()).unwrap_or("cpu"), + config.gpu_batch_size, + gpu_devices, + cpu_workers, + config.gpu_throttle_ms, + ); if let Some(ref engine) = cpu_engine { let name = engine.name(); diff --git a/crates/miner-service/src/quic.rs b/crates/miner-service/src/quic.rs index 6a20abb..e477e39 100644 --- a/crates/miner-service/src/quic.rs +++ b/crates/miner-service/src/quic.rs @@ -38,6 +38,8 @@ pub async fn connect_and_mine( let mut reconnect_delay = Duration::from_secs(1); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(30); + // lair: time without a connection, counted after the first success. + let mut disconnected_since: Option = None; loop { log::info!("⛏️ Connecting to node at {}...", node_addr); @@ -45,6 +47,10 @@ pub async fn connect_and_mine( match establish_connection(node_addr, auth_token, tls_cert_sha256).await { Ok((connection, send, recv)) => { log::info!("⛏️ Connected to node at {}", node_addr); + metrics::record_connected(); // lair + if let Some(t) = disconnected_since.take() { + metrics::record_disconnected_time(t.elapsed()); + } let mut authenticated = false; if let Err(e) = @@ -61,6 +67,8 @@ pub async fn connect_and_mine( } log::info!("⛏️ Connection lost: {}", e); } + metrics::record_disconnected(); // lair + disconnected_since = Some(Instant::now()); // Only clear backoff after the node accepted auth (first NewJob). // connect() already treats explicit "auth failed" as permanent; // this covers any other post-Ready close that looked like success. @@ -76,6 +84,7 @@ pub async fn connect_and_mine( return Err(e); } log::warn!("⛏️ Failed to connect to node: {}", e); + metrics::record_connect_failed(); // lair } } @@ -145,6 +154,8 @@ async fn handle_connection( let mut cpu_hashes: u64 = 0; let mut gpu_hashes: u64 = 0; let mut result_sent_for_current_job = false; + // lair: when the last result went out, for node-attributable idle time. + let mut result_sent_at: Option = None; log::info!("⛏️ Waiting for mining jobs from node..."); @@ -175,6 +186,13 @@ async fn handle_connection( // Check if this result is for the current job (not stale) if worker_result.job_id != internal_job_id { + metrics::record_stale_hashes( + match worker_result.engine_type { + EngineType::Cpu => "CPU", + EngineType::Gpu => "GPU", + }, + worker_result.hash_count, + ); // lair log::debug!( "⏰ Discarding stale result from worker {} (result job_id {} != current {})", worker_result.thread_id, @@ -214,7 +232,15 @@ async fn handle_connection( }; let msg = MinerMessage::JobResult(result); - send_message_checked(&connection, &mut send, &msg).await?; + if let Err(e) = send_message_checked(&connection, &mut send, &msg).await { + metrics::record_result_send_failed(); // lair + return Err(e); + } + // lair: seal latency and the start of the idle window. + metrics::record_result_submitted( + worker_result.found_at.map(|t| t.elapsed()), + ); + result_sent_at = Some(Instant::now()); } } } @@ -239,6 +265,10 @@ async fn handle_connection( match msg_result { Ok(MinerMessage::NewJob(request)) => { *authenticated = true; + metrics::record_job_received(); // lair + if let Some(t) = result_sent_at.take() { + metrics::record_job_idle(t.elapsed()); + } log::info!( "⛏️ Received job: id={}, hash=0x{}", request.job_id,