From 5c6bbd6f9686dd9e3e86da7d62137ee487d166b4 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Thu, 4 Dec 2025 23:35:04 -0700 Subject: [PATCH] rename "plotter" as "prometheus", make plot feature optional in signal-gateway --- Cargo.toml | 2 +- signal-gateway/Cargo.toml | 13 +- signal-gateway/src/gateway/circular_buffer.rs | 2 + signal-gateway/src/gateway/mod.rs | 93 +++--- signal-gateway/src/main.rs | 2 +- .../src/{plotter.rs => prometheus/mod.rs} | 296 ++++++------------ signal-gateway/src/prometheus/plot.rs | 141 +++++++++ 7 files changed, 311 insertions(+), 238 deletions(-) rename signal-gateway/src/{plotter.rs => prometheus/mod.rs} (51%) create mode 100644 signal-gateway/src/prometheus/plot.rs diff --git a/Cargo.toml b/Cargo.toml index a2765db..c93b02d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,7 +21,7 @@ assigning_clones = "allow" result_large_err = "allow" [workspace.dependencies] -prometheus-http-client = { path = "prometheus-http-client" } +prometheus-http-client = { path = "prometheus-http-client", default-features = false } async-trait = "0.1" chrono = { version = "0.4", default-features = false, features = ["clock", "serde", "std"] } diff --git a/signal-gateway/Cargo.toml b/signal-gateway/Cargo.toml index ef72b67..1d93562 100644 --- a/signal-gateway/Cargo.toml +++ b/signal-gateway/Cargo.toml @@ -6,11 +6,14 @@ edition.workspace = true [lints] workspace = true +[features] +default = ["plot"] +plot = ["prometheus-http-client/plot", "dep:chrono-tz", "dep:rand", "dep:walkdir"] + [dependencies] -prometheus-http-client = { workspace = true } +prometheus-http-client = { workspace = true, default-features = false } chrono = { workspace = true } -chrono-tz = { workspace = true } conf = { workspace = true } conf-extra = { workspace = true } displaydoc = { workspace = true } @@ -21,7 +24,6 @@ humantime = { workspace = true } hyper = { workspace = true } hyper-util = { workspace = true } jsonrpsee = { workspace = true } -rand = { workspace = true } reqwest = { workspace = true } rustls = { workspace = true } serde = { workspace = true } @@ -33,4 +35,7 @@ tokio-util = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } url = { workspace = true } -walkdir = { workspace = true } + +chrono-tz = { workspace = true, optional = true } +rand = { workspace = true, optional = true } +walkdir = { workspace = true, optional = true } diff --git a/signal-gateway/src/gateway/circular_buffer.rs b/signal-gateway/src/gateway/circular_buffer.rs index f6620d0..c405787 100644 --- a/signal-gateway/src/gateway/circular_buffer.rs +++ b/signal-gateway/src/gateway/circular_buffer.rs @@ -44,11 +44,13 @@ impl CircularBuffer { } /// Returns true if the buffer is empty. + #[allow(unused)] pub fn is_empty(&self) -> bool { self.buf.is_empty() } /// Returns the capacity of the buffer. + #[allow(unused)] pub fn capacity(&self) -> usize { self.capacity } diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 7ac20d0..69aa24f 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -1,7 +1,7 @@ use crate::{ http::{AlertMessage, Status}, jsonrpc::{Envelope, RpcClient, RpcClientError, SignalMessage, connect_tcp}, - plotter::{Plotter, PlotterConfig}, + prometheus::{Prometheus, PrometheusConfig}, }; use conf::{Conf, Subcommands}; use futures_util::FutureExt; @@ -10,7 +10,9 @@ use hyper::{Method, Request, Response, StatusCode, body::Incoming}; use prometheus_http_client::{AlertStatus, ExtractLabels}; //use jsonrpsee::async_client::{Client as JsonRpcClient, Error as JsonRpcError}; use chrono::Utc; -use std::{collections::HashMap, error::Error, fmt::Write, net::SocketAddr, time::Duration}; +use std::{ + collections::HashMap, error::Error, fmt::Write, net::SocketAddr, path::PathBuf, time::Duration, +}; use syslog_rfc5424::SyslogMessage; use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, @@ -45,7 +47,7 @@ pub struct GatewayConfig { #[conf(repeat, long, env)] pub admin_uuid: Vec, #[conf(flatten)] - pub plotter: Option, + pub prometheus: Option, #[conf(flatten)] pub log_handler: LogHandlerConfig, } @@ -58,6 +60,7 @@ struct GatewayCommandWrapper { } /// Commands that can be sent to the gateway (prefixed with /) +#[allow(unused)] #[derive(Clone, Debug, Subcommands)] enum GatewayCommand { /// Show recent log messages @@ -171,7 +174,7 @@ struct AdminMessage { /// The origin of the message (app + host), if from syslog origin: Option, text: String, - attachment_paths: Vec, + attachment_paths: Vec, /// Short summary for logging (e.g., alert names for prometheus). /// If None, the consumer will use a truncated slice of `text` for logging. summary: Option, @@ -193,7 +196,7 @@ pub struct Gateway { admin_mq_tx: UnboundedSender, admin_mq_rx: Mutex>, token: CancellationToken, - plotter: Option, + prometheus: Option, /// Log handlers keyed by origin (app + host). Lazily created when first message from an origin arrives. log_handlers: RwLock>, } @@ -202,19 +205,19 @@ impl Gateway { pub async fn new(config: GatewayConfig, token: CancellationToken) -> Self { let (admin_mq_tx, admin_mq_rx) = unbounded_channel(); - let plotter = config - .plotter + let prometheus = config + .prometheus .as_ref() - .map(|plotter_config| Plotter::new(plotter_config.clone())) + .map(|pc| Prometheus::new(pc.clone())) .transpose() - .expect("Invalid plotter config"); + .expect("Invalid prometheus config"); Self { config, admin_mq_tx, admin_mq_rx: Mutex::new(admin_mq_rx), token, - plotter, + prometheus, log_handlers: RwLock::new(HashMap::new()), } } @@ -273,11 +276,12 @@ impl Gateway { } else { msg.text }; + let attachments = msg.attachment_paths.into_iter().map(|p| p.to_str().unwrap().to_owned()).collect(); SignalMessage { sender: self.config.signal_account.clone(), recipient: self.config.admin_uuid.clone(), message, - attachments: msg.attachment_paths, + attachments, }.send(signal_cli).await?; } else { warn!("admin_mq_rx is closed, halting service"); @@ -307,12 +311,15 @@ impl Gateway { continue; } - let (resp, _) = join!(self.handle_signal_admin_message(&msg.envelope), - msg.envelope.send_read_receipt(signal_cli, &self.config.signal_account).map(|result| { - if let Err(err) = result { - warn!("Couldn't send read receipt: {err}"); - } - })); + let (resp, _) = join!( + self.handle_signal_admin_message(&msg.envelope), + msg.envelope.send_read_receipt(signal_cli, &self.config.signal_account).map(|result| { + if let Err(err) = result { + warn!("Couldn't send read receipt: {err}"); + } + }) + ); + let (message, attachments) = resp.unwrap_or_else( |(code, msg)| { let text = format!("{code}: {msg}"); @@ -321,6 +328,8 @@ impl Gateway { } ); + let attachments = attachments.into_iter().map(|p| p.to_str().expect("attachments must have utf8 paths").to_owned()).collect(); + SignalMessage { sender: self.config.signal_account.clone(), recipient: vec![msg.envelope.source_uuid.clone()], @@ -339,7 +348,7 @@ impl Gateway { async fn handle_signal_admin_message( &self, msg: &Envelope, - ) -> Result<(String, Vec), (u16, Box)> { + ) -> Result<(String, Vec), (u16, Box)> { let data = msg.data_message.as_ref().unwrap(); // Admin messages starting with / are handled by gateway @@ -432,7 +441,7 @@ impl Gateway { async fn handle_gateway_command( &self, cmd: GatewayCommand, - ) -> Result<(String, Vec), (u16, Box)> { + ) -> Result<(String, Vec), (u16, Box)> { match cmd { GatewayCommand::Log { filter } => { let handlers = self.log_handlers.read().await; @@ -457,12 +466,12 @@ impl Gateway { Ok((text, vec![])) } GatewayCommand::Query { query } => { - let plotter = self - .plotter + let prometheus = self + .prometheus .as_ref() .ok_or_else(|| (500, "prometheus was not configured".into()))?; - match plotter.oneoff_query(query).await { + match prometheus.oneoff_query(query).await { Ok(( ExtractLabels { name, @@ -491,26 +500,31 @@ impl Gateway { Err(err) => Err((500, err)), } } + #[cfg(feature = "plot")] GatewayCommand::Plot { query, duration } => { - let plotter = self - .plotter + let prometheus = self + .prometheus .as_ref() .ok_or_else(|| (500, "prometheus was not configured".into()))?; - plotter.purge_old_plots(); - match plotter.create_oneoff_plot(query.clone(), duration).await { + prometheus.purge_old_plots(); + match prometheus.create_oneoff_plot(query.clone(), duration).await { Ok(filename) => Ok((query, vec![filename])), Err(err) => Err((500, err)), } } + #[cfg(not(feature = "plot"))] + GatewayCommand::Plot { .. } => { + Err((500, "the plot feature was not enabled at build time".into())) + } GatewayCommand::Series { matchers } => { - let plotter = self - .plotter + let prometheus = self + .prometheus .as_ref() .ok_or_else(|| (500, "prometheus was not configured".into()))?; let matcher_refs: Vec<&str> = matchers.iter().map(|s| s.as_str()).collect(); - match plotter.series(&matcher_refs).await { + match prometheus.series(&matcher_refs).await { Ok(data) => { let mut text = data.iter().fold(String::default(), |mut buf, kv| { writeln!(&mut buf, "{kv:?}").unwrap(); @@ -525,13 +539,13 @@ impl Gateway { } } GatewayCommand::Labels { matchers } => { - let plotter = self - .plotter + let prometheus = self + .prometheus .as_ref() .ok_or_else(|| (500, "prometheus was not configured".into()))?; let matcher_refs: Vec<&str> = matchers.iter().map(|s| s.as_str()).collect(); - match plotter.labels(&matcher_refs).await { + match prometheus.labels(&matcher_refs).await { Ok(data) => { let mut text = data.iter().fold(String::default(), |mut buf, l| { writeln!(&mut buf, "{l}").unwrap(); @@ -546,12 +560,12 @@ impl Gateway { } } GatewayCommand::Alerts => { - let plotter = self - .plotter + let prometheus = self + .prometheus .as_ref() .ok_or_else(|| (500, "prometheus was not configured".into()))?; - match plotter.alerts().await { + match prometheus.alerts().await { Ok(data) => { let now = Utc::now(); let mut text = data.iter().fold(String::default(), |mut buf, alert| { @@ -603,11 +617,14 @@ impl Gateway { .format_alert_text(&alert_msg) .unwrap_or_else(|err| format!("error formatting alert text: {err}:\n{alert_msg:#?}")); + #[allow(unused_mut)] let mut attachment_paths = vec![]; - if let Some(plotter) = self.plotter.as_ref() { - plotter.purge_old_plots(); + + #[cfg(feature = "plot")] + if let Some(prometheus) = self.prometheus.as_ref() { + prometheus.purge_old_plots(); for alert in alert_msg.alerts.iter() { - match plotter.create_alert_plot(alert).await { + match prometheus.create_alert_plot(alert).await { Ok(path) => { attachment_paths.push(path); } diff --git a/signal-gateway/src/main.rs b/signal-gateway/src/main.rs index 56ef26b..1c55c96 100644 --- a/signal-gateway/src/main.rs +++ b/signal-gateway/src/main.rs @@ -14,7 +14,7 @@ pub mod http; mod human_duration; mod init_logging; pub mod jsonrpc; -pub mod plotter; +pub mod prometheus; pub mod transports; use config::Config; diff --git a/signal-gateway/src/plotter.rs b/signal-gateway/src/prometheus/mod.rs similarity index 51% rename from signal-gateway/src/plotter.rs rename to signal-gateway/src/prometheus/mod.rs index ad530ea..565acda 100644 --- a/signal-gateway/src/plotter.rs +++ b/signal-gateway/src/prometheus/mod.rs @@ -1,217 +1,54 @@ -use crate::http::Alert; -use chrono::{FixedOffset, Local, Offset, Utc}; -use chrono_tz::Tz; use conf::Conf; use prometheus_http_client::{ - AlertInfo, AlertsRequest, ExtractLabels, Labels, LabelsRequest, MetricTimeseries, MetricVal, - MetricValue, PromRequest, QueryRangeRequest, QueryRequest, SeriesRequest, - plot::{PlotStyle, PlotThreshold}, + AlertInfo, AlertsRequest, ExtractLabels, Labels, LabelsRequest, MetricVal, MetricValue, + PromRequest, QueryRequest, SeriesRequest, }; -use rand::RngCore; use reqwest::Client as ReqwestClient; -use std::{error::Error, str::FromStr, time::Duration}; -use tracing::{info, warn}; -use walkdir::WalkDir; +use std::error::Error; +use tracing::info; +#[cfg(feature = "plot")] +mod plot; +#[cfg(feature = "plot")] +use plot::{PlotConfig, PlotThreshold}; + +#[cfg(feature = "plot")] +use crate::http::Alert; +#[cfg(feature = "plot")] +use prometheus_http_client::QueryRangeRequest; +#[cfg(feature = "plot")] +use std::{path::PathBuf, str::FromStr, time::Duration}; + +/// Configures our prometheus API client #[derive(Clone, Conf, Debug)] -#[conf(at_most_one_of_fields(utc_offset, timezone))] -pub struct PlotterConfig { +pub struct PrometheusConfig { + /// Address of prometheus host. Should start with http and usually indicate port 9090 #[conf(long, env)] pub prometheus_host: String, - #[conf(long, env, default_value = "30m", value_parser = conf_extra::parse_duration)] - pub plot_age_limit: Duration, - /// Fixed timezone offset for plot timestamps, e.g. "+05:30" or "-08:00". - /// Mutually exclusive with --timezone. - #[conf(long, env)] - pub utc_offset: Option, - /// Timezone name for plot timestamps, e.g. "US/Mountain" or "Europe/Paris". - /// Mutually exclusive with --utc-offset. - #[conf(long, env)] - pub timezone: Option, - /// Stroke width for plot lines - #[conf(long, env, default_value = "2")] - pub line_width: u32, - /// Plot dimensions in pixels (width, height), e.g. "1920,1200" or "(1920, 1200)" - #[conf(long, env, default_value = "1920,1200", value_parser = parse_dimensions)] - pub plot_dimensions: (u32, u32), + #[cfg(feature = "plot")] + /// Configuration options for generated plots + #[conf(flatten, prefix)] + pub plot: PlotConfig, } -fn parse_dimensions(s: &str) -> Result<(u32, u32), String> { - let s = s.trim(); - // Strip optional parentheses - let s = s - .strip_prefix('(') - .and_then(|s| s.strip_suffix(')')) - .unwrap_or(s); - - let (left, right) = s - .split_once(',') - .ok_or_else(|| format!("expected 'width,height', got '{s}'"))?; - - let width = left - .trim() - .parse::() - .map_err(|e| format!("invalid width: {e}"))?; - let height = right - .trim() - .parse::() - .map_err(|e| format!("invalid height: {e}"))?; - - Ok((width, height)) -} - -pub struct Plotter { - config: PlotterConfig, - plot_dir: String, +/// Prometheus HTTP API + plotting if configured +pub struct Prometheus { + config: PrometheusConfig, reqwest_client: ReqwestClient, - skip_labels: Vec, } -impl Plotter { - pub fn new(config: PlotterConfig) -> Result { - if config.utc_offset.is_some() && config.timezone.is_some() { - return Err("Cannot specify both --utc-offset and --timezone"); - } - - let plot_dir = "/tmp".into(); +impl Prometheus { + /// Create a new Prometheus client object + pub fn new(config: PrometheusConfig) -> Result { let reqwest_client = ReqwestClient::new(); - let skip_labels = vec!["job".into(), "instance".into()]; Ok(Self { config, - plot_dir, reqwest_client, - skip_labels, }) } - fn create_plot( - &self, - matrix: &[MetricTimeseries], - threshold: Option, - title: Option<&str>, - ) -> Result> { - let filename = format!( - "{dir}/plot-{num}.png", - dir = self.plot_dir, - num = rand::rng().next_u64() - ); - let utc_offset = if let Some(offset) = self.config.utc_offset { - offset - } else if let Some(tz) = self.config.timezone { - Utc::now().with_timezone(&tz).offset().fix() - } else { - *Local::now().offset() - }; - let mut plot_style = PlotStyle::default() - .dark_mode() - .with_line_width(self.config.line_width) - .with_drawing_area(self.config.plot_dimensions) - .with_utc_offset(utc_offset) - .with_skip_labels(self.skip_labels.clone()); - if let Some(title) = title { - plot_style = plot_style.with_title(title); - } - - plot_style.plot_timeseries(&filename, matrix, threshold)?; - Ok(filename) - } - - pub fn purge_old_plots(&self) { - for entry in WalkDir::new(&self.plot_dir) - .min_depth(1) - .max_depth(1) - .into_iter() - .filter_map(|e| e.ok()) - { - if !entry.file_type().is_file() { - continue; - } - let Some(ext) = entry.path().extension() else { - continue; - }; - if ext != "png" && ext != ".png" { - continue; - } - let path = entry.path().display(); - let Ok(metadata) = entry - .metadata() - .inspect_err(|err| warn!("Couldn't get metadata for {path}: {err}")) - else { - continue; - }; - let Ok(time) = metadata - .created() - .inspect_err(|err| warn!("Couldn't get creation time for {path}: {err}")) - else { - continue; - }; - let Ok(elapsed) = time - .elapsed() - .inspect_err(|err| warn!("Elapsed time calculation failed for {path}: {err}")) - else { - continue; - }; - if elapsed > self.config.plot_age_limit - && let Err(err) = std::fs::remove_file(entry.path()) - { - warn!("Couldn't remove old png file {path}: {err}"); - } - } - } - - pub async fn create_alert_plot(&self, alert: &Alert) -> Result> { - let expr = alert.parse_expr_from_generator_url()?; - - // Parse expressions like "query < 0.09" or "query < 0.09 and on (instance) up{...}" - // We look for comparison operators and extract the threshold - let (base_query, threshold) = parse_alert_expr(&expr)?; - - // Build label selector from alert labels (excluding job/instance) - let label_selector = build_label_selector(&alert.labels, &self.skip_labels); - let query = if label_selector.is_empty() { - base_query.to_owned() - } else { - format!("{base_query}{{{label_selector}}}") - }; - - let now = Utc::now(); - let elapsed = now - alert.starts_at; - // Extend elapsed by 210%, but use at least plot_age_limit (default 30m) - let lengthen = elapsed - .checked_mul(31) - .and_then(|e| e.checked_div(10)) - .ok_or("timedelta overflow")?; - let min_range = chrono::TimeDelta::from_std(self.config.plot_age_limit)?; - let range = lengthen.max(min_range); - - info!("Prom range query: {query}"); - let matrix = QueryRangeRequest::builder(query.clone()) - .range(now - range..now) - .build() - .send_with_client(&self.reqwest_client, &self.config.prometheus_host) - .await? - .into_matrix()?; - - self.create_plot(&matrix, Some(threshold), Some(&query)) - } - - pub async fn create_oneoff_plot( - &self, - query: String, - since: Duration, - ) -> Result> { - info!("Prom range query: {query}"); - let matrix = QueryRangeRequest::builder(query.to_owned()) - .since(since) - .build() - .send_with_client(&self.reqwest_client, &self.config.prometheus_host) - .await? - .into_matrix()?; - - self.create_plot(&matrix, None, Some(&query)) - } - + /// Evaluate a PromQL query, returning a list of matching current timeseries values and their labels. #[allow(clippy::type_complexity)] pub async fn oneoff_query( &self, @@ -223,12 +60,13 @@ impl Plotter { .await? .into_vector()?; - let labels = ExtractLabels::new(vector.iter().map(|mv| &mv.metric), &self.skip_labels); + let labels = ExtractLabels::new(vector.iter().map(|mv| &mv.metric), &[]); let values = vector.into_iter().map(|mv| mv.value).collect(); Ok((labels, values)) } + /// Get the list of all timeseries, and all of their labels, matching given filters. pub async fn series( &self, matches: impl IntoIterator>, @@ -240,6 +78,7 @@ impl Plotter { .await?) } + /// Get the list of all existing labels, matching given filters pub async fn labels( &self, matches: impl IntoIterator>, @@ -251,6 +90,7 @@ impl Plotter { .await?) } + /// Get the list of current alerts pub async fn alerts(&self) -> Result, Box> { Ok(AlertsRequest {} .send_with_client(&self.reqwest_client, &self.config.prometheus_host) @@ -259,8 +99,75 @@ impl Plotter { } } +#[cfg(feature = "plot")] +impl Prometheus { + /// Purge old plots from the plot directory + pub fn purge_old_plots(&self) { + self.config.plot.purge_old_plots(); + } + + /// Create a new plot corresponding to a given alert. Returns a pathbuf if it is present + pub async fn create_alert_plot(&self, alert: &Alert) -> Result> { + use chrono::{TimeDelta, Utc}; + + let expr = alert.parse_expr_from_generator_url()?; + + // Parse expressions like "query < 0.09" or "query < 0.09 and on (instance) up{...}" + // We look for comparison operators and extract the threshold + let (base_query, threshold) = parse_alert_expr(&expr)?; + + // Build label selector from alert labels (excluding job/instance) + let label_selector = build_label_selector(&alert.labels, &[]); + let query = if label_selector.is_empty() { + base_query.to_owned() + } else { + format!("{base_query}{{{label_selector}}}") + }; + + let now = Utc::now(); + let elapsed = now - alert.starts_at; + // Extend elapsed by 110%, but use at least 30m + let lengthen = elapsed + .checked_mul(21) + .and_then(|e| e.checked_div(10)) + .ok_or("timedelta overflow")?; + let min_range = TimeDelta::minutes(30); + let range = lengthen.max(min_range); + + info!("Prom range query: {query}"); + let matrix = QueryRangeRequest::builder(query.clone()) + .range(now - range..now) + .build() + .send_with_client(&self.reqwest_client, &self.config.prometheus_host) + .await? + .into_matrix()?; + + self.config + .plot + .create_plot(&matrix, Some(threshold), Some(&query)) + } + + /// Create a plot for a given query and timeframe + pub async fn create_oneoff_plot( + &self, + query: String, + since: Duration, + ) -> Result> { + info!("Prom range query: {query}"); + let matrix = QueryRangeRequest::builder(query.to_owned()) + .since(since) + .build() + .send_with_client(&self.reqwest_client, &self.config.prometheus_host) + .await? + .into_matrix()?; + + self.config.plot.create_plot(&matrix, None, Some(&query)) + } +} + /// Build a prometheus label selector string from alert labels, excluding specified labels. /// E.g., {"asset": "ETH", "job": "ec2"} with skip=["job"] -> `asset="ETH"` +#[cfg(feature = "plot")] fn build_label_selector( labels: &std::collections::BTreeMap, skip: &[String], @@ -281,6 +188,7 @@ fn build_label_selector( /// - "rate(tick_successes[5m]) < 0.09 and on (instance) up{job=\"ec2\"}" /// /// Returns (base_query, threshold) where base_query is the part before the comparator. +#[cfg(feature = "plot")] fn parse_alert_expr(expr: &str) -> Result<(String, PlotThreshold), Box> { // Look for comparison operators with surrounding spaces for comparator in [" < ", " > "] { diff --git a/signal-gateway/src/prometheus/plot.rs b/signal-gateway/src/prometheus/plot.rs new file mode 100644 index 0000000..1da04f3 --- /dev/null +++ b/signal-gateway/src/prometheus/plot.rs @@ -0,0 +1,141 @@ +use chrono::{FixedOffset, Local, Offset, Utc}; +use chrono_tz::Tz; +use conf::Conf; +use rand::RngCore; +use std::{error::Error, path::PathBuf, time::Duration}; +use tracing::{error, warn}; +use walkdir::WalkDir; + +pub use prometheus_http_client::{ + MetricTimeseries, + plot::{PlotStyle, PlotThreshold}, +}; + +/// Configuration for plots generated from prometheus data +#[derive(Clone, Conf, Debug)] +#[conf(at_most_one_of_fields(utc_offset, timezone))] +pub struct PlotConfig { + /// Working directory for temporary plot files + #[conf(long, env, default_value = "/tmp")] + dir: PathBuf, + /// How long before temporary plot files are removed from the working directory + #[conf(long, env, default_value = "30m", value_parser = conf_extra::parse_duration)] + age_limit: Duration, + /// Fixed timezone offset for plot timestamps, e.g. "+05:30" or "-08:00". + /// Mutually exclusive with --timezone. + #[conf(long, env)] + utc_offset: Option, + /// Timezone name for plot timestamps, e.g. "US/Mountain" or "Europe/Paris". + /// Mutually exclusive with --utc-offset. + #[conf(long, env)] + timezone: Option, + /// Stroke width for plot lines + #[conf(long, env, default_value = "2")] + line_width: u32, + /// Plot dimensions in pixels (width, height), e.g. "1920,1200" or "(1920, 1200)" + #[conf(long, env, default_value = "1920,1200", value_parser = parse_dimensions)] + dimensions: (u32, u32), + /// Don't show these labels in the plot title or legend + #[conf(long, env, default_value = "[\"job\", \"instance\"]", value_parser = serde_json::from_str)] + skip_labels: Vec, +} + +impl PlotConfig { + pub fn create_plot( + &self, + matrix: &[MetricTimeseries], + threshold: Option, + title: Option<&str>, + ) -> Result> { + let mut filename = self.dir.clone(); + filename.push(format!("plot-{}", rand::rng().next_u64())); + filename.set_extension("png"); + + let utc_offset = if let Some(offset) = self.utc_offset { + offset + } else if let Some(tz) = self.timezone { + Utc::now().with_timezone(&tz).offset().fix() + } else { + *Local::now().offset() + }; + let mut plot_style = PlotStyle::default() + .dark_mode() + .with_line_width(self.line_width) + .with_drawing_area(self.dimensions) + .with_utc_offset(utc_offset) + .with_skip_labels(self.skip_labels.clone()); + if let Some(title) = title { + plot_style = plot_style.with_title(title); + } + + plot_style.plot_timeseries(&filename, matrix, threshold)?; + Ok(filename) + } + + pub fn purge_old_plots(&self) { + for entry in WalkDir::new(&self.dir) + .min_depth(1) + .max_depth(1) + .into_iter() + .filter_map(|e| e.ok()) + { + if !entry.file_type().is_file() { + continue; + } + let Some(ext) = entry.path().extension() else { + continue; + }; + if ext != "png" && ext != ".png" { + continue; + } + let path = entry.path().display(); + let Ok(metadata) = entry + .metadata() + .inspect_err(|err| warn!("Couldn't get metadata for {path}: {err}")) + else { + continue; + }; + let Ok(time) = metadata + .created() + .inspect_err(|err| warn!("Couldn't get creation time for {path}: {err}")) + else { + continue; + }; + let Ok(elapsed) = time + .elapsed() + .inspect_err(|err| warn!("Elapsed time calculation failed for {path}: {err}")) + else { + continue; + }; + if elapsed > self.age_limit + && let Err(err) = std::fs::remove_file(entry.path()) + { + error!("Couldn't remove old png file {path}: {err}"); + } + } + } +} + +fn parse_dimensions(s: &str) -> Result<(u32, u32), String> { + let s = s.trim(); + // Strip optional parentheses + let s = s + .strip_prefix('(') + .and_then(|s| s.strip_suffix(')')) + .unwrap_or(s); + + let (left, right) = s + .split_once(',') + .ok_or_else(|| format!("expected 'width,height', got '{s}'"))?; + + let width = left + .trim() + .parse::() + .map_err(|e| format!("invalid width: {e}"))?; + let height = right + .trim() + .parse::() + .map_err(|e| format!("invalid height: {e}"))?; + + Ok((width, height)) +}