source_to_sink_latency_seconds/sink_to_source_latency_seconds metrics (…

…#1395)
shotover · Jan 10, 2024 · 8740efb · 8740efb
1 parent a7d8aed
commit 8740efb
Show file tree

Hide file tree

Showing 25 changed files with 341 additions and 128 deletions.
diff --git a/docs/src/user-guide/observability.md b/docs/src/user-guide/observability.md
@@ -13,6 +13,8 @@ This interface will serve Prometheus metrics from `/metrics`. The following metr
 | `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_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 |
 
 ## Metric data types
 
@@ -24,6 +26,9 @@ A single value, which can only be incremented, not decremented. Starts out with
 
 Measures the distribution of values for a set of measurements and starts with no initial values.
 
+Every 20 seconds one of the 3 chunks of historical values are cleared.
+This means that values are held onto for around 60 seconds.
+
 ### Gauge
 
 A single value that can increment or decrement over time. Starts out with an initial value of zero.

diff --git a/shotover-proxy/tests/runner/observability_int_tests.rs b/shotover-proxy/tests/runner/observability_int_tests.rs
@@ -16,6 +16,7 @@ async fn test_metrics() {
 # TYPE shotover_chain_messages_per_batch_count summary
 # TYPE shotover_chain_total_count counter
 # TYPE shotover_query_count counter
+# TYPE shotover_sink_to_source_latency_seconds summary
 # TYPE shotover_transform_failures_count counter
 # TYPE shotover_transform_latency_seconds summary
 # TYPE shotover_transform_pushed_failures_count counter
@@ -27,6 +28,7 @@ shotover_chain_failures_count{chain="redis"}
 shotover_chain_messages_per_batch_count_count{chain="redis"}
 shotover_chain_messages_per_batch_count_sum{chain="redis"}
 shotover_chain_messages_per_batch_count{chain="redis",quantile="0"}
+shotover_chain_messages_per_batch_count{chain="redis",quantile="0.1"}
 shotover_chain_messages_per_batch_count{chain="redis",quantile="0.5"}
 shotover_chain_messages_per_batch_count{chain="redis",quantile="0.9"}
 shotover_chain_messages_per_batch_count{chain="redis",quantile="0.95"}
@@ -35,20 +37,32 @@ shotover_chain_messages_per_batch_count{chain="redis",quantile="0.999"}
 shotover_chain_messages_per_batch_count{chain="redis",quantile="1"}
 shotover_chain_total_count{chain="redis"}
 shotover_query_count{name="redis-chain"}
+shotover_sink_to_source_latency_seconds_count{source="redis"}
+shotover_sink_to_source_latency_seconds_sum{source="redis"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0.1"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0.5"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0.9"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0.95"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0.99"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="0.999"}
+shotover_sink_to_source_latency_seconds{source="redis",quantile="1"}
 shotover_transform_failures_count{transform="NullSink"}
 shotover_transform_failures_count{transform="QueryCounter"}
 shotover_transform_latency_seconds_count{transform="NullSink"}
 shotover_transform_latency_seconds_count{transform="QueryCounter"}
 shotover_transform_latency_seconds_sum{transform="NullSink"}
 shotover_transform_latency_seconds_sum{transform="QueryCounter"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="0"}
+shotover_transform_latency_seconds{transform="NullSink",quantile="0.1"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="0.5"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="0.9"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="0.95"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="0.99"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="0.999"}
 shotover_transform_latency_seconds{transform="NullSink",quantile="1"}
 shotover_transform_latency_seconds{transform="QueryCounter",quantile="0"}
+shotover_transform_latency_seconds{transform="QueryCounter",quantile="0.1"}
 shotover_transform_latency_seconds{transform="QueryCounter",quantile="0.5"}
 shotover_transform_latency_seconds{transform="QueryCounter",quantile="0.9"}
 shotover_transform_latency_seconds{transform="QueryCounter",quantile="0.95"}
@@ -62,13 +76,15 @@ shotover_transform_pushed_latency_seconds_count{transform="QueryCounter"}
 shotover_transform_pushed_latency_seconds_sum{transform="NullSink"}
 shotover_transform_pushed_latency_seconds_sum{transform="QueryCounter"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0"}
+shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0.1"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0.5"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0.9"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0.95"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0.99"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="0.999"}
 shotover_transform_pushed_latency_seconds{transform="NullSink",quantile="1"}
 shotover_transform_pushed_latency_seconds{transform="QueryCounter",quantile="0"}
+shotover_transform_pushed_latency_seconds{transform="QueryCounter",quantile="0.1"}
 shotover_transform_pushed_latency_seconds{transform="QueryCounter",quantile="0.5"}
 shotover_transform_pushed_latency_seconds{transform="QueryCounter",quantile="0.9"}
 shotover_transform_pushed_latency_seconds{transform="QueryCounter",quantile="0.95"}
@@ -108,6 +124,7 @@ shotover_transform_total_count{transform="QueryCounter"}
 shotover_chain_latency_seconds_count{chain="redis",client_details="127.0.0.1"}
 shotover_chain_latency_seconds_sum{chain="redis",client_details="127.0.0.1"}
 shotover_chain_latency_seconds{chain="redis",client_details="127.0.0.1",quantile="0"}
+shotover_chain_latency_seconds{chain="redis",client_details="127.0.0.1",quantile="0.1"}
 shotover_chain_latency_seconds{chain="redis",client_details="127.0.0.1",quantile="0.5"}
 shotover_chain_latency_seconds{chain="redis",client_details="127.0.0.1",quantile="0.9"}
 shotover_chain_latency_seconds{chain="redis",client_details="127.0.0.1",quantile="0.95"}

diff --git a/shotover/benches/benches/codec.rs b/shotover/benches/benches/codec.rs
@@ -44,7 +44,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                     || {
                         (
                             // recreate codec since it is stateful
-                            KafkaCodecBuilder::new(Direction::Source).build(),
+                            KafkaCodecBuilder::new(Direction::Source, "kafka".to_owned()).build(),
                             input.clone(),
                         )
                     },
@@ -77,7 +77,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                     || {
                         (
                             // recreate codec since it is stateful
-                            KafkaCodecBuilder::new(Direction::Sink).build(),
+                            KafkaCodecBuilder::new(Direction::Sink, "kafka".to_owned()).build(),
                             messages.clone(),
                         )
                     },
@@ -102,7 +102,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                 || {
                     (
                         // recreate codec since it is stateful
-                        KafkaCodecBuilder::new(Direction::Source).build(),
+                        KafkaCodecBuilder::new(Direction::Source, "kafka".to_owned()).build(),
                         input.clone(),
                     )
                 },
@@ -139,7 +139,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                 || {
                     (
                         // recreate codec since it is stateful
-                        KafkaCodecBuilder::new(Direction::Sink).build(),
+                        KafkaCodecBuilder::new(Direction::Sink, "kafka".to_owned()).build(),
                         messages.clone(),
                     )
                 },
@@ -165,7 +165,8 @@ fn criterion_benchmark(c: &mut Criterion) {
             },
         }))];
 
-        let (_, mut encoder) = CassandraCodecBuilder::new(Direction::Sink).build();
+        let (_, mut encoder) =
+            CassandraCodecBuilder::new(Direction::Sink, "cassandra".to_owned()).build();
 
         group.bench_function("encode_cassandra_system.local_query", |b| {
             b.iter_batched(
@@ -189,7 +190,8 @@ fn criterion_benchmark(c: &mut Criterion) {
             operation: CassandraOperation::Result(peers_v2_result()),
         }))];
 
-        let (_, mut encoder) = CassandraCodecBuilder::new(Direction::Sink).build();
+        let (_, mut encoder) =
+            CassandraCodecBuilder::new(Direction::Sink, "cassandra".to_owned()).build();
 
         group.bench_function("encode_cassandra_system.local_result", |b| {
             b.iter_batched(