chore: add ingress rpc metrics (#105)

* chore: add ingress rpc metrics

* bump logs

* remove non error log

Author: danyalprout

crates/ingress-rpc/src/metrics.rs

@@ -18,6 +18,24 @@ pub fn record_histogram(rpc_latency: Duration, rpc: String) {
 #[derive(Metrics, Clone)]
 #[metrics(scope = "tips_ingress_rpc")]
 pub struct Metrics {
+    #[metric(describe = "Number of valid transactions received")]
+    pub transactions_received: Counter,
+
+    #[metric(describe = "Number of valid bundles parsed")]
+    pub bundles_parsed: Counter,
+
+    #[metric(describe = "Number of bundles simulated")]
+    pub successful_simulations: Counter,
+
+    #[metric(describe = "Number of bundles simulated")]
+    pub failed_simulations: Counter,
+
+    #[metric(describe = "Number of bundles sent to kafka")]
+    pub sent_to_kafka: Counter,
+
+    #[metric(describe = "Number of transactions sent to mempool")]
+    pub sent_to_mempool: Counter,
+
     #[metric(describe = "Duration of validate_tx")]
     pub validate_tx_duration: Histogram,
 

crates/ingress-rpc/src/service.rs

@@ -204,6 +204,8 @@ impl<Q: MessageQueue + 'static> IngressApiServer for IngressService<Q> {
         let start = Instant::now();
         let transaction = self.get_tx(&data).await?;
 
+        self.metrics.transactions_received.increment(1);
+
         let send_to_kafka = matches!(
             self.tx_submission_method,
             TxSubmissionMethod::Kafka | TxSubmissionMethod::MempoolAndKafka
@@ -225,17 +227,23 @@ impl<Q: MessageQueue + 'static> IngressApiServer for IngressService<Q> {
             reverting_tx_hashes: vec![transaction.tx_hash()],
             ..Default::default()
         };
+
         let parsed_bundle: ParsedBundle = bundle
             .clone()
             .try_into()
             .map_err(|e: String| EthApiError::InvalidParams(e).into_rpc_err())?;
 
         let bundle_hash = &parsed_bundle.bundle_hash();
 
+        self.metrics.bundles_parsed.increment(1);
+
         let meter_bundle_response = self.meter_bundle(&bundle, bundle_hash).await.ok();
 
         if let Some(meter_info) = meter_bundle_response.as_ref() {
+            self.metrics.successful_simulations.increment(1);
             _ = self.builder_tx.send(meter_info.clone());
+        } else {
+            self.metrics.failed_simulations.increment(1);
         }
 
         let accepted_bundle =
@@ -250,6 +258,7 @@ impl<Q: MessageQueue + 'static> IngressApiServer for IngressService<Q> {
                 warn!(message = "Failed to publish Queue::enqueue_bundle", bundle_hash = %bundle_hash, error = %e);
             }
 
+            self.metrics.sent_to_kafka.increment(1);
             info!(message="queued singleton bundle", txn_hash=%transaction.tx_hash());
         }
 
@@ -260,6 +269,7 @@ impl<Q: MessageQueue + 'static> IngressApiServer for IngressService<Q> {
                 .await;
             match response {
                 Ok(_) => {
+                    self.metrics.sent_to_mempool.increment(1);
                     debug!(message = "sent transaction to the mempool", hash=%transaction.tx_hash());
                 }
                 Err(e) => {
@@ -287,7 +297,7 @@ impl<Q: MessageQueue + 'static> IngressApiServer for IngressService<Q> {
             });
         }
 
-        debug!(
+        info!(
             message = "processed transaction",
             bundle_hash = %bundle_hash,
             transaction_hash = %transaction.tx_hash(),

ui/src/lib/s3.ts

@@ -60,7 +60,6 @@ async function getObjectContent(key: string): Promise<string | null> {
     const body = await response.Body?.transformToString();
     return body || null;
   } catch (error) {
-    console.error(`Failed to get S3 object ${key}:`, error);
     return null;
   }
 }