Struct StorageStatistics

pub struct StorageStatistics<Stats, Metrics, Meta> {
    stats: Rc<RefCell<StatsInner<Stats, Metrics>>>,
    meta: Meta,
}

A helper struct designed to make it easy for operators to update user-facing metrics. This struct also ensures that each stack is also incremented in prometheus.

Caveats:

• There is one Prometheus timeseries-per-worker, and we label it with the source id and the source id of the parent source (if there is no parent, these labels have the same value).
  • Some metrics also have the shard id we are writing metrics for.
• The prometheus metrics do not have the same timestamps as the ones exposed in sql, because they are written at different times.
  • This may be fixed in the future when we write the metrics from storaged directly.
  • The value also eventually converge to the same value.

Fields

stats: Rc<RefCell<StatsInner<Stats, Metrics>>>

meta: Meta

Meta data needed to maintain statistics.

Implementations

impl StorageStatistics<SourceStatisticsRecord, SourceStatisticsMetrics, SourceStatisticsMetadata>

pub(crate) fn new( id: GlobalId, worker_id: usize, metrics: &SourceStatisticsMetricDefs, parent_source_id: GlobalId, shard_id: &ShardId, envelope: SourceEnvelope, resume_upper: Antichain<Timestamp>, ) -> Self

pub fn reset_gauges(&self)

Reset gauges to uninitialized state. This does not reset prometheus gauges, which will be overwritten with new values.

pub fn snapshot(&self) -> Option<SourceStatisticsRecord>

Get a snapshot of the data, returning None if all gauges are not initialized.

This also resets counters, so that we continue to move diffs around.

pub fn initialize_snapshot_committed(&self, upper: &Antichain<Timestamp>)

Set the snapshot_committed stat based on the reported upper, and mark the stats as initialized.

• In sql, we ensure that we never reset snapshot_committed to false, but gauges and counters are ordinarily reset to 0 in Prometheus, so on restarts this value may be inconsistent.

pub fn update_snapshot_committed(&self, upper: &Antichain<Timestamp>)

Set the snapshot_committed stat based on the reported upper.

pub fn inc_messages_received_by(&self, value: u64)

Increment the messages_received stat.

pub fn inc_updates_staged_by(&self, value: u64)

Increment the updates stat.

pub fn inc_updates_committed_by(&self, value: u64)

Increment the messages_committed stat.

pub fn inc_bytes_received_by(&self, value: u64)

Increment the bytes_received stat.

pub fn update_bytes_indexed_by(&self, value: i64)

Update the bytes_indexed stat. A positive value will add and a negative value will subtract.

pub fn set_bytes_indexed(&self, value: i64)

Set the bytes_indexed to the given value

pub fn update_records_indexed_by(&self, value: i64)

Update the records_indexed stat. A positive value will add and a negative value will subtract.

pub fn set_records_indexed(&self, value: i64)

Set the records_indexed to the given value

pub fn initialize_rehydration_latency_ms(&self)

Initialize the rehydration_latency_ms stat as NULL.

pub fn update_envelope_state_tombstones_by(&self, value: i64)

Update the envelope_state_tombstones stat. A positive value will add and a negative value will subtract.

pub fn update_rehydration_latency_ms(&self, upper: &Antichain<Timestamp>)

Set the rehydration_latency_ms stat based on the reported upper.

pub fn set_offset_known(&self, value: u64)

Set the offset_known stat to the given value.

pub fn set_offset_committed(&self, value: u64)

Set the offset_committed stat to the given value.

pub fn set_snapshot_records_known(&self, value: u64)

Set the snapshot_records_known stat to the given value.

pub fn set_snapshot_records_staged(&self, value: u64)

Set the snapshot_records_known stat to the given value.

impl StorageStatistics<SinkStatisticsRecord, SinkStatisticsMetrics, ()>

pub(crate) fn new( id: GlobalId, worker_id: usize, metrics: &SinkStatisticsMetricDefs, ) -> Self

pub fn reset_gauges(&self)

Reset gauges to uninitialized state. This does not reset prometheus gauges, which will be overwritten with new values.

pub fn snapshot(&self) -> Option<SinkStatisticsRecord>

Get a snapshot of the data, returning None if all gauges are not initialized.

This also resets counters, so that we continue to move diffs around.

pub fn inc_messages_staged_by(&self, value: u64)

Increment the messages_staged stat.

pub fn inc_bytes_staged_by(&self, value: u64)

Increment the bytes_received stat.

pub fn inc_messages_committed_by(&self, value: u64)

Increment the messages_committed stat.

pub fn inc_bytes_committed_by(&self, value: u64)

Increment the bytes_committed stat.

Trait Implementations

impl<Stats, Metrics, Meta: Clone> Clone for StorageStatistics<Stats, Metrics, Meta>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<Stats: Debug, Metrics: Debug, Meta: Debug> Debug for StorageStatistics<Stats, Metrics, Meta>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.