add prober_id cli arg

Author: matthieugouel

src/handler.rs

@@ -138,6 +138,7 @@ fn generate_probes(target: &Target) -> Result<Vec<Probe>> {
 }
 
 pub async fn handle(
+    prober_id: u16,
     brokers: &str,
     _in_topics: &str,
     _in_group_id: &str,
@@ -154,7 +155,7 @@ pub async fn handle(
     let (_, _, results) = result?;
 
     // Produce the results to Kafka topic
-    produce(brokers, out_topic, results).await;
+    produce(brokers, out_topic, prober_id, results).await;
 
     Ok(())
 }

src/main.rs

@@ -16,6 +16,10 @@ use crate::handler::handle;
 #[derive(CliParser, Debug, Clone)]
 #[command(version, about, long_about = None)]
 struct Cli {
+    /// Prober ID
+    #[clap(long, default_value = "0")]
+    prober_id: u16,
+
     /// Kafka brokers
     #[clap(long, default_value = "localhost:9092")]
     brokers: String,
@@ -62,6 +66,7 @@ async fn main() -> Result<()> {
     set_logging(&cli);
 
     match handle(
+        cli.prober_id,
         &cli.brokers,
         &cli.in_topics,
         &cli.in_group_id,

src/producer.rs

@@ -20,9 +20,10 @@ fn format_mpls_labels(mpls_labels: &Vec<MPLSLabel>) -> String {
         + "]"
 }
 
-fn format_reply(reply: &Reply) -> String {
+fn format_reply(prober_id: u16, reply: &Reply) -> String {
     let mut output = vec![];
     output.push(format!("{}", reply.capture_timestamp.as_millis()));
+    output.push(format!("{}", prober_id));
     output.push(format!("{}", reply.reply_src_addr));
     output.push(format!("{}", reply.reply_dst_addr));
     output.push(format!("{}", reply.reply_id));
@@ -46,7 +47,12 @@ fn format_reply(reply: &Reply) -> String {
     output.join(",")
 }
 
-pub async fn produce(brokers: &str, topic_name: &str, results: Arc<Mutex<Vec<Reply>>>) {
+pub async fn produce(
+    brokers: &str,
+    topic_name: &str,
+    prober_id: u16,
+    results: Arc<Mutex<Vec<Reply>>>,
+) {
     let producer: &FutureProducer = &ClientConfig::new()
         .set("bootstrap.servers", brokers)
         .set("message.timeout.ms", "5000")
@@ -57,7 +63,7 @@ pub async fn produce(brokers: &str, topic_name: &str, results: Arc<Mutex<Vec<Rep
         let _ = producer
             .send(
                 FutureRecord::to(topic_name)
-                    .payload(&format!("{}", format_reply(result)))
+                    .payload(&format!("{}", format_reply(prober_id, result)))
                     .key(&format!("Key"))
                     .headers(OwnedHeaders::new().insert(Header {
                         key: "header_key",

testbed/config/clickhouse/docker-entrypoint-initdb.d/init.sql

@@ -2,6 +2,7 @@ CREATE DATABASE osiris;
 CREATE TABLE osiris.results_broker
 (
     timestamp DateTime64,
+    prober_id UInt16,
     reply_src_addr IPv6,
     reply_dst_addr IPv6,
     reply_id UInt16,
@@ -32,6 +33,7 @@ SETTINGS
 CREATE TABLE osiris.results
 (
     timestamp DateTime64,
+    prober_id UInt16,
     reply_src_addr IPv6,
     reply_dst_addr IPv6,
     reply_id UInt16,