Instrument connection limit and current count for http blackholes

lading/src/blackhole/common.rs

@@ -6,6 +6,7 @@ use hyper_util::{
     server::conn::auto,
 };
 use lading_signal::Watcher;
+use metrics::gauge;
 use std::{net::SocketAddr, sync::Arc};
 use tokio::{
     net::TcpListener,
@@ -26,6 +27,7 @@ pub(crate) async fn run_httpd<SF, S>(
     addr: SocketAddr,
     concurrency_limit: usize,
     shutdown: Watcher,
+    labels: Vec<(String, String)>,
     make_service: SF,
 ) -> Result<(), Error>
 where
@@ -48,9 +50,14 @@ where
     let sem = Arc::new(Semaphore::new(concurrency_limit));
     let mut join_set = JoinSet::new();
 
+    gauge!("connection.limit", &labels).set(concurrency_limit as f64);
+
     let shutdown_fut = shutdown.recv();
     pin!(shutdown_fut);
     loop {
+        let claimed_permits = concurrency_limit - sem.available_permits();
+
+        gauge!("connection.current", &labels).set(claimed_permits as f64);
         tokio::select! {
             () = &mut shutdown_fut => {
                 info!("Shutdown signal received, stopping accept loop.");

lading/src/blackhole/http.rs

@@ -13,7 +13,7 @@ use hyper::{header, Request, Response, StatusCode};
 use metrics::counter;
 use serde::{Deserialize, Serialize};
 use std::{net::SocketAddr, time::Duration};
-use tracing::{debug, error};
+use tracing::error;
 
 use super::General;
 
@@ -238,6 +238,7 @@ impl Http {
             self.httpd_addr,
             self.concurrency_limit,
             self.shutdown,
+            self.metric_labels.clone(),
             move || {
                 let metric_labels = self.metric_labels.clone();
                 let body_bytes = self.body_bytes.clone();
@@ -246,7 +247,6 @@ impl Http {
                 let response_delay = self.response_delay;
 
                 hyper::service::service_fn(move |req| {
-                    debug!("REQUEST: {:?}", req);
                     srv(
                         status,
                         metric_labels.clone(),

lading/src/blackhole/splunk_hec.rs

@@ -8,10 +8,7 @@
 
 use std::{
     net::SocketAddr,
-    sync::{
-        atomic::{AtomicU64, Ordering},
-        Arc,
-    },
+    sync::atomic::{AtomicU64, Ordering},
 };
 
 use bytes::Bytes;
@@ -20,7 +17,6 @@ use hyper::{header, Method, Request, Response, StatusCode};
 use metrics::counter;
 use rustc_hash::FxHashMap;
 use serde::{Deserialize, Serialize};
-use tracing::debug;
 
 use super::General;
 
@@ -90,7 +86,7 @@ struct HecResponse {
 
 async fn srv(
     req: Request<hyper::body::Incoming>,
-    labels: Arc<Vec<(String, String)>>,
+    labels: Vec<(String, String)>,
 ) -> Result<hyper::Response<BoxBody<Bytes, hyper::Error>>, hyper::Error> {
     counter!("requests_received", &*labels).increment(1);
 
@@ -203,18 +199,14 @@ impl SplunkHec {
     ///
     /// None known.
     pub async fn run(self) -> Result<(), Error> {
-        let metric_labels = Arc::new(self.metric_labels.clone());
-
         crate::blackhole::common::run_httpd(
             self.httpd_addr,
             self.concurrency_limit,
             self.shutdown,
+            self.metric_labels.clone(),
             move || {
-                let metric_labels = Arc::clone(&metric_labels);
-                hyper::service::service_fn(move |req| {
-                    debug!("REQUEST: {:?}", req);
-                    srv(req, Arc::clone(&metric_labels))
-                })
+                let metric_labels = self.metric_labels.clone();
+                hyper::service::service_fn(move |req| srv(req, metric_labels.clone()))
             },
         )
         .await?;

lading/src/blackhole/sqs.rs

@@ -101,13 +101,10 @@ impl Sqs {
             self.httpd_addr,
             self.concurrency_limit,
             self.shutdown,
+            self.metric_labels.clone(),
             move || {
                 let metric_labels = self.metric_labels.clone();
-
-                hyper::service::service_fn(move |req| {
-                    debug!("REQUEST: {:?}", req);
-                    srv(req, metric_labels.clone())
-                })
+                hyper::service::service_fn(move |req| srv(req, metric_labels.clone()))
             },
         )
         .await?;