Add some new metrics for errors and per table stuff

src/bin/stream/bootstrap.rs

@@ -428,6 +428,7 @@ pub(crate) async fn run_bootstrap(
                         by_database: vec![boot_db.series()],
                         ..stage_gauges(&StageCounters {
                             emitter: Some(&stats),
+                            db_name: &|_| boot_db.database.clone(),
                             oracle: [None, Some(&oracle_stats)],
                             bootstrap: Some(&progress),
                             bootstrap_attempt: attempt,

src/bin/stream/metrics_publish.rs

@@ -10,6 +10,7 @@ use walshadow::boundary_hold::BoundaryHoldStats;
 use walshadow::ch_emitter::EmitterStats;
 use walshadow::config::ConfigResolver;
 use walshadow::metrics::{DbSeries, MetricsRegistry, MetricsSnapshot};
+use walshadow::pipeline::row_ledger::TableRowCounts;
 use walshadow::pos::{
     Drain, EmitterAck, FilterDispatched, Floor, Pos, ShadowReplay, SourceReceived,
 };
@@ -416,6 +417,8 @@ pub(crate) async fn populate_pipeline_metrics(
 /// counters sum into one series
 pub(crate) struct StageCounters<'a> {
     pub(crate) emitter: Option<&'a walshadow::ch_emitter::EmitterStats>,
+    /// `database=` label for a source database oid
+    pub(crate) db_name: &'a (dyn Fn(u32) -> String + Sync),
     /// Live first, bootstrap's second. Only one of the pair is ever serving
     pub(crate) oracle: [Option<&'a walshadow::oracle::OracleStats>; 2],
     pub(crate) bootstrap: Option<&'a BootstrapProgress>,
@@ -432,12 +435,26 @@ pub(crate) fn emitter_counts<const N: usize>(
     picks.map(|pick| stats.map_or(0, |s| pick(s).load(Ordering::Relaxed)))
 }
 
+/// Stop conditions beyond the crossing and shadow views
 pub(crate) fn stage_gauges(v: &StageCounters<'_>) -> MetricsSnapshot {
     stage_gauges_on(v, MetricsSnapshot::default())
 }
 
+const UNMAPPED_STATUS_ROWS: usize = 20;
+
 pub(crate) fn stage_gauges_on(v: &StageCounters<'_>, base: MetricsSnapshot) -> MetricsSnapshot {
     let (proc_cpu, proc_rss, proc_threads) = read_process_stats();
+    let by_table = |pick: fn(&EmitterStats) -> &TableRowCounts| {
+        v.emitter.map_or_else(Vec::new, |s| {
+            pick(s)
+                .snapshot()
+                .into_iter()
+                .map(|(key, rows)| ((v.db_name)(key.db_oid), key.rel.to_string(), rows))
+                .collect::<Vec<_>>()
+        })
+    };
+    let mut unmapped_rows_by_table = by_table(|s| &s.unmapped_rows_by_table);
+    unmapped_rows_by_table.truncate(UNMAPPED_STATUS_ROWS);
     let emitter = |pick: fn(&EmitterStats) -> &AtomicU64| -> u64 {
         v.emitter.map_or(0, |s| pick(s).load(Ordering::Relaxed))
     };
@@ -538,6 +555,17 @@ pub(crate) fn stage_gauges_on(v: &StageCounters<'_>, base: MetricsSnapshot) -> M
         emitter_xacts_total: emitter(|s| &s.xacts_committed),
         emitter_unsupported_relations: emitter(|s| &s.unsupported_relations),
         emitter_deletes_discarded: emitter(|s| &s.deletes_discarded),
+        unmapped_rows_by_table,
+        rows_inserted_by_table: by_table(|s| &s.rows_inserted_by_table),
+        cells_inserted_by_type: v.emitter.map_or_else(Vec::new, |s| {
+            s.cells_inserted_by_type
+                .snapshot()
+                .into_iter()
+                .map(|((ty, enc), n)| (ty, enc, n))
+                .collect()
+        }),
+        emitter_retries_attempted_total: emitter(|s| &s.retries_attempted),
+        emitter_reconnects_total: emitter(|s| &s.reconnects),
         oracle_local_columns_total: emitter(|s| &s.oracle_local_columns),
         oracle_blocks_total: oracle(|s| &s.blocks),
         oracle_rows_total: oracle(|s| &s.rows),
@@ -643,6 +671,7 @@ mod tests {
                     boundary_hold: &boundary,
                     by_database: Vec::new(),
                     counters: StageCounters {
+                        db_name: &|_| String::new(),
                         emitter: Some(&emitter),
                         oracle: [None, None],
                         bootstrap: None,
@@ -667,11 +696,50 @@ mod tests {
         }
     }
 
+    /// The publisher names each counted table by resolving its `db_oid`, so a
+    /// ledger with rows has to reach the snapshot under the database's name
+    #[test]
+    fn per_table_counters_resolve_their_database_name() {
+        use walshadow::schema::{RelName, TableKey};
+        let emitter = EmitterStats::default();
+        let orders = TableKey::new(5, RelName::new("public", "orders"));
+        let noise = TableKey::new(9, RelName::new("public", "noise"));
+        emitter
+            .rows_inserted_by_table
+            .counter(&orders)
+            .fetch_add(7, Ordering::Relaxed);
+        emitter
+            .unmapped_rows_by_table
+            .counter(&noise)
+            .fetch_add(3, Ordering::Relaxed);
+        let snap = stage_gauges(&StageCounters {
+            db_name: &|oid| match oid {
+                5 => "app".to_owned(),
+                _ => format!("db{oid}"),
+            },
+            emitter: Some(&emitter),
+            oracle: [None, None],
+            bootstrap: None,
+            bootstrap_attempt: 0,
+            uptime_secs: 0,
+        });
+        assert_eq!(
+            snap.rows_inserted_by_table,
+            vec![("app".to_owned(), "public.orders".to_owned(), 7)],
+        );
+        assert_eq!(
+            snap.unmapped_rows_by_table,
+            vec![("db9".to_owned(), "public.noise".to_owned(), 3)],
+            "an oid the config does not name still has to be reportable",
+        );
+    }
+
     #[test]
     fn archive_stage_metrics_refresh_without_resetting_recovery_state() {
         use walshadow::record::WAL_SEG_SIZE;
         let emitter = EmitterStats::default();
         let counters = StageCounters {
+            db_name: &|_| String::new(),
             emitter: Some(&emitter),
             oracle: [None, None],
             bootstrap: None,
@@ -726,6 +794,7 @@ mod tests {
         boot_oracle.rows.fetch_add(7, Ordering::Relaxed);
 
         let snap = stage_gauges(&StageCounters {
+            db_name: &|_| String::new(),
             emitter: None,
             oracle: [Some(&live_oracle), Some(&boot_oracle)],
             bootstrap: None,
@@ -737,6 +806,7 @@ mod tests {
         assert_eq!(snap.bootstrap_attempt, 2);
 
         let boot_only = stage_gauges(&StageCounters {
+            db_name: &|_| String::new(),
             emitter: None,
             oracle: [None, Some(&boot_oracle)],
             bootstrap: None,

src/bin/stream/session.rs

@@ -1895,6 +1895,13 @@ pub(crate) async fn run_session(
             metrics_dbs.iter().map(DbMetricSources::series).collect(),
             StageCounters {
                 emitter: emitter_stats,
+                db_name: &|oid| {
+                    db_conns
+                        .iter()
+                        .find(|c| c.oid == oid)
+                        .map(|c| c.name.clone())
+                        .unwrap_or_default()
+                },
                 oracle: [oracle_stats, bootstrap_metrics.as_ref().map(|b| &*b.oracle)],
                 bootstrap: bootstrap_metrics.as_ref().map(|b| &b.progress),
                 bootstrap_attempt: 0,

src/bin/stream/tracing_setup.rs

@@ -81,6 +81,7 @@ pub(crate) fn init_tracing(
         .with(filter)
         .with(fmt_layer)
         .with(otel_layer)
+        .with(walshadow::log_events::LogEventLayer)
         .try_init();
     provider
 }

src/emit/ch_emitter.rs

@@ -1394,6 +1394,13 @@ pub(crate) enum ColumnEncoding {
 }
 
 impl ColumnEncoding {
+    pub(crate) fn label(&self) -> &'static str {
+        match self {
+            Self::Local => "local",
+            Self::Oracle { .. } => "oracle",
+        }
+    }
+
     fn choose(att: Option<&RelAttr>, ast: &TypeAst) -> Self {
         let Some(att) = att else {
             // Predeclared post-ALTER column has only default cells
@@ -2551,6 +2558,9 @@ crate::atomic_stats! {
         /// (memoised)
         pub route_snapshots_mapped,
         pub route_snapshots_unmapped,
+        pub unmapped_rows_by_table: crate::emit::pipeline::row_ledger::TableRowCounts,
+        pub rows_inserted_by_table: crate::emit::pipeline::row_ledger::TableRowCounts,
+        pub cells_inserted_by_type: crate::emit::pipeline::row_ledger::CellCounts,
         /// Commit-resolve raw decode: records by verdict kind, per op
         /// (`raw_decode_records_total{kind,op}`)
         pub raw_decode_toast_ops: crate::decode::heap_decoder::OpCounters,

src/emit/pipeline/batcher.rs

@@ -20,7 +20,7 @@
 
 use std::collections::hash_map::Entry;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::time::Duration;
 
 use clickhouse_c::Allocator;
@@ -32,7 +32,8 @@ use tokio::time::Instant;
 use crate::config::ResolvedConfig;
 use crate::decode::heap_decoder::{CommittedTuple, HeapOp};
 use crate::emit::ch_emitter::{
-    Append, ColumnBuf, EmitterStats, OP_DELETE, OP_INSERT, OP_UPDATE, TableEncoder, TablePlan,
+    Append, ColumnBuf, ColumnEncoding, EmitterStats, OP_DELETE, OP_INSERT, OP_UPDATE, TableEncoder,
+    TablePlan,
 };
 use crate::emit::pipeline::{DEFAULT_PIPELINE_FLUSH, Fatal};
 use crate::emit::route::RouteSnapshot;
@@ -67,6 +68,7 @@ pub struct RowChunk {
 pub struct ColMeta {
     pub name: String,
     pub type_repr: String,
+    pub cells_inserted: Arc<AtomicU64>,
 }
 
 /// Immutable per-table block shape, shared by every batch of that table
@@ -78,16 +80,26 @@ pub struct BatchMeta {
     /// synthetic ones (lsn, xid, commit_ts, delete marker when configured).
     pub columns: Vec<ColMeta>,
     pub schema_epoch: u64,
+    pub rows_inserted: Arc<AtomicU64>,
 }
 
 impl BatchMeta {
-    fn from_plan(plan: &TablePlan, table_key: TableKey, schema_epoch: u64) -> Self {
+    fn from_plan(
+        plan: &TablePlan,
+        table_key: TableKey,
+        schema_epoch: u64,
+        stats: &EmitterStats,
+    ) -> Self {
+        let col_meta = |name: &String, type_repr: &String, encoding: &ColumnEncoding| ColMeta {
+            name: name.clone(),
+            type_repr: type_repr.clone(),
+            cells_inserted: stats
+                .cells_inserted_by_type
+                .counter(&(type_repr.clone(), encoding.label())),
+        };
         let mut columns = Vec::with_capacity(plan.columns.len() + 4);
         for c in &plan.columns {
-            columns.push(ColMeta {
-                name: c.name.clone(),
-                type_repr: c.type_repr.clone(),
-            });
+            columns.push(col_meta(&c.name, &c.type_repr, &c.encoding));
         }
         for synth in [
             Some(&plan.synth_lsn),
@@ -98,12 +110,10 @@ impl BatchMeta {
         .into_iter()
         .flatten()
         {
-            columns.push(ColMeta {
-                name: synth.name.clone(),
-                type_repr: synth.type_repr.clone(),
-            });
+            columns.push(col_meta(&synth.name, &synth.type_repr, &synth.encoding));
         }
         Self {
+            rows_inserted: stats.rows_inserted_by_table.counter(&table_key),
             table_key,
             insert_sql: plan.insert_sql.clone(),
             columns,
@@ -333,7 +343,12 @@ async fn handle_row(
                 row.route.system_columns(),
             )
             .map_err(|e| e.to_string())?;
-            let meta = Arc::new(BatchMeta::from_plan(&plan, e.key().clone(), ctx.epoch));
+            let meta = Arc::new(BatchMeta::from_plan(
+                &plan,
+                e.key().clone(),
+                ctx.epoch,
+                ctx.stats,
+            ));
             let enc = TableEncoder::new(plan).map_err(|e| e.to_string())?;
             e.insert(Table {
                 enc,

src/emit/pipeline/inserter.rs

@@ -172,6 +172,14 @@ impl Inserter {
                     self.stats
                         .rows_emitted
                         .fetch_add(batch.n_rows as u64, Ordering::Relaxed);
+                    batch
+                        .meta
+                        .rows_inserted
+                        .fetch_add(batch.n_rows as u64, Ordering::Relaxed);
+                    for col in &batch.meta.columns {
+                        col.cells_inserted
+                            .fetch_add(batch.n_rows as u64, Ordering::Relaxed);
+                    }
                     self.stats.blocks_sent.fetch_add(1, Ordering::Relaxed);
                     self.stats
                         .inserter_batches_in

src/emit/pipeline/mod.rs

@@ -19,6 +19,7 @@ pub mod plan_spool;
 pub mod planner;
 pub mod reorder;
 pub mod resolver;
+pub mod row_ledger;
 pub mod tail;
 
 use std::sync::Arc;

src/emit/pipeline/reorder.rs

@@ -30,7 +30,7 @@ use crate::decode::visibility::{PgXactPatch, PgXactView, read_pg_xact};
 use crate::emit::ch_ddl::DdlApplicator;
 use crate::emit::ch_emitter::EmitterStats;
 use crate::record::{Record, RecordSink, SinkError};
-use crate::schema::{RelDescriptor, RelName, SchemaEvent};
+use crate::schema::{RelDescriptor, RelName, SchemaEvent, TableKey};
 use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
 use tracing::Instrument;
 
@@ -1161,31 +1161,40 @@ impl<'a> ReorderRouteView<'a> {
 impl PlanRouteView for ReorderRouteView<'_> {
     fn route_for(&mut self, heap: &DescribedHeap) -> Option<Arc<RouteSnapshot>> {
         let rel_name = &heap.descriptor.rel_name;
-        if let Some(r) = self.memo.get(rel_name) {
-            return r.clone();
-        }
-        let mapped = match self.overlay.get(rel_name) {
-            Some(o) => o.clone(),
-            None => self.mapping.as_ref().and_then(|m| m.get(rel_name)).cloned(),
+        let route = if let Some(r) = self.memo.get(rel_name) {
+            r.clone()
+        } else {
+            let mapped = match self.overlay.get(rel_name) {
+                Some(o) => o.clone(),
+                None => self.mapping.as_ref().and_then(|m| m.get(rel_name)).cloned(),
+            };
+            let route = mapped.map(|m| {
+                let rules = self
+                    .config
+                    .as_ref()
+                    .map_or_else(Arc::default, |rc| rc.column_rules.clone());
+                let policy = self.row_policy.for_rel(self.config.as_deref(), rel_name);
+                RouteSnapshot::freeze(Arc::new(m), rules, policy)
+            });
+            let result = if route.is_none() {
+                self.stats
+                    .unsupported_relations
+                    .fetch_add(1, Ordering::Relaxed);
+                &self.stats.route_snapshots_unmapped
+            } else {
+                &self.stats.route_snapshots_mapped
+            };
+            result.fetch_add(1, Ordering::Relaxed);
+            self.memo.insert(rel_name.clone(), route.clone());
+            route
         };
-        let route = mapped.map(|m| {
-            let rules = self
-                .config
-                .as_ref()
-                .map_or_else(Arc::default, |rc| rc.column_rules.clone());
-            let policy = self.row_policy.for_rel(self.config.as_deref(), rel_name);
-            RouteSnapshot::freeze(Arc::new(m), rules, policy)
-        });
-        let result = if route.is_none() {
+        if route.is_none() {
+            let key = TableKey::new(heap.descriptor.rfn.db_node, rel_name.clone());
             self.stats
-                .unsupported_relations
+                .unmapped_rows_by_table
+                .counter(&key)
                 .fetch_add(1, Ordering::Relaxed);
-            &self.stats.route_snapshots_unmapped
-        } else {
-            &self.stats.route_snapshots_mapped
-        };
-        result.fetch_add(1, Ordering::Relaxed);
-        self.memo.insert(rel_name.clone(), route.clone());
+        }
         route
     }