Add connections_opened metric

docs/src/user-guide/observability.md

@@ -11,7 +11,8 @@ This optional interface will serve Prometheus metrics from `/metrics`. It will b
 | `shotover_chain_failures_count`            | `chain`     | [counter](#counter)     | Counts the amount of times `chain` fails                                  |
 | `shotover_chain_latency_seconds`           | `chain`     | [histogram](#histogram) | The latency for running `chain`                                           |
 | `shotover_chain_messages_per_batch_count`  | `chain`     | [histogram](#histogram) | The number of messages in each batch passing through `chain`.             |
-| `shotover_available_connections_count`     | `source`    | [gauge](#gauge)         | The number of connections currently connected to `source`                 |
+| `shotover_available_connections_count`     | `source`    | [gauge](#gauge)         | How many more connections can be opened to `source` before new connections will be rejected. |
+| `connections_opened`                       | `source`    | [counter](#counter)     | Counts the total number of connections that clients have opened against this source.         |
 | `shotover_source_to_sink_latency_seconds`  | `sink`      | [histogram](#histogram) | The milliseconds between reading a request from a source TCP connection and writing it to a sink TCP connection  |
 | `shotover_sink_to_source_latency_seconds`  | `source`    | [histogram](#histogram) | The milliseconds between reading a response from a sink TCP connection and writing it to a source TCP connection |
 

shotover-proxy/tests/runner/observability_int_tests.rs

@@ -11,6 +11,7 @@ async fn test_metrics() {
 
     // Expected string looks unnatural because it is sorted in alphabetical order to make it match the sorted error output
     let expected = r#"
+# TYPE connections_opened counter
 # TYPE shotover_available_connections_count gauge
 # TYPE shotover_chain_failures_count counter
 # TYPE shotover_chain_messages_per_batch_count summary
@@ -20,6 +21,7 @@ async fn test_metrics() {
 # TYPE shotover_transform_failures_count counter
 # TYPE shotover_transform_latency_seconds summary
 # TYPE shotover_transform_total_count counter
+connections_opened{source="redis"}
 shotover_available_connections_count{source="redis"}
 shotover_chain_failures_count{chain="redis"}
 shotover_chain_messages_per_batch_count_count{chain="redis"}

shotover/src/server.rs

@@ -10,7 +10,7 @@ use anyhow::{anyhow, Result};
 use bytes::BytesMut;
 use futures::future::join_all;
 use futures::{SinkExt, StreamExt};
-use metrics::{gauge, Gauge};
+use metrics::{counter, gauge, Counter, Gauge};
 use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::sync::Arc;
@@ -65,6 +65,7 @@ pub struct TcpCodecListener<C: CodecBuilder> {
     /// Keep track of how many connections we have received so we can use it as a request id.
     connection_count: u64,
 
+    connections_opened: Counter,
     available_connections_gauge: Gauge,
 
     /// Timeout after which to kill an idle connection. No timeout means connections will never be timed out.
@@ -91,6 +92,7 @@ impl<C: CodecBuilder + 'static> TcpCodecListener<C> {
     ) -> Result<Self, Vec<String>> {
         let available_connections_gauge =
             gauge!("shotover_available_connections_count", "source" => source_name.clone());
+        let connections_opened = counter!("connections_opened", "source" => source_name.clone());
         available_connections_gauge.set(limit_connections.available_permits() as f64);
 
         let chain_usage_config = TransformContextConfig {
@@ -133,6 +135,7 @@ impl<C: CodecBuilder + 'static> TcpCodecListener<C> {
             tls,
             connection_count: 0,
             available_connections_gauge,
+            connections_opened,
             timeout,
             connection_handles: vec![],
             transport,
@@ -187,6 +190,7 @@ impl<C: CodecBuilder + 'static> TcpCodecListener<C> {
                 debug!("got socket");
                 self.available_connections_gauge
                     .set(self.limit_connections.available_permits() as f64);
+                self.connections_opened.increment(1);
 
                 let client_details = stream
                     .peer_addr()