Metrics | Carbon docs

Carbon's metrics model (Counter, Gauge, Histogram, MetricsRegistry, MetricsExporter), LogMetrics versus Prometheus, and registering your own metrics.

Carbon counts what flows through a pipeline: updates received, processed and failed, queue depth and processing time, plus each datasource's own numbers. You add your own metrics the same way and choose an exporter to publish them. The metrics reference lists every built-in metric name.

The model#

Metrics live in one process-wide registry, MetricsRegistry::global(). The registry holds &'static references to three kinds of metric, all lock-free:

Type	Constructor	Methods	Use for
`Counter`	`const fn new(name, help)`	`inc()`, `inc_by(u64)`, `get()`	Totals that only go up: updates, errors, rows written.
`Gauge`	`const fn new(name, help)`	`set(f64)`, `add(f64)`, `get()`	Values that go up and down: queue depth, last slot seen.
`Histogram`	`fn new(name, help, boundaries: Vec<f64>)`	`record(f64)`, `get()`	Distributions: latencies, sizes. Counts values per bucket and keeps a running sum.

Metrics have a name and a help string, and no labels. Updating one is a single relaxed atomic operation, so it's cheap enough for the hot path.

An exporter publishes the registry. Exporters implement MetricsExporter:

Rust

pub trait MetricsExporter: Send + Sync {
    fn initialize(self: Arc<Self>) -> CarbonResult<()> { Ok(()) }
    fn export(&self, snapshot: &MetricsSnapshot) -> CarbonResult<()>;
    fn shutdown(&self) -> CarbonResult<()> { Ok(()) }
}

MetricsRegistry::global().snapshot() returns a MetricsSnapshot with the current value of every registered metric: counters and gauges as (name, help, value) tuples, and histograms with their bucket counts, boundaries and sum.

How the pipeline uses metrics#

build() registers the pipeline's metrics: carbon_updates_received_total, carbon_updates_processed_total, carbon_updates_successful_total, carbon_updates_failed_total, carbon_updates_queued, two processing-time histograms and one counter per update type. Datasources register theirs when they start, such as the transaction crawler's transaction_crawler_* metrics.
run() calls initialize on each exporter you added with .metrics(…), updates the pipeline metrics on every update, and calls export with a final snapshot followed by shutdown when it stops.
The pipeline doesn't export periodically. Each exporter decides when to publish.

Because the registry is global, every pipeline in a process shares the same metrics.

Exporters#

Log metrics#

carbon-log-metrics writes the registry to the log crate at info level: a summary line, then every metric.

Rust

.metrics(Arc::new(LogMetrics::new()))                       // every 5 seconds
.metrics(Arc::new(LogMetrics::new_with_flush_interval(30))) // every 30 seconds

initialize starts a task that exports on that interval; shutdown stops it. You need a logger, such as env_logger with RUST_LOG=info, to see the output. Use it in development and for small jobs.

Prometheus#

carbon-prometheus-metrics serves the registry in the Prometheus text format:

Rust

use carbon_prometheus_metrics::{PrometheusMetrics, PrometheusServerConfig};

.metrics(Arc::new(PrometheusMetrics::with_server(PrometheusServerConfig::default())))

with_server starts an HTTP server during initialize, on 0.0.0.0:9464 by default, with the metrics at /metrics. Each scrape reads the registry directly, so values are always current and export does nothing. PrometheusMetrics::new() doesn't start a server; pair it with PrometheusMetrics::start_server(config) or call PrometheusMetrics::prometheus_text() to serve the text yourself. See Monitor with Prometheus.

Your own exporter#

Implement MetricsExporter to send metrics elsewhere. Spawn a periodic task in initialize if you want regular exports, as LogMetrics does.

Rust

pub struct StdoutExporter;

impl MetricsExporter for StdoutExporter {
    fn export(&self, snapshot: &MetricsSnapshot) -> CarbonResult<()> {
        for (name, _help, value) in &snapshot.counters {
            println!("{name} {value}");
        }
        Ok(())
    }
}

You can add several exporters to one pipeline.

Register your own metrics#

Declare each metric as a static. Counter and Gauge have const constructors; wrap a Histogram in LazyLock.
Register each one once, before the pipeline runs, with MetricsRegistry::global().register_counter(&METRIC) (or register_gauge, register_histogram). Registering a metric twice makes it appear twice in snapshots.
Update it anywhere: in processors, filters or your own tasks.

Prefix names with your application's name so they don't collide with Carbon's.

Example#

This pipeline records two custom metrics in its processor and logs every metric when it finishes. Three hand-built account updates stand in for a datasource.

Terminal

cargo add carbon-core@2.0.0 carbon-log-metrics@2.0.0 carbon-stream-message-datasource@2.0.0
cargo add solana-account@4.3.2 solana-pubkey@4.2.1
cargo add tokio --features full
cargo add env_logger

src/main.rs

use {
    carbon_core::{
        account::{AccountDecoder, AccountProcessorInputType, DecodedAccount},
        datasource::AccountUpdate,
        error::CarbonResult,
        metrics::{Counter, Histogram, MetricsRegistry},
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    carbon_stream_message_datasource::{StreamMessageClient, UnifiedMessage},
    solana_account::Account,
    solana_pubkey::Pubkey,
    std::sync::{Arc, LazyLock},
};

static LARGE_ACCOUNTS: Counter =
    Counter::new("myapp_large_accounts_total", "Accounts holding at least 1 SOL");

static ACCOUNT_SOL: LazyLock<Histogram> = LazyLock::new(|| {
    Histogram::new("myapp_account_sol", "Account balances in SOL", vec![0.1, 1.0, 10.0])
});

fn register_metrics() {
    let registry = MetricsRegistry::global();
    registry.register_counter(&LARGE_ACCOUNTS);
    registry.register_histogram(&ACCOUNT_SOL);
}

struct AnyAccount;

impl<'a> AccountDecoder<'a> for AnyAccount {
    type AccountType = ();

    fn decode_account(&self, account: &'a Account) -> Option<DecodedAccount<()>> {
        Some(DecodedAccount {
            lamports: account.lamports,
            data: (),
            owner: account.owner,
            executable: account.executable,
            rent_epoch: account.rent_epoch,
        })
    }
}

struct Measure;

impl Processor<AccountProcessorInputType<'_, ()>> for Measure {
    async fn process(&mut self, input: &AccountProcessorInputType<'_, ()>) -> CarbonResult<()> {
        let sol = input.raw_account.lamports as f64 / 1e9;
        ACCOUNT_SOL.record(sol);
        if sol >= 1.0 {
            LARGE_ACCOUNTS.inc();
        }
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::init();
    register_metrics();

    let (tx, rx) = tokio::sync::mpsc::channel(16);
    for lamports in [50_000_000, 2_000_000_000, 30_000_000_000] {
        let update = AccountUpdate {
            pubkey: Pubkey::new_unique(),
            account: Account { lamports, ..Account::default() },
            slot: 1,
            transaction_signature: None,
        };
        tx.send(UnifiedMessage::Account(update)).await.unwrap();
    }
    drop(tx);

    Pipeline::builder()
        .datasource(StreamMessageClient::new(rx))
        .account(AnyAccount, Measure)
        .metrics(Arc::new(LogMetrics::new()))
        .build()?
        .run()
        .await
}

Terminal

RUST_LOG=info cargo run

Output

[2026-11-09T00:08:19Z INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 1, num_account_pipes: 1, num_account_deletion_pipes: 0, num_instruction_pipes: 0, num_transaction_pipes: 0
[2026-11-09T00:08:19Z WARN  carbon_stream_message_datasource] Receiver closed
[2026-11-09T00:08:19Z INFO  carbon_stream_message_datasource] Geyser message loop exited
[2026-11-09T00:08:19Z INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[2026-11-09T00:08:19Z INFO  carbon_log_metrics] 00:00:00 (+616.231µs) | 3 processed (100%), 3 successful, 0 failed (0%), 0 in queue
[2026-11-09T00:08:19Z INFO  carbon_log_metrics]   myapp_large_accounts_total: 2
[2026-11-09T00:08:19Z INFO  carbon_log_metrics]   carbon_updates_received_total: 3
[2026-11-09T00:08:19Z INFO  carbon_log_metrics]   carbon_updates_processed_total: 3
[2026-11-09T00:08:19Z INFO  carbon_log_metrics]   carbon_updates_successful_total: 3
[2026-11-09T00:08:19Z INFO  carbon_log_metrics]   carbon_updates_failed_total: 0
…
[2026-11-09T00:08:19Z INFO  carbon_log_metrics]   myapp_account_sol: count=3
…
[2026-11-09T00:08:19Z INFO  carbon_core::pipeline] pipeline shutdown complete.

Two of the three accounts hold at least 1 SOL. The WARN … Receiver closed line is the stream message datasource noticing that its input channel closed, which is what ends this finite example. LogMetrics prints only the sample count of each histogram; use Prometheus to see the buckets.

Next steps#

Metrics reference

Every built-in metric and what it measures.

Monitor with Prometheus

Scrape a pipeline and build dashboards and alerts.

Run in production

What to watch and how to size the pipeline.