Metrics

This page lists every built-in metric in Carbon 2.0.0, the primitives in carbon_core::metrics, and the two exporter crates. For how metrics fit into a pipeline, see Metrics; to scrape them, see Prometheus.

How metrics are collected#

• Every metric is a static Counter, Gauge or Histogram registered in one process-wide MetricsRegistry.
• Metrics have no labels. Each crate prefixes its own names instead (carbon_, yellowstone_grpc_, transaction_crawler_ and so on).
• Pipeline metrics are registered by PipelineBuilder::build. With the carbon-core postgres feature enabled, build also registers the Postgres metrics, whether or not you use a Postgres processor.
• Datasource metrics are registered when the datasource's consume starts, so they appear in snapshots only after Pipeline::run has started that datasource.
• Pipeline::run itself exports a snapshot only once, at shutdown. Periodic output comes from the exporter: LogMetrics runs its own timer, and PrometheusMetrics renders the registry on each scrape.

Built-in metrics#

Histogram buckets are listed as their upper bounds. Each histogram also has an implicit overflow bucket (+Inf).

carbon-core pipeline#

Defined in pipeline.rs. Updated by Pipeline::run once per update it receives.

carbon-core postgres feature#

Defined in postgres/processors.rs. Updated by PostgresAccountProcessor, PostgresJsonAccountProcessor, PostgresInstructionProcessor and PostgresJsonInstructionProcessor (see the Postgres reference). The counter names do not end in _total.

Datasource histogram buckets#

Most datasource histograms use one of three bucket sets:

carbon-yellowstone-grpc-datasource#

See Yellowstone gRPC. Updates are queued with a non-blocking try_send. When the pipeline channel is full, the update is dropped and an error is logged.

carbon-helius-laserstream-datasource#

See Helius LaserStream. Same behaviour as Yellowstone, including try_send.

carbon-stream-message-datasource#

See Stream message. Names use the agave_grpc_ prefix. Sends wait for channel capacity, so updates are not dropped when the pipeline is slow, and the process-time histograms include that wait.

carbon-helius-atlas-ws-datasource#

See Helius Atlas WebSocket. Counters are incremented before the non-blocking try_send, so they include dropped updates.

carbon-rpc-block-subscribe-datasource#

See RPC block subscribe. Failed transactions (non-Ok status) are skipped and not counted. If try_send fails, the remaining transactions of that block are dropped.

carbon-rpc-block-crawler-datasource#

See RPC block crawler. Failed transactions are skipped and not counted. If try_send fails, the remaining transactions of that block are dropped.

carbon-rpc-transaction-crawler-datasource#

See RPC transaction crawler.

carbon-rpc-gpa-datasource#

See RPC getProgramAccounts.

carbon-rpc-program-subscribe-datasource#

See RPC program subscribe. If try_send fails, the subscription loop exits.

carbon-helius-gpa-v2-datasource#

See Helius getProgramAccountsV2.

carbon-helius-gtfa-datasource#

See Helius getTransactionsForAddress.

carbon-validator-snapshot-datasource#

See Validator snapshot. Histogram buckets: 1, 5, 10, 25, 50, 100, 250, 500, 1000, 5000, 10000 ms. A snapshot load usually takes longer than 10 s, so expect it in the +Inf bucket.

carbon-jetstreamer-datasource (Carbon 1.x)#

carbon-jetstreamer-datasource is published at 1.0.0 on Carbon 1.x and is not compatible with carbon-core 2.0.0. Listed for completeness.

carbon-jito-shredstream-grpc-datasource (Carbon 1.x)#

Also published at 1.0.0 on Carbon 1.x only.

Metric#

Name and help text shared by Counter, Gauge and Histogram.

pub trait Metric: Send + Sync {
    fn name(&self) -> &'static str;
    fn help(&self) -> &'static str { "" }
}

Details. Implemented by the three primitives. The registry stores the concrete types, not dyn Metric, so implementing Metric on your own type does not make it exportable.

Example.

use carbon_core::metrics::{Counter, Metric};

static SEEN: Counter = Counter::new("my_seen_total", "Things seen");
assert_eq!(SEEN.name(), "my_seen_total");

See also. Counter, docs.rs.

Counter#

A monotonically increasing u64.

pub struct Counter { /* private */ }

impl Counter {
    pub const fn new(name: &'static str, help: &'static str) -> Self;
    pub fn inc(&self);
    pub fn inc_by(&self, v: u64);
    pub fn get(&self) -> u64;
}

Details. new is const, so you can declare a counter as a plain static. inc and inc_by are a single relaxed atomic add, cheap enough for the hot path. A counter is exported only after you pass it to MetricsRegistry::register_counter.

Example. See Register custom metrics.

See also. Gauge, docs.rs.

Gauge#

An f64 value that can go up and down.

pub struct Gauge { /* private */ }

impl Gauge {
    pub const fn new(name: &'static str, help: &'static str) -> Self;
    pub fn set(&self, value: f64);
    pub fn add(&self, delta: f64);
    pub fn get(&self) -> f64;
}

Details. Stored as the bit pattern of an f64 in an AtomicU64. Starts at 0.0. add accepts negative deltas.

Example. See Register custom metrics.

See also. Counter, docs.rs.

Histogram#

A bucketed distribution of f64 observations with fixed upper bounds.

pub struct Histogram { /* private */ }

impl Histogram {
    pub fn new(name: &'static str, help: &'static str, boundaries: Vec<f64>) -> Self;
    pub fn record(&self, value: f64);
    pub fn get(&self) -> HistogramSnapshot;
}

Details. boundaries are inclusive upper bounds and must be sorted ascending. The histogram keeps boundaries.len() + 1 buckets: a value goes into the first bucket whose bound is >= the value, and values above the last bound go into the final overflow bucket. record also adds the value to a running sum. Bucket counts are per bucket, not cumulative; PrometheusMetrics makes them cumulative when it renders them.

new allocates, so it is not const. Wrap a static histogram in std::sync::LazyLock.

Example. See Register custom metrics.

See also. HistogramSnapshot, docs.rs.

HistogramSnapshot#

Point-in-time copy of a histogram.

pub struct HistogramSnapshot {
    pub counts: Vec<u64>,
    pub boundaries: Vec<f64>,
    pub sum: f64,
}

Details.

The total count is counts.iter().sum::<u64>().

Example.

let h = SWAP_SIZE.get();
let total: u64 = h.counts.iter().sum();
println!("{} observations, mean {}", total, h.sum / total.max(1) as f64);

See also. MetricsSnapshot, docs.rs.

MetricsSnapshot#

Point-in-time values of every registered metric, passed to MetricsExporter::export.

pub struct MetricsSnapshot {
    pub counters: Vec<(&'static str, &'static str, u64)>,
    pub gauges: Vec<(&'static str, &'static str, f64)>,
    pub histograms: Vec<(&'static str, &'static str, HistogramSnapshot)>,
}

Details. Each tuple is (name, help, value). Entries are in registration order. A metric registered twice appears twice.

Example.

let snapshot = carbon_core::metrics::MetricsRegistry::global().snapshot();
let received = snapshot
    .counters
    .iter()
    .find(|(name, _, _)| *name == "carbon_updates_received_total")
    .map(|(_, _, v)| *v)
    .unwrap_or(0);

See also. MetricsRegistry, docs.rs.

MetricsRegistry#

The process-wide registry that exporters read from.

pub struct MetricsRegistry { /* private */ }

impl MetricsRegistry {
    pub fn global() -> &'static MetricsRegistry;
    pub fn register_counter(&self, counter: &'static Counter);
    pub fn register_gauge(&self, gauge: &'static Gauge);
    pub fn register_histogram(&self, histogram: &'static Histogram);
    pub fn snapshot(&self) -> MetricsSnapshot;
}

Details. There is one registry per process, returned by global(); you cannot create another. It holds &'static references, so metrics must live in statics. Registration appends without deduplication (see the warning above). There is no way to unregister a metric. The internal locks panic with metrics counters lock poisoned (or gauges, histograms) if a thread panicked while holding them.

Example. See Register custom metrics.

See also. MetricsSnapshot, docs.rs.

Register custom metrics#

Declare metrics as statics, register them once at startup (before Pipeline::run), and update them from your processors. Every exporter picks them up with the built-in metrics.

use {
    carbon_core::metrics::{Counter, Gauge, Histogram, MetricsRegistry},
    std::sync::LazyLock,
};

static SWAPS_SEEN: Counter = Counter::new("my_swaps_seen_total", "Swaps decoded by my processor");
static LAST_SLOT: Gauge = Gauge::new("my_last_slot", "Slot of the last processed swap");
static SWAP_SIZE: LazyLock<Histogram> = LazyLock::new(|| {
    Histogram::new(
        "my_swap_amount_in",
        "Swap input amount in lamports",
        vec![1_000.0, 1_000_000.0, 1_000_000_000.0],
    )
});

fn main() {
    let registry = MetricsRegistry::global();
    registry.register_counter(&SWAPS_SEEN);
    registry.register_gauge(&LAST_SLOT);
    registry.register_histogram(&SWAP_SIZE);

    SWAPS_SEEN.inc();
    SWAPS_SEEN.inc_by(2);
    LAST_SLOT.set(312_000_000.0);
    SWAP_SIZE.record(500.0);
    SWAP_SIZE.record(2_500_000.0);

    let snapshot = registry.snapshot();
    for (name, _help, value) in &snapshot.counters {
        println!("counter {name} = {value}");
    }
    for (name, _help, value) in &snapshot.gauges {
        println!("gauge {name} = {value}");
    }
    for (name, _help, h) in &snapshot.histograms {
        println!("histogram {name}: counts={:?} boundaries={:?} sum={}", h.counts, h.boundaries, h.sum);
    }
}

counter my_swaps_seen_total = 3
gauge my_last_slot = 312000000
histogram my_swap_amount_in: counts=[1, 0, 1, 0] boundaries=[1000.0, 1000000.0, 1000000000.0] sum=2500500

This program needs only carbon-core = "2.0.0".

MetricsExporter#

A sink for metrics snapshots, registered with PipelineBuilder::metrics.

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

Details. Pipeline::run calls these methods at fixed points:

The final export and shutdown run on every shutdown path: Ctrl-C, cancellation through datasource_cancellation_token, and all datasources finishing. An Err from either makes run return early with that error.

Example. An exporter that prints counters every 10 seconds and once more at shutdown:

use {
    carbon_core::{
        error::CarbonResult,
        metrics::{MetricsExporter, MetricsRegistry, MetricsSnapshot},
    },
    std::{sync::Arc, time::Duration},
};

/// Prints every counter to stdout every 10 seconds and once more at shutdown.
struct StdoutExporter;

impl MetricsExporter for StdoutExporter {
    fn initialize(self: Arc<Self>) -> CarbonResult<()> {
        // The pipeline only calls `export` at shutdown, so run your own loop
        // for periodic exports.
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(Duration::from_secs(10));
            loop {
                interval.tick().await;
                let _ = self.export(&MetricsRegistry::global().snapshot());
            }
        });
        Ok(())
    }

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

Register it with .metrics(Arc::new(StdoutExporter)).

See also. LogMetrics, PrometheusMetrics, Metrics, docs.rs.

LogMetrics#

An exporter that writes a summary and every metric to the log crate at info level. Crate: carbon-log-metrics = "2.0.0".

pub struct LogMetrics { /* private */ }

impl LogMetrics {
    pub fn new() -> Self;
    pub fn new_with_flush_interval(interval_secs: u64) -> Self;
}

impl Default for LogMetrics { /* same as new() */ }
impl MetricsExporter for LogMetrics { /* … */ }

Details.

• initialize spawns a Tokio task that logs a snapshot every interval. The first tick fires immediately, so you see one block at startup.
• The pipeline's shutdown export logs one last block, then shutdown cancels the task.
• Output goes through log::info! with target carbon_log_metrics. Nothing is printed unless you install a logger (for example env_logger) with info enabled for that target.
• Each block starts with a summary line: HH:MM:SS (+time since last flush) | N processed (P%), N successful, N failed (F%), Q in queue. The processed percentage is carbon_updates_processed_total divided by carbon_updates_received_total plus carbon_updates_queued; the failed percentage is relative to processed.
• Then one line per counter and gauge ( name: value) and one per histogram ( name: count=N; buckets and sum are not logged).

Example. This program runs a pipeline with a datasource that emits three empty blocks and then returns, which closes the channel and ends the pipeline.

use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{BlockDetails, Datasource, DatasourceId, Update, UpdateType},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_log_metrics::LogMetrics,
    std::sync::Arc,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Emits three empty blocks, then returns, which ends the pipeline.
struct ThreeBlocks;

#[async_trait]
impl Datasource for ThreeBlocks {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        _cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        for slot in 1..=3 {
            let block = BlockDetails {
                slot,
                block_hash: None,
                previous_block_hash: None,
                rewards: None,
                num_reward_partitions: None,
                block_time: None,
                block_height: None,
            };
            let _ = sender.send((Update::BlockDetails(block), id.clone())).await;
        }
        Ok(())
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::BlockDetails]
    }
}

struct Noop;

impl Processor<BlockDetails> for Noop {
    async fn process(&mut self, _data: &BlockDetails) -> CarbonResult<()> {
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();

    Pipeline::builder()
        .datasource(ThreeBlocks)
        .block_details(Noop)
        .metrics(Arc::new(LogMetrics::new()))
        .build()?
        .run()
        .await
}

[dependencies]
async-trait = "0.1.89"
carbon-core = "2.0.0"
carbon-log-metrics = "2.0.0"
env_logger = "0.11.8"
tokio = { version = "1.48.0", features = ["full"] }
tokio-util = "0.7.17"

cargo run

[2026-11-08T23:53:11Z INFO  carbon_core::pipeline] starting pipeline. num_datasources: 1, num_exporters: 1, num_account_pipes: 0, num_account_deletion_pipes: 0, num_instruction_pipes: 0, num_transaction_pipes: 0
[2026-11-08T23:53:11Z INFO  carbon_core::pipeline] update_receiver closed, shutting down.
[2026-11-08T23:53:11Z INFO  carbon_log_metrics] 00:00:00 (+4.145248ms) | 3 processed (100%), 3 successful, 0 failed (0%), 0 in queue
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_received_total: 3
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_processed_total: 3
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_successful_total: 3
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_failed_total: 0
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_account_updates_processed_total: 0
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_transaction_updates_processed_total: 0
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_account_deletions_processed_total: 0
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_block_details_processed_total: 3
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_queued: 0
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_process_time_nanoseconds: count=3
[2026-11-08T23:53:11Z INFO  carbon_log_metrics]   carbon_updates_process_time_milliseconds: count=3
[2026-11-08T23:53:11Z INFO  carbon_core::pipeline] pipeline shutdown complete.

The pipeline ended before the first periodic tick ran, so only the shutdown export appears. With the carbon-core postgres feature enabled, the six postgres_* metrics are listed too.

See also. MetricsExporter, docs.rs.

PrometheusMetrics#

An exporter that serves the registry in the Prometheus text format over HTTP. Crate: carbon-prometheus-metrics = "2.0.0".

pub struct PrometheusMetrics { /* private */ }

impl PrometheusMetrics {
    pub fn new() -> Self;
    pub fn with_server(config: PrometheusServerConfig) -> Self;              // feature "http-server"
    pub fn start_server(
        config: PrometheusServerConfig,
    ) -> tokio::task::JoinHandle<Result<(), Box<dyn std::error::Error + Send + Sync>>>; // feature "http-server"
    pub fn prometheus_text() -> String;
    pub fn format_snapshot(snapshot: &MetricsSnapshot) -> String;
}

impl Default for PrometheusMetrics { /* same as new() */ }
impl MetricsExporter for PrometheusMetrics { /* … */ }

Details.

The server:

• Serves GET /metrics only, with content-type: text/plain; charset=utf-8; version=0.0.4 and cache-control: no-cache, no-store, must-revalidate.
• Applies a request timeout (408 Request Timeout after request_timeout_secs) and, when enable_cors is set, a permissive CORS layer.
• Logs Prometheus metrics server starting on http://<addr>/metrics at startup.

Rendering:

• Characters other than ASCII letters, digits, _ and : in names are replaced with _. All built-in names are already valid.
• # HELP is written when the help text is non-empty, then # TYPE (counter, gauge or histogram).
• Histograms are written as cumulative <name>_bucket{le="…"} lines, a le="+Inf" bucket, <name>_count and <name>_sum. There are no other labels.

Example. Serve metrics on 127.0.0.1:9464 for the lifetime of the pipeline. The datasource emits three blocks and then waits for Ctrl-C.

use {
    async_trait::async_trait,
    carbon_core::{
        datasource::{BlockDetails, Datasource, DatasourceId, Update, UpdateType},
        error::CarbonResult,
        pipeline::Pipeline,
        processor::Processor,
    },
    carbon_prometheus_metrics::{PrometheusMetrics, PrometheusServerConfig},
    std::sync::Arc,
    tokio::sync::mpsc::Sender,
    tokio_util::sync::CancellationToken,
};

/// Emits three empty blocks, then waits for shutdown.
struct ThreeBlocks;

#[async_trait]
impl Datasource for ThreeBlocks {
    async fn consume(
        &self,
        id: DatasourceId,
        sender: Sender<(Update, DatasourceId)>,
        cancellation_token: CancellationToken,
    ) -> CarbonResult<()> {
        for slot in 1..=3 {
            let block = BlockDetails {
                slot,
                block_hash: None,
                previous_block_hash: None,
                rewards: None,
                num_reward_partitions: None,
                block_time: None,
                block_height: None,
            };
            let _ = sender.send((Update::BlockDetails(block), id.clone())).await;
        }
        cancellation_token.cancelled().await;
        Ok(())
    }

    fn update_types(&self) -> Vec<UpdateType> {
        vec![UpdateType::BlockDetails]
    }
}

struct Noop;

impl Processor<BlockDetails> for Noop {
    async fn process(&mut self, _data: &BlockDetails) -> CarbonResult<()> {
        Ok(())
    }
}

#[tokio::main]
async fn main() -> CarbonResult<()> {
    env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init();

    let config = PrometheusServerConfig::new().listen_addr("127.0.0.1:9464".parse().unwrap());

    Pipeline::builder()
        .datasource(ThreeBlocks)
        .block_details(Noop)
        .metrics(Arc::new(PrometheusMetrics::with_server(config)))
        .build()?
        .run()
        .await
}

Add carbon-prometheus-metrics = "2.0.0" to the dependencies from the LogMetrics example. Start it with cargo run, then scrape it from another terminal:

curl -s http://127.0.0.1:9464/metrics

# HELP carbon_updates_received_total Total updates pulled from datasources
# TYPE carbon_updates_received_total counter
carbon_updates_received_total 3
# HELP carbon_updates_processed_total Total updates processed by the pipeline
# TYPE carbon_updates_processed_total counter
carbon_updates_processed_total 3
…
# HELP carbon_updates_queued Current number of updates waiting in queue
# TYPE carbon_updates_queued gauge
carbon_updates_queued 0
# HELP carbon_updates_process_time_nanoseconds Time taken to process updates in nanoseconds
# TYPE carbon_updates_process_time_nanoseconds histogram
carbon_updates_process_time_nanoseconds_bucket{le="1000"} 2
carbon_updates_process_time_nanoseconds_bucket{le="10000"} 3
carbon_updates_process_time_nanoseconds_bucket{le="100000"} 3
carbon_updates_process_time_nanoseconds_bucket{le="1000000"} 3
carbon_updates_process_time_nanoseconds_bucket{le="10000000"} 3
carbon_updates_process_time_nanoseconds_bucket{le="100000000"} 3
carbon_updates_process_time_nanoseconds_bucket{le="1000000000"} 3
carbon_updates_process_time_nanoseconds_bucket{le="+Inf"} 3
carbon_updates_process_time_nanoseconds_count 3
carbon_updates_process_time_nanoseconds_sum 7030
…

Press Ctrl-C to stop both the pipeline and the server.

See also. PrometheusServerConfig, Prometheus, docs.rs.

PrometheusServerConfig#

Configuration for the /metrics HTTP server. Requires the http-server feature (on by default).

#[derive(Debug, Clone)]
pub struct PrometheusServerConfig {
    pub listen_addr: std::net::SocketAddr,
    pub enable_cors: bool,
    pub request_timeout_secs: u64,
}

impl PrometheusServerConfig {
    pub fn new() -> Self;
    pub fn listen_addr(self, addr: std::net::SocketAddr) -> Self;
    pub fn enable_cors(self, enable: bool) -> Self;
    pub fn request_timeout_secs(self, timeout: u64) -> Self;
}

Details. new() and Default give these values; each builder method replaces one.

Example. Serve metrics yourself, without a pipeline exporter:

use carbon_prometheus_metrics::{PrometheusMetrics, PrometheusServerConfig};

#[tokio::main]
async fn main() {
    // Serve /metrics yourself instead of through the exporter.
    let server = PrometheusMetrics::start_server(
        PrometheusServerConfig::new()
            .listen_addr("0.0.0.0:9100".parse().unwrap())
            .enable_cors(false)
            .request_timeout_secs(10),
    );

    // Or render the exposition text directly, e.g. from your own HTTP handler.
    let text: String = PrometheusMetrics::prometheus_text();
    println!("{text}");

    server.abort();
}

See also. run_metrics_server, docs.rs.

run_metrics_server#

Free functions that run the /metrics server on the current task until Ctrl-C. Require the http-server feature.

pub async fn run_metrics_server(
    listen_addr: std::net::SocketAddr,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;

pub async fn run_default_metrics_server() -> Result<(), Box<dyn std::error::Error + Send + Sync>>;

pub async fn run_metrics_server_with_config(
    config: PrometheusServerConfig,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>>;

Details. run_metrics_server uses listen_addr with the other defaults; run_default_metrics_server uses PrometheusServerConfig::default() (0.0.0.0:9464). They return Err only if binding fails. A server error after startup is logged and the function returns Ok(()). Each function also returns Ok(()) after Ctrl-C. PrometheusMetrics::with_server and start_server call run_metrics_server_with_config inside tokio::spawn.

Example.

tokio::spawn(carbon_prometheus_metrics::run_metrics_server(
    "127.0.0.1:9464".parse().unwrap(),
));

See also. PrometheusMetrics, docs.rs.