allow to disable Kafka producer

Author: matthieugouel

src/agent/handler.rs

@@ -108,7 +108,7 @@ pub async fn handle(config: &AppConfig) -> Result<()> {
     info!("Starting receiver");
     ReceiveLoop::new(tx, config.caracat.clone());
 
-    // Start the Kafka producer task
+    // Start the Kafka producer task if enabled
     let config_task = config.clone();
     let kafka_auth_task = kafka_auth.clone();
     task::spawn(async move {

src/agent/producer.rs

@@ -1,5 +1,5 @@
 use caracat::models::{MPLSLabel, Reply};
-use log::info;
+use log::{info, warn};
 use rdkafka::config::ClientConfig;
 use rdkafka::message::{Header, OwnedHeaders};
 use rdkafka::producer::{FutureProducer, FutureRecord};
@@ -52,6 +52,13 @@ fn format_reply(agent_id: String, reply: &Reply) -> String {
 }
 
 pub async fn produce(config: &AppConfig, auth: KafkaAuth, rx: Receiver<Reply>) {
+    if config.kafka.out_enable == false {
+        warn!("Kafka producer is disabled");
+        loop {
+            rx.recv().unwrap();
+        }
+    }
+
     let producer: &FutureProducer = match auth {
         KafkaAuth::PlainText => &ClientConfig::new()
             .set("bootstrap.servers", config.kafka.brokers.clone())
@@ -69,7 +76,6 @@ pub async fn produce(config: &AppConfig, auth: KafkaAuth, rx: Receiver<Reply>) {
             .expect("Producer creation error"),
     };
 
-    // TODO: send batch of replies
     loop {
         let result = rx.recv().unwrap();
         let delivery_status = producer

src/config.rs

@@ -65,6 +65,9 @@ pub struct KafkaConfig {
 
     /// Kafka producer topic
     pub out_topic: String,
+
+    /// Enable Kafka producer
+    pub out_enable: bool,
 }
 
 #[derive(Debug, Clone)]
@@ -136,6 +139,7 @@ pub fn app_config(config_path: &str) -> AppConfig {
             out_topic: config
                 .get_string("kafka.out_topic")
                 .unwrap_or("saimiris-results".to_string()),
+            out_enable: config.get_bool("kafka.out_enable").unwrap_or(true),
         },
 
         // Agent configuration

testbed/config/saimiris/saimiris.yml

@@ -5,6 +5,7 @@ kafka:
   in_topics: saimiris-targets
   in_group_id: saimiris-agent-wbmwwp9vna
 
+  out_enable: true
   out_topic: saimiris-results
 
 agent: