laminar_connectors/mongodb/metrics/

mod.rs

//! `MongoDB` connector metrics.
//!
//! Prometheus-backed counters for tracking CDC source and sink performance.

use prometheus::{IntCounter, IntGauge, Registry};

use super::change_event::ChangeOperation;
use crate::prom::reg_or_local;

/// Metrics for the `MongoDB` CDC source connector.
#[derive(Debug, Clone)]
pub struct MongoDbCdcMetrics {
    /// Total change events received.
    pub events_received: IntCounter,
    /// Total bytes received from the change stream.
    pub bytes_received: IntCounter,
    /// Total errors encountered.
    pub errors: IntCounter,
    /// Total batches produced for downstream.
    pub batches_produced: IntCounter,
    /// Total INSERT operations received.
    pub inserts: IntCounter,
    /// Total UPDATE operations received.
    pub updates: IntCounter,
    /// Total REPLACE operations received.
    pub replaces: IntCounter,
    /// Total DELETE operations received.
    pub deletes: IntCounter,
    /// Total lifecycle, metadata, and unrecognized events.
    pub lifecycle_events: IntCounter,
    /// Collection-metadata events document mode skipped because they leave documents unchanged.
    pub metadata_events_skipped: IntCounter,
    /// Documents copied by the initial snapshot.
    pub snapshot_documents: IntCounter,
    /// 1 while emitted progress is inside the initial snapshot copy, 0 once streaming.
    pub snapshot_in_progress: IntGauge,
    /// Total reconnection attempts.
    pub reconnects: IntCounter,
}

impl MongoDbCdcMetrics {
    /// Creates a new metrics instance with all counters at zero.
    #[must_use]
    #[allow(clippy::missing_panics_doc)]
    pub fn new(registry: Option<&Registry>) -> Self {
        let mut local = None;
        let reg = reg_or_local(registry, &mut local);

        Self {
            events_received: reg.counter(
                "mongodb_cdc_events_received_total",
                "Total MongoDB CDC events received",
            ),
            bytes_received: reg.counter(
                "mongodb_cdc_bytes_received_total",
                "Total bytes from change stream",
            ),
            errors: reg.counter("mongodb_cdc_errors_total", "Total MongoDB CDC errors"),
            batches_produced: reg.counter(
                "mongodb_cdc_batches_produced_total",
                "Total batches produced",
            ),
            inserts: reg.counter("mongodb_cdc_inserts_total", "Total INSERT events"),
            updates: reg.counter("mongodb_cdc_updates_total", "Total UPDATE events"),
            replaces: reg.counter("mongodb_cdc_replaces_total", "Total REPLACE events"),
            deletes: reg.counter("mongodb_cdc_deletes_total", "Total DELETE events"),
            lifecycle_events: reg.counter(
                "mongodb_cdc_lifecycle_events_total",
                "Total lifecycle events",
            ),
            metadata_events_skipped: reg.counter(
                "mongodb_cdc_metadata_events_skipped_total",
                "Collection metadata events skipped by document replication",
            ),
            snapshot_documents: reg.counter(
                "mongodb_cdc_snapshot_documents_total",
                "Documents emitted by the initial snapshot",
            ),
            snapshot_in_progress: reg.gauge(
                "mongodb_cdc_snapshot_in_progress",
                "1 while the initial snapshot copy is being emitted",
            ),
            reconnects: reg.counter(
                "mongodb_cdc_reconnects_total",
                "Total reconnection attempts",
            ),
        }
    }

    /// Records a received change event by operation type.
    pub fn record_event(&self, operation: ChangeOperation) {
        self.events_received.inc();
        match operation {
            ChangeOperation::Insert => self.inserts.inc(),
            ChangeOperation::Update => self.updates.inc(),
            ChangeOperation::Replace => self.replaces.inc(),
            ChangeOperation::Delete => self.deletes.inc(),
            ChangeOperation::Invalidate
            | ChangeOperation::Drop
            | ChangeOperation::Rename
            | ChangeOperation::DropDatabase
            | ChangeOperation::Metadata
            | ChangeOperation::Unknown => self.lifecycle_events.inc(),
        }
    }

    /// Records bytes received from the change stream.
    pub fn record_bytes(&self, bytes: u64) {
        self.bytes_received.inc_by(bytes);
    }

    /// Records an error.
    pub fn record_error(&self) {
        self.errors.inc();
    }

    /// Records a batch produced for downstream.
    pub fn record_batch(&self) {
        self.batches_produced.inc();
    }

    /// Records a reconnection attempt.
    pub fn record_reconnect(&self) {
        self.reconnects.inc();
    }
}

impl Default for MongoDbCdcMetrics {
    fn default() -> Self {
        Self::new(None)
    }
}

/// Metrics for the `MongoDB` sink connector.
#[derive(Debug, Clone)]
pub struct MongoDbSinkMetrics {
    /// Total records written.
    pub records_written: IntCounter,
    /// Total bytes written.
    pub bytes_written: IntCounter,
    /// Total errors encountered.
    pub errors: IntCounter,
    /// Total batches flushed.
    pub batches_flushed: IntCounter,
    /// Total bulk write operations issued.
    pub bulk_writes: IntCounter,
    /// Total individual insert operations.
    pub inserts: IntCounter,
    /// Total upsert operations.
    pub upserts: IntCounter,
    /// Total delete operations.
    pub deletes: IntCounter,
}

impl MongoDbSinkMetrics {
    /// Creates a new metrics instance with all counters at zero.
    #[must_use]
    #[allow(clippy::missing_panics_doc)]
    pub fn new(registry: Option<&Registry>) -> Self {
        let mut local = None;
        let reg = reg_or_local(registry, &mut local);

        Self {
            records_written: reg.counter(
                "mongodb_sink_records_written_total",
                "Total MongoDB sink records written",
            ),
            bytes_written: reg.counter(
                "mongodb_sink_bytes_written_total",
                "Total MongoDB sink bytes written",
            ),
            errors: reg.counter("mongodb_sink_errors_total", "Total MongoDB sink errors"),
            batches_flushed: reg.counter(
                "mongodb_sink_batches_flushed_total",
                "Total batches flushed",
            ),
            bulk_writes: reg.counter(
                "mongodb_sink_bulk_writes_total",
                "Total bulk write operations",
            ),
            inserts: reg.counter("mongodb_sink_inserts_total", "Total insert operations"),
            upserts: reg.counter("mongodb_sink_upserts_total", "Total upsert operations"),
            deletes: reg.counter("mongodb_sink_deletes_total", "Total delete operations"),
        }
    }

    /// Records a successful batch flush.
    pub fn record_flush(&self, records: u64, bytes: u64) {
        self.records_written.inc_by(records);
        self.bytes_written.inc_by(bytes);
        self.batches_flushed.inc();
    }

    /// Records a bulk write operation.
    pub fn record_bulk_write(&self) {
        self.bulk_writes.inc();
    }

    /// Records an error.
    pub fn record_error(&self) {
        self.errors.inc();
    }

    /// Records insert operations.
    pub fn record_inserts(&self, count: u64) {
        self.inserts.inc_by(count);
    }

    /// Records upsert operations.
    pub fn record_upserts(&self, count: u64) {
        self.upserts.inc_by(count);
    }

    /// Records delete operations.
    pub fn record_deletes(&self, count: u64) {
        self.deletes.inc_by(count);
    }
}

impl Default for MongoDbSinkMetrics {
    fn default() -> Self {
        Self::new(None)
    }
}

#[cfg(test)]
mod tests;