Measure RocksDB batch put bytes by cf (#19038)

We measure number of bytes by cf for `put`, and measure total written in
the batch by DB, but do not have breakdown for batch written bytes by
cf. This metrics adds the breakdown by cf for batch writes

crates/typed-store/src/metrics.rs

@@ -250,6 +250,7 @@ pub struct OperationMetrics {
     pub rocksdb_multiget_bytes: HistogramVec,
     pub rocksdb_put_latency_seconds: HistogramVec,
     pub rocksdb_put_bytes: HistogramVec,
+    pub rocksdb_batch_put_bytes: HistogramVec,
     pub rocksdb_delete_latency_seconds: HistogramVec,
     pub rocksdb_deletes: IntCounterVec,
     pub rocksdb_batch_commit_latency_seconds: HistogramVec,
@@ -343,6 +344,16 @@ impl OperationMetrics {
                 registry,
             )
             .unwrap(),
+            rocksdb_batch_put_bytes: register_histogram_vec_with_registry!(
+                "rocksdb_batch_put_bytes",
+                "Rocksdb batch put call puts data size in bytes",
+                &["cf_name"],
+                prometheus::exponential_buckets(1.0, 4.0, 15)
+                    .unwrap()
+                    .to_vec(),
+                registry,
+            )
+            .unwrap(),
             rocksdb_delete_latency_seconds: register_histogram_vec_with_registry!(
                 "rocksdb_delete_latency_seconds",
                 "Rocksdb delete latency in seconds",

crates/typed-store/src/rocks/mod.rs

@@ -1451,15 +1451,21 @@ impl DBBatch {
         if !Arc::ptr_eq(&db.rocksdb, &self.rocksdb) {
             return Err(TypedStoreError::CrossDBBatch);
         }
-
+        let mut total = 0usize;
         new_vals
             .into_iter()
             .try_for_each::<_, Result<_, TypedStoreError>>(|(k, v)| {
                 let k_buf = be_fix_int_ser(k.borrow())?;
                 let v_buf = bcs::to_bytes(v.borrow()).map_err(typed_store_err_from_bcs_err)?;
+                total += k_buf.len() + v_buf.len();
                 self.batch.put_cf(&db.cf(), k_buf, v_buf);
                 Ok(())
             })?;
+        self.db_metrics
+            .op_metrics
+            .rocksdb_batch_put_bytes
+            .with_label_values(&[&db.cf])
+            .observe(total as f64);
         Ok(self)
     }