diff --git a/src/api/metrics.rs b/src/api/metrics.rs new file mode 100644 index 00000000..f4e13c98 --- /dev/null +++ b/src/api/metrics.rs @@ -0,0 +1,39 @@ +/* + * Copyright (C) 2026 Fastly, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +use prometheus::Counter; +use std::sync::OnceLock; + +static TOTAL_REQUESTS: OnceLock = OnceLock::new(); + +pub fn total_requests() -> &'static Counter { + TOTAL_REQUESTS.get_or_init(|| { + Counter::new( + "requests_total", + "Total number of requests processed by api", + ) + .expect("failed to create total_requests counter") + }) +} + +pub fn init(registry: &prometheus::Registry) { + // Pre-initialize metrics so they appear even before any requests arrive, + // and register them with the provided registry. + + registry + .register(Box::new(total_requests().clone())) + .expect("failed to register total_requests counter"); +} diff --git a/src/api/mod.rs b/src/api/mod.rs index 76e79e78..3a32a59a 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -15,9 +15,13 @@ */ pub mod item; +pub mod metrics; pub mod server; use self::server::Server; +use crate::core::config::NetListenConfig; +use crate::core::net::NetListener; +use crate::core::prometheus::{try_register_process_collector, PrometheusServer}; use crate::core::zmq::SpecInfo; use log::{debug, info}; use signal_hook; @@ -40,6 +44,11 @@ pub enum ListenSpec { }, } +pub struct PrometheusConfig { + pub listen_config: NetListenConfig, + pub prefix: String, +} + pub struct Config { pub maxconn: usize, pub buffer_size: usize, @@ -49,11 +58,13 @@ pub struct Config { pub item_out: Vec, pub item_out_bind: bool, pub ipc_file_mode: u32, + pub prometheus: Option, } pub struct App { _server: Server, _zmq_context: zmq::Context, + _prometheus: Option, } impl App { @@ -78,6 +89,28 @@ impl App { let zmq_context = zmq::Context::new(); + let mut prometheus = None; + + if let Some(config) = &config.prometheus { + let trimmed = config.prefix.trim_end_matches('_'); + let registry = prometheus::Registry::new_custom( + (!trimmed.is_empty()).then(|| trimmed.to_string()), + None, + ) + .expect("failed to create prometheus registry"); + + try_register_process_collector(®istry) + .expect("failed to register process collector"); + + crate::core::log::init_metrics(®istry); + metrics::init(®istry); + + let l = NetListener::bind_config(&config.listen_config) + .map_err(|e| format!("prometheus listener: {e}"))?; + + prometheus = Some(PrometheusServer::new(l, registry)); + } + let server = Server::new( config.maxconn, config.buffer_size, @@ -91,6 +124,7 @@ impl App { Ok(Self { _server: server, _zmq_context: zmq_context, + _prometheus: prometheus, }) } diff --git a/src/api/server.rs b/src/api/server.rs index 1ac2eb02..87e04712 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -15,6 +15,7 @@ */ use crate::api::item::{validate_item, InPublishItems}; +use crate::api::metrics; use crate::api::ListenSpec; use crate::core::fs::{set_group, set_user}; use crate::core::net::NetListener; @@ -150,6 +151,8 @@ async fn handle_request(state: &State, req: Request) -> Response { item_out_sock: &state.item_out_sock, }; + metrics::total_requests().inc(); + match req.uri.as_str() { "/publish" => publish_noslash(), "/publish/" => publish(&ctx, req).await, diff --git a/src/bin/pushpin-api.rs b/src/bin/pushpin-api.rs index ecd6e763..1b31ed77 100644 --- a/src/bin/pushpin-api.rs +++ b/src/bin/pushpin-api.rs @@ -16,7 +16,7 @@ use clap::{Arg, ArgAction, Command}; use log::{error, LevelFilter}; -use pushpin::api::{run, Config, ListenSpec}; +use pushpin::api::{run, Config, ListenSpec, PrometheusConfig}; use pushpin::core::config::NetListenConfig; use pushpin::core::log::{get_simple_logger, local_offset_check}; use pushpin::core::version; @@ -35,6 +35,7 @@ struct Args { item_out_specs: Vec, item_out_bind: bool, ipc_file_mode: u32, + prometheus: Option, } fn process_args_and_run(args: Args) -> Result<(), Box> { @@ -51,6 +52,7 @@ fn process_args_and_run(args: Args) -> Result<(), Box> { item_out: args.item_out_specs, item_out_bind: args.item_out_bind, ipc_file_mode: args.ipc_file_mode, + prometheus: None, }; for v in args.listen.iter() { @@ -83,6 +85,31 @@ fn process_args_and_run(args: Args) -> Result<(), Box> { config.listen.push(spec); } + if let Some(v) = &args.prometheus { + let mut spec: NetListenConfig = match v.parse() { + Ok(s) => s, + Err(e) => { + return Err(format!("failed to parse prometheus: {}", e).into()); + } + }; + + let params = match &mut spec { + NetListenConfig::Tcp(c) => &mut c.params, + NetListenConfig::Unix(c) => &mut c.params, + }; + + let prefix = params.remove("prefix").unwrap_or("api".to_string()); + + if let Some(k) = params.keys().next() { + return Err(format!("failed to parse prometheus: invalid param: {}", k).into()); + } + + config.prometheus = Some(PrometheusConfig { + listen_config: spec, + prefix, + }); + } + run(&config) } @@ -160,6 +187,13 @@ fn main() { .value_name("octal") .help("Permissions for ZeroMQ IPC binds"), ) + .arg( + Arg::new("prometheus") + .long("prometheus") + .num_args(1) + .value_name("[addr:]port[,params...]") + .help("Address/port to serve Prometheus metrics on"), + ) .get_matches(); // Allow all log levels globally so individual loggers can limit as they choose @@ -264,6 +298,8 @@ fn main() { listen.push("0.0.0.0:5561".to_string()); } + let prometheus = matches.get_one::("prometheus").cloned(); + let args = Args { maxconn, buffer_size, @@ -273,6 +309,7 @@ fn main() { item_out_specs, item_out_bind, ipc_file_mode, + prometheus, }; if let Err(e) = process_args_and_run(args) {