alloc-metrics: Record memory allocator metrics

As we want observability into the jemalloc allocator, this commit
introduces a new crate, `readyset-alloc-metrics`, that fetches the
current jemalloc state via `readyset-alloc`, and publishes those
as metrics (gauges) to our prometheus handler. As the values are
gauges, we need to sample the jemalloc stats periodically.

This new crate is enabled in both adapter and server.

Change-Id: If428829f3f3015ee8d40fdae4655161a8923ba88
Reviewed-on: https://gerrit.readyset.name/c/readyset/+/6246
Tested-by: Buildkite CI
Reviewed-by: Luke Osborne <luke@readyset.io>

Cargo.toml

@@ -30,6 +30,7 @@ members = [
     "readyset",
     "readyset-adapter",
     "readyset-alloc",
+    "readyset-alloc-metrics",
     "readyset-client",
     "readyset-client-metrics",
     "readyset-client-test-helpers",

readyset-alloc-metrics/Cargo.toml

@@ -0,0 +1,17 @@
+[package]
+name = "readyset-alloc-metrics"
+version = "0.1.0"
+publish = false
+authors = ["ReadySet Technology, Inc. <info@readyset.io>"]
+edition = "2021"
+
+
+[dependencies]
+metrics = { workspace = true }
+metrics-exporter-prometheus = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+tracing = { version = "0.1", features = ["release_max_level_debug"] }
+
+# Local dependencies
+readyset-alloc = { path = "../readyset-alloc" }
+readyset-util = { path = "../readyset-util" }

readyset-alloc-metrics/src/lib.rs

@@ -0,0 +1,73 @@
+use std::time::Duration;
+
+use metrics::Gauge;
+use readyset_alloc::fetch_stats;
+use readyset_util::shutdown::ShutdownReceiver;
+use tokio::select;
+use tracing::info;
+
+const REPORTING_INTERVAL: Duration = Duration::from_secs(2);
+
+const ALLOCATED_BYTES: &str = "readyset_allocator_allocated_bytes";
+const ACTIVE_BYTES: &str = "readyset_allocator_active_bytes";
+const RETAINED_BYTES: &str = "readyset_allocator_retained_bytes";
+const MAPPED_BYTES: &str = "readyset_allocator_mapped_bytes";
+const DIRTY_BYTES: &str = "readyset_allocator_dirty_bytes";
+const FRAGMENTED_BYTES: &str = "readyset_allocator_fragmented_bytes";
+
+pub async fn report_allocator_metrics(mut shutdown_rx: ShutdownReceiver) {
+    let mut interval = tokio::time::interval(REPORTING_INTERVAL);
+    let mut reporter = AllocatorMetricsReporter::new();
+
+    loop {
+        select! {
+            _ = interval.tick() => reporter.report_metrics(),
+            _ = shutdown_rx.recv() => break,
+        }
+    }
+}
+
+struct AllocatorMetricsReporter {
+    allocated: Gauge,
+    active: Gauge,
+    retained: Gauge,
+    mapped: Gauge,
+    dirty: Gauge,
+    fragmented: Gauge,
+}
+
+impl AllocatorMetricsReporter {
+    fn new() -> Self {
+        Self {
+            allocated: metrics::register_gauge!(ALLOCATED_BYTES),
+            active: metrics::register_gauge!(ACTIVE_BYTES),
+            retained: metrics::register_gauge!(RETAINED_BYTES),
+            mapped: metrics::register_gauge!(MAPPED_BYTES),
+            dirty: metrics::register_gauge!(DIRTY_BYTES),
+            fragmented: metrics::register_gauge!(FRAGMENTED_BYTES),
+        }
+    }
+
+    fn report_metrics(&mut self) {
+        // Note: we could call alloc::jemalloc::iterate_thread_allocation_stats() to get the
+        // per-thread dump. `fetch_stats` is sufficient for now.
+        // Additional note: iterate_thread_allocation_stats() doesn't automatically bump the
+        // it's epoch, so you'll need to do that (just read the code if you are interested).
+
+        match fetch_stats() {
+            Ok(stats) => {
+                info!("NEXT alloc stats: {:?}", stats);
+                self.allocated.set(stats.allocated as f64);
+                self.active.set(stats.active as f64);
+                self.retained.set(stats.retained as f64);
+                self.mapped.set(stats.mapped as f64);
+                self.dirty.set(stats.dirty as f64);
+                self.fragmented.set(stats.fragmentation as f64);
+            }
+            Err(e) => {
+                // not sure what else to do but log, at a low level :shrug:
+                info!("Failed to fetch memory allocator stats: {:?}", e);
+            }
+        }
+    }
+}

readyset-alloc/src/lib.rs

@@ -75,38 +75,38 @@ pub mod trace;
 /// perspective of jemalloc(3)
 pub struct AllocStats {
     /// Total number of bytes allocated by the application.
-    allocated: usize,
+    pub allocated: usize,
     /// Total number of bytes in active pages allocated by the application. This is a multiple of
     /// the page size, and greater than or equal to "stats.allocated". This does not include
     /// (jemalloc(3)) "stats.arenas.<i>.pdirty" and pages entirely devoted to allocator metadata.
-    active: usize,
+    pub active: usize,
     /// Total number of bytes dedicated to metadata, which comprise base allocations used for
     /// bootstrap-sensitive allocator metadata structures (see jemalloc(3) stats.arenas.<i>.base)
     /// and internal allocations (see stats.arenas.<i>.internal). Transparent huge page
     /// (enabled with opt.metadata_thp) usage is not considered.
-    metadata: usize,
+    pub metadata: usize,
     /// Maximum number of bytes in physically resident data pages mapped by the allocator,
     /// comprising all pages dedicated to allocator metadata, pages backing active allocations, and
     /// unused dirty pages. This is a maximum rather than precise because pages may not actually be
     /// physically resident if they correspond to demand-zeroed virtual memory that has not yet
     /// been touched. This is a multiple of the page size, and is larger than stats.active.
-    resident: usize,
+    pub resident: usize,
     /// Total number of bytes in chunks mapped on behalf of the application. This is a multiple of
     /// the chunk size, and is at least as large as "stats.active". This does not include inactive
     /// chunks.
-    mapped: usize,
+    pub mapped: usize,
     /// Total number of bytes in virtual memory mappings that were retained rather than being
     /// returned to the operating system via e.g. munmap(2) or similar. Retained virtual memory is
     /// typically untouched, decommitted, or purged, so it has no strongly associated physical
     /// memory (see extent hooks for details). Retained memory is excluded from mapped memory
     /// statistics, e.g. stats.mapped.
-    retained: usize,
+    pub retained: usize,
     /// Total number of bytes that are resident but not "active" or "metadata". These bytes were
     /// once used by the process but have not been reclaimed by the OS.
-    dirty: usize,
+    pub dirty: usize,
     /// Total number of bytes that are in active pages but are not "allocated" by the
     /// process--meaning they exist as the result of memory fragmentation.
-    fragmentation: usize,
+    pub fragmentation: usize,
 }
 
 #[path = "jemalloc.rs"]

readyset-server/Cargo.toml

@@ -82,6 +82,7 @@ dataflow = { path = "../readyset-dataflow", package = "readyset-dataflow" }
 mir = { path = "../readyset-mir", package = "readyset-mir" }
 common = { path = "../readyset-common", package = "readyset-common" }
 readyset-alloc = { path = "../readyset-alloc" }
+readyset-alloc-metrics = { path = "../readyset-alloc-metrics" }
 readyset-client = { path = "../readyset-client" }
 failpoint-macros = { path = "../failpoint-macros" }
 readyset-errors = { path = "../readyset-errors" }

readyset-server/src/startup.rs

@@ -56,6 +56,7 @@ use dataflow::Readers;
 use failpoint_macros::set_failpoint;
 use futures_util::future::{Either, TryFutureExt};
 use health_reporter::{HealthReporter, State as ServerState};
+use readyset_alloc_metrics::report_allocator_metrics;
 use readyset_client::consensus::{Authority, WorkerSchedulingConfig};
 use readyset_client::{ControllerDescriptor, WorkerDescriptor};
 use readyset_telemetry_reporter::{TelemetryBuilder, TelemetryEvent, TelemetrySender};
@@ -272,6 +273,9 @@ pub(crate) async fn start_instance_inner(
         ..
     } = config;
 
+    let alloc_shutdown = shutdown_rx.clone();
+    tokio::spawn(report_allocator_metrics(alloc_shutdown));
+
     let (tx, rx) = maybe_create_failpoint_chann(wait_for_failpoint);
     let mut health_reporter = HealthReporter::new();
     let http_uri = start_request_router(

readyset/Cargo.toml

@@ -34,6 +34,7 @@ psql-srv = { path = "../psql-srv" }
 database-utils = { path = "../database-utils" }
 readyset-adapter = { path = "../readyset-adapter" }
 readyset-alloc = { path = "../readyset-alloc" }
+readyset-alloc-metrics = { path = "../readyset-alloc-metrics" }
 readyset-client = { path = "../readyset-client/" }
 readyset-client-metrics = { path = "../readyset-client-metrics" }
 readyset-common = { path = "../readyset-common" }

readyset/src/lib.rs

@@ -38,6 +38,7 @@ use readyset_adapter::query_status_cache::{MigrationStyle, QueryStatusCache};
 use readyset_adapter::views_synchronizer::ViewsSynchronizer;
 use readyset_adapter::{Backend, BackendBuilder, QueryHandler, UpstreamDatabase};
 use readyset_alloc::{StdThreadBuildWrapper, ThreadBuildWrapper};
+use readyset_alloc_metrics::report_allocator_metrics;
 use readyset_client::consensus::AuthorityType;
 #[cfg(feature = "failure_injection")]
 use readyset_client::failpoints;
@@ -730,6 +731,13 @@ where
 
         let (shutdown_tx, shutdown_rx) = shutdown::channel();
 
+        // if we're running in standalone mode, server will already
+        // spawn it's own allocator metrics reporter.
+        if prometheus_handle.is_some() && !options.standalone {
+            let alloc_shutdown = shutdown_rx.clone();
+            rt.handle().spawn(report_allocator_metrics(alloc_shutdown));
+        }
+
         // Gate query log code path on the log flag existing.
         let qlog_sender = if options.query_log_mode.is_enabled() {
             rs_connect.in_scope(|| info!("Query logs are enabled. Spawning query logger"));