api: add prometheus metrics

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<Counter> = 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");
+}

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<String>,
     pub item_out_bind: bool,
     pub ipc_file_mode: u32,
+    pub prometheus: Option<PrometheusConfig>,
 }
 
 pub struct App {
     _server: Server,
     _zmq_context: zmq::Context,
+    _prometheus: Option<PrometheusServer>,
 }
 
 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(&registry)
+                .expect("failed to register process collector");
+
+            crate::core::log::init_metrics(&registry);
+            metrics::init(&registry);
+
+            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,
         })
     }
 

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,

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<String>,
     item_out_bind: bool,
     ipc_file_mode: u32,
+    prometheus: Option<String>,
 }
 
 fn process_args_and_run(args: Args) -> Result<(), Box<dyn Error>> {
@@ -51,6 +52,7 @@ fn process_args_and_run(args: Args) -> Result<(), Box<dyn Error>> {
         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<dyn Error>> {
         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::<String>("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) {