add: hook up validation::block_connected tracepoint

extractor/build.rs

@@ -17,7 +17,7 @@ fn main() {
     {
         Ok(_) => (),
         Err(e) => {
-            println!("{}", e);
+            println!("Error: {}", e);
             panic!("could not compile {}", SOURCE);
         }
     }

extractor/src/bpf/tracing.bpf.c

@@ -422,4 +422,33 @@ int BPF_USDT(handle_mempool_rejected, void *txid, void *reason) {
     return bpf_ringbuf_output(&mempool_rejected, &rejected, sizeof(rejected), 0);
 };
 
+// VALIDATION
+
+#define VALIDATION_BLOCK_CONNECTED_PAGES 64
+
+RINGBUFFER(validation_block_connected, VALIDATION_BLOCK_CONNECTED_PAGES)
+
+#define HASH_LENGHT 32
+
+struct BlockConnected {
+  u8     hash[HASH_LENGHT];
+  s32    height;
+  u64    transactions;
+  s32    inputs;
+  u64    sigops;
+  u64    connection_time;  
+};
+
+SEC("usdt")
+int BPF_USDT(handle_validation_block_connected, void *hash, s32 height, u64 transactions, s32 inputs, u64 sigops, u64 connection_time) {
+    struct BlockConnected connected = {};
+    bpf_probe_read(&connected.hash, sizeof(connected.hash), hash);
+    connected.height = height;
+    connected.transactions = transactions;
+    connected.inputs = inputs; 
+    connected.sigops = sigops; 
+    connected.connection_time = connection_time;
+    return bpf_ringbuf_output(&validation_block_connected, &connected, sizeof(connected), 0);
+};
+
 char LICENSE[] SEC("license") = "Dual BSD/GPL";

extractor/src/main.rs

@@ -16,11 +16,11 @@ use prost::Message;
 use shared::ctypes::{
     AddrmanInsertNew, AddrmanInsertTried, ClosedConnection, InboundConnection, MempoolAdded,
     MempoolRejected, MempoolRemoved, MempoolReplaced, MisbehavingConnection, OutboundConnection,
-    P2PMessage, P2PMessageSize,
+    P2PMessage, P2PMessageSize, ValidationBlockConnected,
 };
 use shared::wrapper::wrapper::Wrap;
 use shared::wrapper::Wrapper;
-use shared::{addrman, mempool, net_conn, net_msg};
+use shared::{addrman, mempool, net_conn, net_msg, validation};
 
 use crate::tracing::OpenTracingSkel;
 
@@ -107,6 +107,12 @@ const _TRACEPOINTS_ADDRMAN: [Tracepoint; 2] = [
     },
 ];
 
+const TRACEPOINTS_VALIDATION: [Tracepoint; 1] = [Tracepoint {
+    context: "validation",
+    name: "block_connected",
+    function: "handle_validation_block_connected",
+}];
+
 fn bump_memlock_rlimit() {
     let rlimit = libc::rlimit {
         rlim_cur: 128 << 20,
@@ -142,6 +148,7 @@ fn main() -> Result<(), libbpf_rs::Error> {
     let active_tracepoints = TRACEPOINTS_NET_MESSAGE
         .iter()
         .chain(TRACEPOINTS_NET_CONN.iter())
+        .chain(TRACEPOINTS_VALIDATION.iter())
         .chain(TRACEPOINTS_MEMPOOL.iter());
     let mut links = Vec::new();
     for tracepoint in active_tracepoints {
@@ -176,7 +183,8 @@ fn main() -> Result<(), libbpf_rs::Error> {
         .add(obj.map("mempool_added").unwrap(), |data| { handle_mempool_added(data, socket.clone()) })?
         .add(obj.map("mempool_removed").unwrap(), |data| { handle_mempool_removed(data, socket.clone()) })?
         .add(obj.map("mempool_rejected").unwrap(), |data| { handle_mempool_rejected(data, socket.clone()) })?
-        .add(obj.map("mempool_replaced").unwrap(), |data| { handle_mempool_replaced(data, socket.clone()) })?;
+        .add(obj.map("mempool_replaced").unwrap(), |data| { handle_mempool_replaced(data, socket.clone()) })?
+        .add(obj.map("validation_block_connected").unwrap(), |data| { handle_validation_block_connected(data, socket.clone()) })?;
     let ring_buffers = ringbuff_builder.build()?;
 
     loop {
@@ -415,3 +423,23 @@ fn handle_mempool_rejected(data: &[u8], s: Socket) -> i32 {
     s.send(&proto.encode_to_vec()).unwrap();
     0
 }
+
+fn handle_validation_block_connected(data: &[u8], s: Socket) -> i32 {
+    let connected = ValidationBlockConnected::from_bytes(data);
+    let now = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap();
+    let timestamp = now.as_secs();
+    let timestamp_subsec_millis = now.subsec_micros();
+    let proto = Wrapper {
+        timestamp: timestamp,
+        timestamp_subsec_micros: timestamp_subsec_millis,
+        wrap: Some(Wrap::Validation(validation::ValidationEvent {
+            event: Some(validation::validation_event::Event::BlockConnected(
+                connected.into(),
+            )),
+        })),
+    };
+    s.send(&proto.encode_to_vec()).unwrap();
+    0
+}

protobuf/proto-types/validation.proto

@@ -0,0 +1,20 @@
+syntax = "proto2";
+
+package validation;
+
+message ValidationEvent {
+  oneof event {
+    BlockConnected block_connected = 1;
+  }
+}
+
+// A block connected to the chain.
+message BlockConnected {
+  required bytes    hash                    = 1;  // Hash of the connected block.
+  required int32    height                  = 2;  // Height of the connected block.
+  required int64    transactions            = 3;  // Number of transactions in the connected block.
+  required int32    inputs                  = 4;  // Number of inputs in the connected block.
+  required int64    sigops                  = 5;  // Number of sigops in the connected block.
+  required int64    connection_time         = 6;  // Time it took to connect the block in microseconds (µs).
+}
+

protobuf/proto-types/wrapper.proto

@@ -6,6 +6,7 @@ import "net_msg.proto";
 import "net_conn.proto";
 import "addrman.proto";
 import "mempool.proto";
+import "validation.proto";
 
 message Wrapper {
   required uint64  timestamp = 10;  // Timestamp (seconds since UNIX epoch) when the message was received.
@@ -15,6 +16,7 @@ message Wrapper {
     net_conn.ConnectionEvent conn = 2;
     addrman.AddrmanEvent addrman = 3;
     mempool.MempoolEvent mempool = 4;
+    validation.ValidationEvent validation = 5;
   }
 }