feat: add metric for certificate expirations

prosa/examples/proc.rs

@@ -6,6 +6,7 @@ use prosa::core::proc::{Proc, ProcBusParam, ProcConfig, proc};
 use prosa::core::settings::tracing::TelemetryFilter;
 use prosa::core::settings::{ProsaConfig, Settings, settings};
 use prosa::event::pending::PendingMsgs;
+use prosa::io::listener::ListenerSetting;
 use prosa::stub::adaptor::StubAsyncParotAdaptor;
 use prosa::stub::proc::{StubProc, StubSettings};
 use prosa::tracing::{debug, info, warn};
@@ -16,6 +17,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::time;
 use tracing::metadata::LevelFilter;
+use url::Url;
 
 #[derive(Default, Adaptor)]
 struct MyAdaptor {}
@@ -235,6 +237,18 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     );
     Proc::<MyAdaptor>::run(proc2)?;
 
+    // SSL listener serving a self signed certificate, exported in the `prosa_ssl_certificates` metric
+    let listener = ListenerSetting::new(Url::parse("https://localhost:4443")?, None)
+        .bind()
+        .await?;
+    tokio::spawn(async move {
+        loop {
+            if let Err(err) = listener.accept().await {
+                warn!("SSL listener can't accept a client: {err}");
+            }
+        }
+    });
+
     // Wait on main task
     main.run().await;
 

prosa/src/core/main.rs

@@ -556,6 +556,14 @@ where
             })
             .build();
 
+        // Monitor SSL certificates expiration
+        #[cfg(feature = "openssl")]
+        self.meter
+            .i64_observable_gauge("prosa_ssl_certificates")
+            .with_description("Expiration date, in Unix seconds, of the SSL certificates loaded")
+            .with_callback(|observer| crate::io::ssl::observe_certificates(observer))
+            .build();
+
         let mut proc_names = IntHashMap::with_hasher(BuildIntHasher::default());
 
         // Monitor processors objects
@@ -834,6 +842,41 @@ mod tests {
             })
     }
 
+    #[cfg(all(feature = "prometheus", feature = "openssl"))]
+    #[tokio::test]
+    async fn exports_ssl_certificates_expiration() {
+        let settings = HealthSettings {
+            observability: Observability::default(),
+        };
+        let (bus, main) = MainProc::<SimpleStringTvf>::create(&settings, Some(1));
+        let registry = bus.get_prometheus_registry().clone();
+        let main_task = tokio::spawn(main.run());
+
+        let listener = crate::io::listener::ListenerSetting::new(
+            url::Url::parse("https://localhost:0").expect("Listener url is valid"),
+            None,
+        )
+        .bind()
+        .await
+        .expect("listener should bind");
+        wait_until(|| gauge_value(&registry, "prosa_ssl_certificates").is_some()).await;
+        let now = std::time::UNIX_EPOCH
+            .elapsed()
+            .expect("time should be after the epoch")
+            .as_secs() as f64;
+        assert!(
+            registry
+                .gather()
+                .into_iter()
+                .filter(|family| family.name() == "prosa_ssl_certificates")
+                .flat_map(|family| family.get_metric().to_vec())
+                .any(|metric| metric.get_gauge().get_value() > now)
+        );
+
+        drop(listener);
+        main_task.abort();
+    }
+
     #[tokio::test]
     async fn readiness_tracks_required_processors_and_services() {
         let observability = yaml_serde::from_str(

prosa/src/io.rs

@@ -18,7 +18,7 @@ pub use prosa_utils::config::{
 pub mod listener;
 pub mod pool;
 #[cfg(feature = "openssl")]
-mod ssl;
+pub(crate) mod ssl;
 pub mod stream;
 
 /// Trait to define ProSA IO.

prosa/src/io/listener.rs

@@ -57,9 +57,10 @@ pub struct SslHandshaker {
     /// Acceptor holding the certificate served to the clients, shared by the clones
     #[cfg(feature = "openssl")]
     pub(super) acceptor: std::sync::Arc<SslContextCache<::openssl::ssl::SslAcceptor>>,
-    /// Configuration and host the acceptor is built from, [`None`] when it was handed over built
+    /// Configuration, host and name of the listener the acceptor is built from, [`None`] when it
+    /// was handed over built
     #[cfg(feature = "openssl")]
-    pub(super) source: Option<std::sync::Arc<(SslConfig, Option<String>)>>,
+    pub(super) source: Option<std::sync::Arc<(SslConfig, Option<String>, String)>>,
     /// Timeout of the SSL handshake with a client
     pub(super) timeout: Duration,
 }
@@ -572,6 +573,7 @@ impl ListenerSetting {
             SslHandshaker::build(
                 self.ssl.clone().unwrap_or_default(),
                 self.url.host_str().map(String::from),
+                self.get_safe_url().to_string(),
             )
             .await
             .map(Some)