Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
1a41d21
make pull request
uri-99 Oct 10, 2024
2bbcb72
feat: qty received proofs counter on batcher
uri-99 Oct 10, 2024
230c9ba
feat: batcher prometheus
uri-99 Oct 10, 2024
975eff4
Merge branch 'staging' into add-observability-batcher
IAvecilla Oct 11, 2024
e323662
Run metrics server in separated tokio task
IAvecilla Oct 11, 2024
196bdaa
fix: increment the prometheus variables correctly
IAvecilla Oct 11, 2024
833d97f
feat: add visualizations for batcher metrics
IAvecilla Oct 11, 2024
89e2e95
feat: adding new vars and removing unused code
uri-99 Oct 11, 2024
9276612
refactor: rename module from prometheus to metrics to avoid name coll…
IAvecilla Oct 11, 2024
33c6404
refactor: move main file to new bin directory
IAvecilla Oct 11, 2024
4c6fa91
feat: sent and reverted batches metrics
uri-99 Oct 11, 2024
a9c27ab
feat: batcher_started metric
uri-99 Oct 11, 2024
496e92d
remove: responded batches, batcher doesnt wait for verification
uri-99 Oct 11, 2024
b7dcb58
fix: open_connections metric
uri-99 Oct 11, 2024
a6c1008
feat: gas_price metric
uri-99 Oct 11, 2024
448151d
fix: init vars and batcher start with 0
uri-99 Oct 11, 2024
b620f91
feat: update grafana dashboard with new batcher metrics
IAvecilla Oct 11, 2024
f18bca8
chore: cargo fmt
uri-99 Oct 11, 2024
66295e1
Merge branch 'add-observability-batcher' of https://github.com/yetano…
uri-99 Oct 11, 2024
01b7e0e
fix: move back main.rs to the root
IAvecilla Oct 11, 2024
27b9836
Remove unneeded comment
MauroToscano Oct 13, 2024
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 147 additions & 5 deletions batcher/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 4 additions & 0 deletions batcher/aligned-batcher/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -31,3 +31,7 @@ bincode = "1.3.3"
aligned-sdk = { path = "../aligned-sdk" }
ciborium = "=0.2.2"
priority-queue = "2.1.0"

once_cell = "1.20.2"
warp = "0.3.7"
prometheus = { version = "0.13.4", features = ["process"] }
8 changes: 8 additions & 0 deletions batcher/aligned-batcher/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ mod config;
mod connection;
mod eth;
pub mod gnark;
pub mod metrics;
pub mod risc_zero;
pub mod s3;
pub mod sp1;
Expand Down Expand Up @@ -212,6 +213,7 @@ impl Batcher {

// Let's spawn the handling of each connection in a separate task.
while let Ok((stream, addr)) = listener.accept().await {
metrics::OPEN_CONNECTIONS.inc();
let batcher = self.clone();
tokio::spawn(batcher.handle_connection(stream, addr));
}
Expand Down Expand Up @@ -294,6 +296,7 @@ impl Batcher {
Ok(_) => info!("{} disconnected", &addr),
}

metrics::OPEN_CONNECTIONS.dec();
Ok(())
}

Expand All @@ -313,6 +316,7 @@ impl Batcher {
};
let msg_nonce = client_msg.verification_data.nonce;
debug!("Received message with nonce: {msg_nonce:?}",);
metrics::RECEIVED_PROOFS.inc();

// * ---------------------------------------------------*
// * Perform validations over the message *
Expand Down Expand Up @@ -1012,6 +1016,8 @@ impl Batcher {

let proof_submitters = finalized_batch.iter().map(|entry| entry.sender).collect();

metrics::GAS_PRICE_USED_ON_LATEST_BATCH.set(gas_price.as_u64() as i64);

match self
.create_new_task(
*batch_merkle_root,
Expand All @@ -1023,6 +1029,7 @@ impl Batcher {
{
Ok(_) => {
info!("Batch verification task created on Aligned contract");
metrics::SENT_BATCHES.inc();
Ok(())
}
Err(e) => {
Expand All @@ -1031,6 +1038,7 @@ impl Batcher {
e
);

metrics::REVERTED_BATCHES.inc();
Err(e)
}
}
Expand Down
12 changes: 12 additions & 0 deletions batcher/aligned-batcher/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,9 @@ use std::sync::Arc;
use clap::Parser;
use env_logger::Env;

use aligned_batcher::metrics;
use aligned_batcher::{types::errors::BatcherError, Batcher};
use warp::Filter;

/// Batcher main flow:
/// There are two main tasks spawned: `listen_connections` and `listen_new_blocks`
Expand Down Expand Up @@ -38,6 +40,14 @@ async fn main() -> Result<(), BatcherError> {

env_logger::Builder::from_env(Env::default().default_filter_or("info")).init();

// Endpoint for Prometheus
metrics::init_variables();
let metrics_route = warp::path!("metrics").and_then(metrics::metrics_handler);
println!("Starting Batcher metrics on port 9093");
tokio::task::spawn(async move {
warp::serve(metrics_route).run(([0, 0, 0, 0], 9093)).await;
}); //TODO read from config

let batcher = Batcher::new(cli.config).await;
let batcher = Arc::new(batcher);

Expand All @@ -53,6 +63,8 @@ async fn main() -> Result<(), BatcherError> {
}
});

metrics::batcher_started();

batcher.listen_connections(&addr).await?;

Ok(())
Expand Down
67 changes: 67 additions & 0 deletions batcher/aligned-batcher/src/metrics.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
// Prometheus
use prometheus::{opts, register_int_counter, register_int_gauge, IntCounter, IntGauge};

use warp::{Rejection, Reply};

use once_cell::sync::Lazy;
use std::{thread, time};

// Prometheus setup
pub static BATCHER_STARTED: Lazy<IntCounter> =
Lazy::new(|| register_int_counter!(opts!("batcher_started", "Batcher Started")).unwrap());

pub static OPEN_CONNECTIONS: Lazy<IntGauge> =
Lazy::new(|| register_int_gauge!(opts!("open_connections", "Open Connections")).unwrap());

pub static RECEIVED_PROOFS: Lazy<IntCounter> =
Lazy::new(|| register_int_counter!(opts!("received_proofs", "Received Proofs")).unwrap());

pub static SENT_BATCHES: Lazy<IntCounter> =
Lazy::new(|| register_int_counter!(opts!("sent_batches", "Sent Batches")).unwrap());

pub static REVERTED_BATCHES: Lazy<IntCounter> =
Lazy::new(|| register_int_counter!(opts!("reverted_batches", "Reverted Batches")).unwrap());

pub static GAS_PRICE_USED_ON_LATEST_BATCH: Lazy<IntGauge> = Lazy::new(|| {
register_int_gauge!(opts!("gas_price_used_on_latest_batch", "Gas Price")).unwrap()
});

// so Prometheus can collect our metrics.
pub async fn metrics_handler() -> Result<impl Reply, Rejection> {
use prometheus::Encoder;
let encoder = prometheus::TextEncoder::new();

let mut buffer = Vec::new();
if let Err(e) = encoder.encode(&prometheus::gather(), &mut buffer) {
eprintln!("could not encode prometheus metrics: {}", e);
};
let res = match String::from_utf8(buffer.clone()) {
Ok(v) => v,
Err(e) => {
eprintln!("prometheus metrics could not be from_utf8'd: {}", e);
String::default()
}
};
buffer.clear();

Ok(res)
}

pub fn init_variables() {
BATCHER_STARTED.reset();

OPEN_CONNECTIONS.set(0);

RECEIVED_PROOFS.reset();

SENT_BATCHES.reset();

REVERTED_BATCHES.reset();

GAS_PRICE_USED_ON_LATEST_BATCH.set(0);
}

pub fn batcher_started() {
thread::sleep(time::Duration::from_secs(10));
BATCHER_STARTED.inc();
}
Loading