feat: Telemetry for the batcher and improve current tracing

aggregator/internal/pkg/aggregator.go

@@ -272,8 +272,9 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 	agg.logger.Info("Sending aggregated response onchain", "taskIndex", blsAggServiceResp.TaskIndex,
  "batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
 	for i := 0; i < MaxSentTxRetries; i++ {
-_, err = agg.sendAggregatedResponse(batchIdentifierHash, batchData.BatchMerkleRoot, batchData.SenderAddress, nonSignerStakesAndSignature)
+receipt, err := agg.sendAggregatedResponse(batchIdentifierHash, batchData.BatchMerkleRoot, batchData.SenderAddress, nonSignerStakesAndSignature)
  if err == nil {
+ agg.telemetry.TaskSentToEthereum(batchData.BatchMerkleRoot, receipt.TxHash.String())
  agg.logger.Info("Aggregator successfully responded to task",
  "taskIndex", blsAggServiceResp.TaskIndex,
  "batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))

aggregator/internal/pkg/telemetry.go

@@ -30,6 +30,11 @@ type TaskErrorMessage struct {
 	TaskError string `json:"error"`
 }
 
+type TaskSentToEthereumMessage struct {
+	MerkleRoot string `json:"merkle_root"`
+	TxHash string `json:"tx_hash"`
+}
+
 type Telemetry struct {
 	client http.Client
 	baseURL url.URL
@@ -91,6 +96,16 @@ func (t *Telemetry) LogTaskError(batchMerkleRoot [32]byte, taskError error) {
 	}
 }
 
+func (t *Telemetry) TaskSentToEthereum(batchMerkleRoot [32]byte, txHash string) {
+	body := TaskSentToEthereumMessage{
+ MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+ TxHash: txHash,
+	}
+	if err := t.sendTelemetryMessage("/api/aggregatorTaskSent", body); err != nil {
+ t.logger.Error("[Telemetry] Error in TaskSentToEthereum", "error", err)
+	}
+}
+
 func (t *Telemetry) FinishTrace(batchMerkleRoot [32]byte) {
 	// In order to wait for all operator responses, even if the quorum is reached, this function has a delayed execution
 	go func() {

batcher/aligned-batcher/Cargo.toml

@@ -28,6 +28,7 @@ bincode = "1.3.3"
 aligned-sdk = { path = "../aligned-sdk" }
 ciborium = "=0.2.2"
 priority-queue = "2.1.0"
+reqwest = { version = "0.12", features = ["json"] }
 
 once_cell = "1.20.2"
 warp = "0.3.7"

batcher/aligned-batcher/src/config/mod.rs

@@ -42,6 +42,7 @@ pub struct BatcherConfigFromYaml {
  pub max_batch_size: usize,
  pub pre_verification_is_enabled: bool,
  pub metrics_port: u16,
+ pub telemetry_ip_port_address: String,
  pub non_paying: Option<NonPayingConfigFromYaml>,
 }
 

batcher/aligned-batcher/src/lib.rs

@@ -48,6 +48,7 @@ use types::batch_queue::{self, BatchQueueEntry, BatchQueueEntryPriority};
 use types::errors::{BatcherError, BatcherSendError};
 
 use crate::config::{ConfigFromYaml, ContractDeploymentOutput};
+use crate::telemetry::sender::TelemetrySender;
 
 mod config;
 mod connection;
@@ -58,6 +59,7 @@ pub mod retry;
 pub mod risc_zero;
 pub mod s3;
 pub mod sp1;
+pub mod telemetry;
 pub mod types;
 mod zk_utils;
 
@@ -86,6 +88,7 @@ pub struct Batcher {
  posting_batch: Mutex<bool>,
  disabled_verifiers: Mutex<U256>,
  pub metrics: metrics::BatcherMetrics,
+ pub telemetry: TelemetrySender,
 }
 
 impl Batcher {
@@ -212,6 +215,11 @@ impl Batcher {
  }
  .expect("Failed to get disabled verifiers");
 
+ let telemetry = TelemetrySender::new(format!(
+ "http://{}",
+ config.batcher.telemetry_ip_port_address
+ ));
+
  Self {
  s3_client,
  s3_bucket_name,
@@ -235,6 +243,7 @@ impl Batcher {
  batch_state: Mutex::new(batch_state),
  disabled_verifiers: Mutex::new(disabled_verifiers),
  metrics,
+ telemetry,
  }
  }
 
@@ -981,6 +990,14 @@ impl Batcher {
  .map(VerificationCommitmentBatch::hash_data)
  .collect();
 
+ if let Err(e) = self
+ .telemetry
+ .init_task_trace(&hex::encode(batch_merkle_tree.root))
+ .await
+ {
+ error!("Failed to initialize task trace on telemetry: {:?}", e);
+ }
+
  if let Err(e) = self
  .submit_batch(
  &batch_bytes,
@@ -991,6 +1008,14 @@ impl Batcher {
  )
  .await
  {
+ let reason = format!("{:?}", e);
+ if let Err(e) = self
+ .telemetry
+ .task_creation_failed(&hex::encode(batch_merkle_tree.root), &reason)
+ .await
+ {
+ error!("Failed to send task status to telemetry: {:?}", e);
+ }
  for entry in finalized_batch.into_iter() {
  if let Some(ws_sink) = entry.messaging_sink {
  let merkle_root = hex::encode(batch_merkle_tree.root);
@@ -1102,6 +1127,13 @@ impl Batcher {
  info!("Uploading batch to S3...");
  self.upload_batch_to_s3(batch_bytes, &file_name).await?;
 
+ if let Err(e) = self
+ .telemetry
+ .task_uploaded_to_s3(&batch_merkle_root_hex)
+ .await
+ {
+ error!("Failed to send task status to telemetry: {:?}", e);
+ };
  info!("Batch sent to S3 with name: {}", file_name);
 
  info!("Uploading batch to contract");
@@ -1134,6 +1166,18 @@ impl Batcher {
  .gas_price_used_on_latest_batch
  .set(gas_price.as_u64() as i64);
 
+ if let Err(e) = self
+ .telemetry
+ .task_created(
+ &hex::encode(batch_merkle_root),
+ ethers::utils::format_ether(fee_per_proof),
+ num_proofs_in_batch,
+ )
+ .await
+ {
+ error!("Failed to send task status to telemetry: {:?}", e);
+ };
+
  match self
  .create_new_task(
  *batch_merkle_root,
@@ -1178,7 +1222,16 @@ impl Batcher {
  )
  .await
  {
- Ok(receipt) => Ok(receipt),
+ Ok(receipt) => {
+ if let Err(e) = self
+ .telemetry
+ .task_sent(&hex::encode(batch_merkle_root), receipt.transaction_hash)
+ .await
+ {
+ error!("Failed to send task status to telemetry: {:?}", e);
+ }
+ Ok(receipt)
+ }
  Err(BatcherSendError::TransactionReverted(err)) => {
  // Since transaction was reverted, we don't want to retry with fallback.
  warn!("Transaction reverted {:?}", err);

batcher/aligned-batcher/src/telemetry/mod.rs

@@ -0,0 +1 @@
+pub mod sender;

batcher/aligned-batcher/src/telemetry/sender.rs

@@ -0,0 +1,121 @@
+use ethers::types::H256;
+
+#[derive(Debug, serde::Serialize)]
+pub enum TraceMessage {
+ CreatingTask([u8; 32]),
+ TaskCreated(H256),
+ TaskCreationFailed(H256),
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct TraceMessageTask {
+ merkle_root: String,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct TraceMessageTaskStarted {
+ merkle_root: String,
+ fee_per_proof: String,
+ num_proofs_in_batch: usize,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct TraceMessageTaskSentToEthereum {
+ merkle_root: String,
+ tx_hash: H256,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct TraceMessageNewBatch {
+ merkle_root: String,
+ proof_count: usize,
+}
+
+#[derive(Debug, serde::Serialize)]
+pub struct TraceMessageTaskError {
+ merkle_root: String,
+ error: String,
+}
+
+pub struct TelemetrySender {
+ base_url: String,
+ client: reqwest::Client,
+}
+
+impl TelemetrySender {
+ pub fn new(base_url: String) -> Self {
+ let client = reqwest::Client::new();
+ Self { base_url, client }
+ }
+
+ pub fn get_full_url(&self, path: &str) -> String {
+ format!("{}/api/{}", self.base_url, path)
+ }
+
+ pub async fn init_task_trace(&self, batch_merkle_root: &str) -> Result<(), reqwest::Error> {
+ let url = self.get_full_url("initBatcherTaskTrace");
+ let formatted_merkle_root = format!("0x{}", batch_merkle_root);
+ let task = TraceMessageTask {
+ merkle_root: formatted_merkle_root,
+ };
+ self.client.post(&url).json(&task).send().await?;
+ Ok(())
+ }
+
+ pub async fn task_sent(
+ &self,
+ batch_merkle_root: &str,
+ tx_hash: H256,
+ ) -> Result<(), reqwest::Error> {
+ let url = self.get_full_url("batcherTaskSent");
+ let formatted_merkle_root = format!("0x{}", batch_merkle_root);
+ let task = TraceMessageTaskSentToEthereum {
+ merkle_root: formatted_merkle_root,
+ tx_hash,
+ };
+ self.client.post(&url).json(&task).send().await?;
+ Ok(())
+ }
+
+ pub async fn task_created(
+ &self,
+ batch_merkle_root: &str,
+ fee_per_proof: String,
+ num_proofs_in_batch: usize,
+ ) -> Result<(), reqwest::Error> {
+ let url = self.get_full_url("batcherTaskStarted");
+ let formatted_merkle_root = format!("0x{}", batch_merkle_root);
+ let task = TraceMessageTaskStarted {
+ merkle_root: formatted_merkle_root,
+ fee_per_proof,
+ num_proofs_in_batch,
+ };
+ self.client.post(&url).json(&task).send().await?;
+ Ok(())
+ }
+
+ pub async fn task_uploaded_to_s3(&self, batch_merkle_root: &str) -> Result<(), reqwest::Error> {
+ let url = self.get_full_url("batcherTaskUploadedToS3");
+ let formatted_merkle_root = format!("0x{}", batch_merkle_root);
+ let task = TraceMessageTask {
+ merkle_root: formatted_merkle_root,
+ };
+ self.client.post(&url).json(&task).send().await?;
+ Ok(())
+ }
+
+ pub async fn task_creation_failed(
+ &self,
+ batch_merkle_root: &str,
+ reason: &str,
+ ) -> Result<(), reqwest::Error> {
+ let url = self.get_full_url("batcherTaskCreationFailed");
+ let formatted_merkle_root = format!("0x{}", batch_merkle_root);
+ let task = TraceMessageTaskError {
+ merkle_root: formatted_merkle_root,
+ error: reason.to_string(),
+ };
+ self.client.post(&url).json(&task).send().await?;
+ Ok(())
+ }
+}

config-files/config-batcher-docker.yaml

@@ -23,6 +23,7 @@ batcher:
  eth_ws_reconnects: 99999999999999
  pre_verification_is_enabled: true
  metrics_port: 9093
+ telemetry_ip_port_address: localhost:4001
  non_paying:
  address: 0xa0Ee7A142d267C1f36714E4a8F75612F20a79720 # Anvil address 9
  replacement_private_key: ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80 # Anvil address 1

config-files/config-batcher.yaml

@@ -22,6 +22,7 @@ batcher:
  max_batch_size: 268435456 # 256 MiB
  pre_verification_is_enabled: true
  metrics_port: 9093
+ telemetry_ip_port_address: localhost:4001
  non_paying:
  address: 0xa0Ee7A142d267C1f36714E4a8F75612F20a79720 # Anvil address 9
  replacement_private_key: ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80 # Anvil address 1

explorer/mix.lock

@@ -23,7 +23,7 @@
  "floki": {:hex, :floki, "0.36.2", "a7da0193538c93f937714a6704369711998a51a6164a222d710ebd54020aa7a3", [:mix], [], "hexpm", "a8766c0bc92f074e5cb36c4f9961982eda84c5d2b8e979ca67f5c268ec8ed580"},
  "gettext": {:hex, :gettext, "0.24.0", "6f4d90ac5f3111673cbefc4ebee96fe5f37a114861ab8c7b7d5b30a1108ce6d8", [:mix], [{:expo, "~> 0.5.1", [hex: :expo, repo: "hexpm", optional: false]}], "hexpm", "bdf75cdfcbe9e4622dd18e034b227d77dd17f0f133853a1c73b97b3d6c770e8b"},
  "hackney": {:hex, :hackney, "1.20.1", "8d97aec62ddddd757d128bfd1df6c5861093419f8f7a4223823537bad5d064e2", [:rebar3], [{:certifi, "~> 2.12.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.4.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "fe9094e5f1a2a2c0a7d10918fee36bfec0ec2a979994cff8cfe8058cd9af38e3"},
- "heroicons": {:git, "https://github.com/tailwindlabs/heroicons.git", "88ab3a0d790e6a47404cba02800a6b25d2afae50", [tag: "v2.1.1", sparse: "optimized", depth: 1]},
+ "heroicons": {:git, "https://github.com/tailwindlabs/heroicons.git", "88ab3a0d790e6a47404cba02800a6b25d2afae50", [tag: "v2.1.1", sparse: "optimized"]},
  "hpax": {:hex, :hpax, "0.2.0", "5a58219adcb75977b2edce5eb22051de9362f08236220c9e859a47111c194ff5", [:mix], [], "hexpm", "bea06558cdae85bed075e6c036993d43cd54d447f76d8190a8db0dc5893fa2f1"},
  "httpoison": {:hex, :httpoison, "2.2.1", "87b7ed6d95db0389f7df02779644171d7319d319178f6680438167d7b69b1f3d", [:mix], [{:hackney, "~> 1.17", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm", "51364e6d2f429d80e14fe4b5f8e39719cacd03eb3f9a9286e61e216feac2d2df"},
  "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},