feat(telemetry): report process memory and process and host uptime
This commit is contained in:
16 files changed
+311
-8
No files matched your search
@@ -43,9 +43,23 @@ pub struct TelemetryGuard {
|
||||
traces: Option<SdkTracerProvider>,
|
||||
logs: Option<SdkLoggerProvider>,
|
||||
metrics: Option<SdkMeterProvider>,
|
||||
process: Option<process::ProcessMetrics>,
|
||||
}
|
||||
|
||||
impl TelemetryGuard {
|
||||
/// Attach platform measurements only when OTLP metrics are enabled. Core
|
||||
/// owns instruments; the supplied reader owns all OS-specific collection.
|
||||
pub fn with_process_metrics(mut self, source: impl ProcessMetricsSource) -> Self {
|
||||
if let Some(metrics) = &self.metrics {
|
||||
use opentelemetry::metrics::MeterProvider as _;
|
||||
self.process = Some(process::ProcessMetrics::new(
|
||||
&metrics.meter("wakey"),
|
||||
source,
|
||||
));
|
||||
}
|
||||
self
|
||||
}
|
||||
|
||||
pub fn shutdown(self) {
|
||||
// Export failures must not turn an otherwise successful daemon exit into a failure.
|
||||
if let Some(provider) = self.metrics {
|
||||
@@ -198,6 +212,7 @@ pub fn init(
|
||||
traces: Some(traces),
|
||||
logs: Some(logs),
|
||||
metrics: Some(metrics),
|
||||
process: None,
|
||||
};
|
||||
} else if config.json_logs {
|
||||
tracing_subscriber::registry()
|
||||
@@ -292,7 +307,9 @@ fn local_timer() -> fmt::time::OffsetTime<time::format_description::well_known::
|
||||
// therefore does no per-PTY-read allocations or instrument registration.
|
||||
static METRICS: OnceLock<Instruments> = OnceLock::new();
|
||||
mod fleet;
|
||||
mod process;
|
||||
pub use fleet::{AgentDeviceCounts, FleetCounts, FleetMetrics};
|
||||
pub use process::ProcessMetricsSource;
|
||||
struct Instruments {
|
||||
operations: opentelemetry::metrics::Counter<u64>,
|
||||
duration: opentelemetry::metrics::Histogram<f64>,
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
//! OS-independent process/host instruments. Callers supply a platform reader.
|
||||
//! Failed reads omit a sample: zero must not masquerade as a restart or freed RAM.
|
||||
|
||||
use opentelemetry::metrics::{Meter, ObservableGauge};
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Values are seconds/bytes, or None when unsupported/unavailable. Readers run
|
||||
/// on the metric collection thread: keep work small and local, never perform
|
||||
/// network requests or emit exporter-directed logs from these callbacks.
|
||||
pub trait ProcessMetricsSource: Send + Sync + 'static {
|
||||
fn process_uptime_seconds(&self) -> Option<f64>;
|
||||
fn resident_memory_bytes(&self) -> Option<u64>;
|
||||
fn host_uptime_seconds(&self) -> Option<f64>;
|
||||
}
|
||||
|
||||
pub(super) struct ProcessMetrics {
|
||||
_uptime: ObservableGauge<f64>,
|
||||
_resident: ObservableGauge<u64>,
|
||||
_host_uptime: ObservableGauge<f64>,
|
||||
}
|
||||
|
||||
impl ProcessMetrics {
|
||||
pub(super) fn new(meter: &Meter, source: impl ProcessMetricsSource) -> Self {
|
||||
let source = Arc::new(source);
|
||||
let uptime = source.clone();
|
||||
let resident = source.clone();
|
||||
Self {
|
||||
_uptime: meter
|
||||
.f64_observable_gauge("wakey.process.uptime")
|
||||
.with_unit("s")
|
||||
.with_description("Process age reported by the platform reader")
|
||||
.with_callback(move |observer| {
|
||||
if let Some(seconds) = uptime.process_uptime_seconds()
|
||||
&& seconds.is_finite()
|
||||
&& seconds >= 0.0
|
||||
{
|
||||
observer.observe(seconds, &[]);
|
||||
}
|
||||
})
|
||||
.build(),
|
||||
_resident: meter
|
||||
.u64_observable_gauge("wakey.process.memory.resident")
|
||||
.with_unit("By")
|
||||
.with_description("Resident process memory reported by the platform reader")
|
||||
.with_callback(move |observer| {
|
||||
if let Some(bytes) = resident.resident_memory_bytes() {
|
||||
observer.observe(bytes, &[]);
|
||||
}
|
||||
})
|
||||
.build(),
|
||||
_host_uptime: meter
|
||||
.f64_observable_gauge("wakey.host.uptime")
|
||||
.with_unit("s")
|
||||
.with_description("Host uptime reported by the platform reader")
|
||||
.with_callback(move |observer| {
|
||||
if let Some(seconds) = source.host_uptime_seconds()
|
||||
&& seconds.is_finite()
|
||||
&& seconds >= 0.0
|
||||
{
|
||||
observer.observe(seconds, &[]);
|
||||
}
|
||||
})
|
||||
.build(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -197,6 +197,20 @@ async fn exports_all_signals_and_preserves_remote_command_parent() {
|
||||
},
|
||||
);
|
||||
let server = tokio::spawn(server);
|
||||
// The exporter contract is OS-independent; Linux procfs collection is
|
||||
// tested in wakey-linux rather than imported into core.
|
||||
struct ProcessSample;
|
||||
impl telemetry::ProcessMetricsSource for ProcessSample {
|
||||
fn process_uptime_seconds(&self) -> Option<f64> {
|
||||
Some(120.5)
|
||||
}
|
||||
fn resident_memory_bytes(&self) -> Option<u64> {
|
||||
Some(32 * 1024 * 1024)
|
||||
}
|
||||
fn host_uptime_seconds(&self) -> Option<f64> {
|
||||
Some(10_000.0)
|
||||
}
|
||||
}
|
||||
let guard = telemetry::init(
|
||||
&TelemetryConfig {
|
||||
otlp_endpoint: Some(endpoint),
|
||||
@@ -211,7 +225,8 @@ async fn exports_all_signals_and_preserves_remote_command_parent() {
|
||||
"info",
|
||||
Some("test-agent"),
|
||||
)
|
||||
.unwrap();
|
||||
.unwrap()
|
||||
.with_process_metrics(ProcessSample);
|
||||
|
||||
let fleet_metrics = telemetry::FleetMetrics::new().unwrap();
|
||||
fleet_metrics.replace(Some(telemetry::FleetCounts {
|
||||
@@ -344,6 +359,44 @@ async fn exports_all_signals_and_preserves_remote_command_parent() {
|
||||
assert!(names.contains(&name), "missing {name}");
|
||||
}
|
||||
let latest = metrics.last().unwrap();
|
||||
{
|
||||
for (name, unit, expected) in [
|
||||
(
|
||||
"wakey.process.uptime",
|
||||
"s",
|
||||
number_data_point::Value::AsDouble(120.5),
|
||||
),
|
||||
(
|
||||
"wakey.process.memory.resident",
|
||||
"By",
|
||||
number_data_point::Value::AsInt(32 * 1024 * 1024),
|
||||
),
|
||||
(
|
||||
"wakey.host.uptime",
|
||||
"s",
|
||||
number_data_point::Value::AsDouble(10_000.0),
|
||||
),
|
||||
] {
|
||||
let metric = latest
|
||||
.resource_metrics
|
||||
.iter()
|
||||
.flat_map(|resource| &resource.scope_metrics)
|
||||
.flat_map(|scope| &scope.metrics)
|
||||
.find(|metric| metric.name == name)
|
||||
.unwrap_or_else(|| panic!("missing {name}"));
|
||||
assert_eq!(metric.unit, unit);
|
||||
let Some(metric::Data::Gauge(gauge)) = &metric.data else {
|
||||
panic!("{name} must be a gauge");
|
||||
};
|
||||
assert_eq!(gauge.data_points.len(), 1);
|
||||
let point = &gauge.data_points[0];
|
||||
assert!(
|
||||
point.attributes.is_empty(),
|
||||
"process metrics need no PID/host/address labels"
|
||||
);
|
||||
assert_eq!(point.value, Some(expected));
|
||||
}
|
||||
}
|
||||
for (name, label, expected) in [
|
||||
(
|
||||
"wakey.stored.devices",
|
||||
|
||||
Reference in new issue
Block a user