fix: Make metric names potentially less confusing

ingester/src/handler.rs

@@ -453,7 +453,7 @@ mod tests {
             .unwrap()
             .get_observer(&Attributes::from(&[
                 ("kafka_topic", "whatevs"),
-                ("sequencer_id", "0"),
+                ("kafka_partition", "0"),
                 ("result", "success"),
             ]))
             .unwrap()
@@ -467,7 +467,7 @@ mod tests {
             .unwrap()
             .get_observer(&Attributes::from(&[
                 ("kafka_topic", "whatevs"),
-                ("sequencer_id", "0"),
+                ("kafka_partition", "0"),
             ]))
             .unwrap()
             .fetch();
@@ -479,7 +479,7 @@ mod tests {
             .unwrap()
             .get_observer(&Attributes::from(&[
                 ("kafka_topic", "whatevs"),
-                ("sequencer_id", "0"),
+                ("kafka_partition", "0"),
             ]))
             .unwrap()
             .fetch();
@@ -491,7 +491,7 @@ mod tests {
             .unwrap()
             .get_observer(&Attributes::from(&[
                 ("kafka_topic", "whatevs"),
-                ("sequencer_id", "0"),
+                ("kafka_partition", "0"),
             ]))
             .unwrap()
             .fetch();
@@ -503,7 +503,7 @@ mod tests {
             .unwrap()
             .get_observer(&Attributes::from(&[
                 ("kafka_topic", "whatevs"),
-                ("sequencer_id", "0"),
+                ("kafka_partition", "0"),
             ]))
             .unwrap()
             .fetch();

ingester/src/stream_handler/handler.rs

@@ -414,7 +414,7 @@ fn metric_attrs(
     data_loss: bool,
 ) -> Attributes {
     let mut attr = Attributes::from([
-        ("sequencer_id", partition.to_string().into()),
+        ("kafka_partition", partition.to_string().into()),
         ("kafka_topic", topic.to_string().into()),
     ]);
 
@@ -649,7 +649,7 @@ mod tests {
                         .expect("did not find ttbr metric")
                         .get_observer(&Attributes::from([
                             ("kafka_topic", TEST_KAFKA_TOPIC.into()),
-                            ("sequencer_id", TEST_KAFKA_PARTITION.to_string().into()),
+                            ("kafka_partition", TEST_KAFKA_PARTITION.to_string().into()),
                         ]))
                         .expect("did not match metric attributes")
                         .fetch();

ingester/src/stream_handler/sink_instrumentation.rs

@@ -84,7 +84,7 @@ where
         metrics: &metric::Registry,
     ) -> Self {
         let attr = Attributes::from([
-            ("sequencer_id", kafka_partition.to_string().into()),
+            ("kafka_partition", kafka_partition.to_string().into()),
             ("kafka_topic", kafka_topic_name.into()),
         ]);
 
@@ -199,7 +199,7 @@ where
             .expect("entry from write buffer must be sequenced");
         assert_eq!(
             sequence.sequencer_id as i32, self.sequencer_id,
-            "instrumentation for sequencer {} saw op from sequencer {}",
+            "instrumentation for kafka partition {} saw op from kafka partition {}",
             self.sequencer_id, sequence.sequencer_id,
         );
 
@@ -279,7 +279,7 @@ mod tests {
         /// The attributes assigned to the metrics emitted by the
         /// instrumentation when using the above sequencer / kafka topic values.
         static ref DEFAULT_ATTRS: Attributes = Attributes::from([
-            ("sequencer_id", SEQUENCER_ID.to_string().into()),
+            ("kafka_partition", SEQUENCER_ID.to_string().into()),
             ("kafka_topic", TEST_KAFKA_TOPIC.into()),
         ]);
     }
@@ -621,12 +621,12 @@ mod tests {
 
     // The instrumentation emits per-sequencer metrics, so upon observing an op
     // for a different sequencer it should panic.
-    #[should_panic = "instrumentation for sequencer 42 saw op from sequencer 52"]
+    #[should_panic = "instrumentation for kafka partition 42 saw op from kafka partition 52"]
     #[tokio::test]
     async fn test_op_different_sequencer_id() {
         let metrics = metric::Registry::default();
         let meta = DmlMeta::sequenced(
-            // A different sequencer ID from what the handler is configured to
+            // A different kafka partition ID from what the handler is configured to
             // be instrumenting
             Sequence::new(SEQUENCER_ID + 10, 100),
             *TEST_TIME,

router/src/dml_handlers/sharded_write_buffer.rs

@@ -119,7 +119,7 @@ where
 
             trace!(
                 %partition_key,
-                sequencer_id=%sequencer.id(),
+                kafka_partition=%sequencer.id(),
                 tables=%dml.table_count(),
                 %namespace,
                 approx_size=%dml.size(),

router/src/sequencer.rs

@@ -45,11 +45,11 @@ impl Sequencer {
         );
 
         let enqueue_success = write.recorder([
-            ("shard_id", Cow::from(id.to_string())),
+            ("kafka_partition", Cow::from(id.to_string())),
             ("result", Cow::from("success")),
         ]);
         let enqueue_error = write.recorder([
-            ("shard_id", Cow::from(id.to_string())),
+            ("kafka_partition", Cow::from(id.to_string())),
             ("result", Cow::from("error")),
         ]);
 

router/tests/http.rs

@@ -227,7 +227,7 @@ async fn test_write_ok() {
         .get_instrument::<Metric<U64Histogram>>("sequencer_enqueue_duration_ms")
         .expect("failed to read metric")
         .get_observer(&Attributes::from(&[
-            ("shard_id", "0"),
+            ("kafka_partition", "0"),
             ("result", "success"),
         ]))
         .expect("failed to get observer")

write_summary/src/lib.rs

@@ -47,7 +47,10 @@ impl WriteSummary {
 
         let mut sequencers = BTreeMap::new();
         for s in sequences {
-            let sequencer_id: i32 = s.sequencer_id.try_into().expect("Invalid sequencer id");
+            let sequencer_id: i32 = s
+                .sequencer_id
+                .try_into()
+                .expect("Invalid Kafka Partition id");
 
             // This is super confusing: "sequencer_id" in the router
             //  and other parts of the codebase refers to what the