Labyricorn/signal-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rename "plotter" as "prometheus", make plot feature optional in signal-gateway

Browse Source
This commit is contained in:
Chris Beck
2025-12-04 23:35:04 -07:00
parent 7b7b6bcac7
commit 5c6bbd6f96
7 changed files with 311 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -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"] }
signal-gateway/Cargo.toml
+9 -4
View File
@@ -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 }
signal-gateway/src/gateway/circular_buffer.rs
+2
View File
@@ -44,11 +44,13 @@ impl<T> CircularBuffer<T> {
}
/// 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
}
signal-gateway/src/gateway/mod.rs
+55 -38
View File
@@ -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<String>,
#[conf(flatten)]
pub plotter: Option<PlotterConfig>,
pub prometheus: Option<PrometheusConfig>,
#[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<Origin>,
text: String,
attachment_paths: Vec<String>,
attachment_paths: Vec<PathBuf>,
/// 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<String>,
@@ -193,7 +196,7 @@ pub struct Gateway {
admin_mq_tx: UnboundedSender<AdminMessage>,
admin_mq_rx: Mutex<UnboundedReceiver<AdminMessage>>,
token: CancellationToken,
plotter: Option<Plotter>,
prometheus: Option<Prometheus>,
/// Log handlers keyed by origin (app + host). Lazily created when first message from an origin arrives.
log_handlers: RwLock<HashMap<Origin, LogHandler>>,
}
@@ -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<String>), (u16, Box<dyn Error>)> {
) -> Result<(String, Vec<PathBuf>), (u16, Box<dyn Error>)> {
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<String>), (u16, Box<dyn Error>)> {
) -> Result<(String, Vec<PathBuf>), (u16, Box<dyn Error>)> {
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);
}
signal-gateway/src/main.rs
+1 -1
View File
@@ -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;
signal-gateway/src/plotter.rs → signal-gateway/src/prometheus/mod.rs
+102 -194
View File
@@ -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<FixedOffset>,
/// Timezone name for plot timestamps, e.g. "US/Mountain" or "Europe/Paris".
/// Mutually exclusive with --utc-offset.
#[conf(long, env)]
pub timezone: Option<Tz>,
/// 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::<u32>()
.map_err(|e| format!("invalid width: {e}"))?;
let height = right
.trim()
.parse::<u32>()
.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<String>,
}
impl Plotter {
pub fn new(config: PlotterConfig) -> Result<Self, &'static str> {
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<Self, &'static str> {
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<PlotThreshold>,
title: Option<&str>,
) -> Result<String, Box<dyn Error>> {
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<String, Box<dyn Error>> {
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<String, Box<dyn Error>> {
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<Item: AsRef<str>>,
@@ -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<Item: AsRef<str>>,
@@ -251,6 +90,7 @@ impl Plotter {
.await?)
}
/// Get the list of current alerts
pub async fn alerts(&self) -> Result<Vec<AlertInfo>, Box<dyn Error>> {
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<PathBuf, Box<dyn Error>> {
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<PathBuf, Box<dyn Error>> {
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<String, String>,
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<dyn std::error::Error>> {
// Look for comparison operators with surrounding spaces
for comparator in [" < ", " > "] {
signal-gateway/src/prometheus/plot.rs
+141
View File
@@ -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<FixedOffset>,
/// Timezone name for plot timestamps, e.g. "US/Mountain" or "Europe/Paris".
/// Mutually exclusive with --utc-offset.
#[conf(long, env)]
timezone: Option<Tz>,
/// 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<String>,
}
impl PlotConfig {
pub fn create_plot(
&self,
matrix: &[MetricTimeseries],
threshold: Option<PlotThreshold>,
title: Option<&str>,
) -> Result<PathBuf, Box<dyn Error>> {
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::<u32>()
.map_err(|e| format!("invalid width: {e}"))?;
let height = right
.trim()
.parse::<u32>()
.map_err(|e| format!("invalid height: {e}"))?;
Ok((width, height))
}