chore: migrate to kafka props files for configs (#25)

* migrate to kafka properties file

* migrate other services to kafka properties file

* fmt

Author: danyalprout

.env.example

@@ -3,16 +3,14 @@ TIPS_INGRESS_ADDRESS=0.0.0.0
 TIPS_INGRESS_PORT=8080
 TIPS_INGRESS_RPC_MEMPOOL=http://localhost:2222
 TIPS_INGRESS_DUAL_WRITE_MEMPOOL=false
-TIPS_INGRESS_KAFKA_BROKERS=localhost:9092
-TIPS_INGRESS_KAFKA_TOPIC=tips-audit
+TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE=/app/docker/ingress-kafka-properties
+TIPS_INGRESS_KAFKA_INGRESS_TOPIC=tips-ingress
 TIPS_INGRESS_LOG_LEVEL=info
-TIPS_INGRESS_KAFKA_QUEUE_TOPIC=tips-ingress-rpc
 TIPS_INGRESS_SEND_TRANSACTION_DEFAULT_LIFETIME_SECONDS=10800
 
 # Audit service configuration
-TIPS_AUDIT_KAFKA_BROKERS=localhost:9092
+TIPS_AUDIT_KAFKA_PROPERTIES_FILE=/app/docker/audit-kafka-properties
 TIPS_AUDIT_KAFKA_TOPIC=tips-audit
-TIPS_AUDIT_KAFKA_GROUP_ID=local-audit
 TIPS_AUDIT_LOG_LEVEL=info
 TIPS_AUDIT_S3_BUCKET=tips
 TIPS_AUDIT_S3_CONFIG_TYPE=manual
@@ -25,7 +23,7 @@ TIPS_AUDIT_S3_SECRET_ACCESS_KEY=minioadmin
 TIPS_MAINTENANCE_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres
 TIPS_MAINTENANCE_RPC_URL=http://localhost:2222
 TIPS_MAINTENANCE_RPC_POLL_INTERVAL_MS=250
-TIPS_MAINTENANCE_KAFKA_BROKERS=localhost:9092
+TIPS_MAINTENANCE_KAFKA_PROPERTIES_FILE=/app/docker/maintenance-kafka-properties
 TIPS_MAINTENANCE_FLASHBLOCKS_WS=ws://localhost:1115/ws
 TIPS_MAINTENANCE_KAFKA_TOPIC=tips-audit
 TIPS_MAINTENANCE_LOG_LEVEL=info
@@ -42,7 +40,7 @@ TIPS_UI_S3_SECRET_ACCESS_KEY=minioadmin
 
 # Ingress Writer
 TIPS_INGRESS_WRITER_DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres
-TIPS_INGRESS_WRITER_KAFKA_BROKERS=localhost:9092
-TIPS_INGRESS_WRITER_KAFKA_TOPIC=tips-ingress-rpc
-TIPS_INGRESS_WRITER_KAFKA_GROUP_ID=local-writer
+TIPS_INGRESS_WRITER_KAFKA_PROPERTIES_FILE=/app/docker/ingress-writer-kafka-properties
+TIPS_INGRESS_KAFKA_TOPIC=tips-ingress
+TIPS_INGRESS_WRITER_AUDIT_TOPIC=tips-audit
 TIPS_INGRESS_WRITER_LOG_LEVEL=info

crates/audit/src/bin/main.rs

@@ -19,15 +19,12 @@ enum S3ConfigType {
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
 struct Args {
-    #[arg(long, env = "TIPS_AUDIT_KAFKA_BROKERS")]
-    kafka_brokers: String,
+    #[arg(long, env = "TIPS_AUDIT_KAFKA_PROPERTIES_FILE")]
+    kafka_properties_file: String,
 
     #[arg(long, env = "TIPS_AUDIT_KAFKA_TOPIC")]
     kafka_topic: String,
 
-    #[arg(long, env = "TIPS_AUDIT_KAFKA_GROUP_ID")]
-    kafka_group_id: String,
-
     #[arg(long, env = "TIPS_AUDIT_S3_BUCKET")]
     s3_bucket: String,
 
@@ -80,14 +77,13 @@ async fn main() -> Result<()> {
         .init();
 
     info!(
-        kafka_brokers = %args.kafka_brokers,
+        kafka_properties_file = %args.kafka_properties_file,
         kafka_topic = %args.kafka_topic,
-        kafka_group_id = %args.kafka_group_id,
         s3_bucket = %args.s3_bucket,
         "Starting audit archiver"
     );
 
-    let consumer = create_kafka_consumer(&args.kafka_brokers, &args.kafka_group_id)?;
+    let consumer = create_kafka_consumer(&args.kafka_properties_file)?;
     consumer.subscribe(&[&args.kafka_topic])?;
 
     let reader = KafkaMempoolReader::new(consumer, args.kafka_topic.clone())?;

crates/audit/src/reader.rs

@@ -7,24 +7,36 @@ use rdkafka::{
     consumer::{Consumer, StreamConsumer},
     message::Message,
 };
+use std::fs;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::time::sleep;
-use tracing::{debug, error};
-
-pub fn create_kafka_consumer(kafka_brokers: &str, group_id: &str) -> Result<StreamConsumer> {
-    let consumer: StreamConsumer = ClientConfig::new()
-        .set("group.id", group_id)
-        .set("bootstrap.servers", kafka_brokers)
-        .set("enable.partition.eof", "false")
-        .set("session.timeout.ms", "6000")
-        .set("enable.auto.commit", "false")
-        .set("auto.offset.reset", "earliest")
-        .set("fetch.wait.max.ms", "100")
-        .set("fetch.min.bytes", "1")
-        .create()?;
+use tracing::{debug, error, info};
+
+pub fn create_kafka_consumer(kafka_properties_file: &str) -> Result<StreamConsumer> {
+    let client_config = load_kafka_config_from_file(kafka_properties_file)?;
+    let consumer: StreamConsumer = client_config.create()?;
     Ok(consumer)
 }
 
+fn load_kafka_config_from_file(properties_file_path: &str) -> Result<ClientConfig> {
+    let kafka_properties = fs::read_to_string(properties_file_path)?;
+    info!("Kafka properties:\n{}", kafka_properties);
+
+    let mut client_config = ClientConfig::new();
+
+    for line in kafka_properties.lines() {
+        let line = line.trim();
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        if let Some((key, value)) = line.split_once('=') {
+            client_config.set(key.trim(), value.trim());
+        }
+    }
+
+    Ok(client_config)
+}
+
 pub fn assign_topic_partition(consumer: &StreamConsumer, topic: &str) -> Result<()> {
     let mut tpl = TopicPartitionList::new();
     tpl.add_partition(topic, 0);

crates/ingress-rpc/src/main.rs

@@ -4,6 +4,7 @@ use jsonrpsee::server::Server;
 use op_alloy_network::Optimism;
 use rdkafka::ClientConfig;
 use rdkafka::producer::FutureProducer;
+use std::fs;
 use std::net::IpAddr;
 use tracing::{info, warn};
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
@@ -35,24 +36,16 @@ struct Config {
     dual_write_mempool: bool,
 
     /// Kafka brokers for publishing mempool events
-    #[arg(long, env = "TIPS_INGRESS_KAFKA_BROKERS")]
-    kafka_brokers: String,
-
-    /// Kafka topic for publishing mempool events
-    #[arg(
-        long,
-        env = "TIPS_INGRESS_KAFKA_TOPIC",
-        default_value = "mempool-events"
-    )]
-    kafka_topic: String,
+    #[arg(long, env = "TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE")]
+    ingress_kafka_properties: String,
 
     /// Kafka topic for queuing transactions before the DB Writer
     #[arg(
         long,
-        env = "TIPS_INGRESS_KAFKA_QUEUE_TOPIC",
-        default_value = "tips-ingress-rpc"
+        env = "TIPS_INGRESS_KAFKA_INGRESS_TOPIC",
+        default_value = "tips-ingress"
     )]
-    queue_topic: String,
+    ingress_topic: String,
 
     #[arg(long, env = "TIPS_INGRESS_LOG_LEVEL", default_value = "info")]
     log_level: String,
@@ -106,12 +99,11 @@ async fn main() -> anyhow::Result<()> {
         .network::<Optimism>()
         .connect_http(config.mempool_url);
 
-    let queue_producer: FutureProducer = ClientConfig::new()
-        .set("bootstrap.servers", &config.kafka_brokers)
-        .set("message.timeout.ms", "5000")
-        .create()?;
+    let client_config = load_kafka_config_from_file(&config.ingress_kafka_properties)?;
+
+    let queue_producer: FutureProducer = client_config.create()?;
 
-    let queue = KafkaQueuePublisher::new(queue_producer, config.queue_topic);
+    let queue = KafkaQueuePublisher::new(queue_producer, config.ingress_topic);
 
     let service = IngressService::new(
         provider,
@@ -133,3 +125,22 @@ async fn main() -> anyhow::Result<()> {
     handle.stopped().await;
     Ok(())
 }
+
+fn load_kafka_config_from_file(properties_file_path: &str) -> anyhow::Result<ClientConfig> {
+    let kafka_properties = fs::read_to_string(properties_file_path)?;
+    info!("Kafka properties:\n{}", kafka_properties);
+
+    let mut client_config = ClientConfig::new();
+
+    for line in kafka_properties.lines() {
+        let line = line.trim();
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        if let Some((key, value)) = line.split_once('=') {
+            client_config.set(key.trim(), value.trim());
+        }
+    }
+
+    Ok(client_config)
+}

crates/ingress-writer/src/main.rs

@@ -8,6 +8,7 @@ use rdkafka::{
     message::Message,
     producer::FutureProducer,
 };
+use std::fs;
 use tips_audit::{BundleEvent, BundleEventPublisher, KafkaBundleEventPublisher};
 use tips_datastore::{BundleDatastore, postgres::PostgresDatastore};
 use tokio::time::Duration;
@@ -20,18 +21,18 @@ struct Args {
     #[arg(long, env = "TIPS_INGRESS_WRITER_DATABASE_URL")]
     database_url: String,
 
-    #[arg(long, env = "TIPS_INGRESS_WRITER_KAFKA_BROKERS")]
-    kafka_brokers: String,
+    #[arg(long, env = "TIPS_INGRESS_WRITER_KAFKA_PROPERTIES_FILE")]
+    kafka_properties_file: String,
+
+    #[arg(long, env = "TIPS_INGRESS_KAFKA_TOPIC", default_value = "tips-ingress")]
+    ingress_topic: String,
 
     #[arg(
         long,
-        env = "TIPS_INGRESS_WRITER_KAFKA_TOPIC",
-        default_value = "tips-ingress-rpc"
+        env = "TIPS_INGRESS_WRITER_AUDIT_TOPIC",
+        default_value = "tips-audit"
     )]
-    kafka_topic: String,
-
-    #[arg(long, env = "TIPS_INGRESS_WRITER_KAFKA_GROUP_ID")]
-    kafka_group_id: String,
+    audit_topic: String,
 
     #[arg(long, env = "TIPS_INGRESS_WRITER_LOG_LEVEL", default_value = "info")]
     log_level: String,
@@ -129,31 +130,25 @@ async fn main() -> Result<()> {
         .with_env_filter(&args.log_level)
         .init();
 
-    let mut config = ClientConfig::new();
-    config
-        .set("group.id", &args.kafka_group_id)
-        .set("bootstrap.servers", &args.kafka_brokers)
-        .set("auto.offset.reset", "earliest")
-        .set("enable.partition.eof", "false")
-        .set("session.timeout.ms", "6000")
-        .set("enable.auto.commit", "true");
-
-    let kafka_producer: FutureProducer = ClientConfig::new()
-        .set("bootstrap.servers", &args.kafka_brokers)
-        .set("message.timeout.ms", "5000")
-        .create()?;
-
-    let publisher = KafkaBundleEventPublisher::new(kafka_producer, "tips-audit".to_string());
+    let config = load_kafka_config_from_file(&args.kafka_properties_file)?;
+    let kafka_producer: FutureProducer = config.create()?;
+
+    let publisher = KafkaBundleEventPublisher::new(kafka_producer, args.audit_topic.clone());
     let consumer = config.create()?;
 
     let bundle_store = PostgresDatastore::connect(args.database_url).await?;
     bundle_store.run_migrations().await?;
 
-    let writer = IngressWriter::new(consumer, args.kafka_topic.clone(), bundle_store, publisher)?;
+    let writer = IngressWriter::new(
+        consumer,
+        args.ingress_topic.clone(),
+        bundle_store,
+        publisher,
+    )?;
 
     info!(
         "Ingress Writer service started, consuming from topic: {}",
-        args.kafka_topic
+        args.ingress_topic
     );
     loop {
         match writer.insert_bundle().await {
@@ -167,3 +162,22 @@ async fn main() -> Result<()> {
         }
     }
 }
+
+fn load_kafka_config_from_file(properties_file_path: &str) -> Result<ClientConfig> {
+    let kafka_properties = fs::read_to_string(properties_file_path)?;
+    info!("Kafka properties:\n{}", kafka_properties);
+
+    let mut client_config = ClientConfig::new();
+
+    for line in kafka_properties.lines() {
+        let line = line.trim();
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        if let Some((key, value)) = line.split_once('=') {
+            client_config.set(key.trim(), value.trim());
+        }
+    }
+
+    Ok(client_config)
+}

crates/maintenance/src/main.rs

@@ -9,6 +9,7 @@ use clap::Parser;
 use op_alloy_network::Optimism;
 use rdkafka::ClientConfig;
 use rdkafka::producer::FutureProducer;
+use std::fs;
 use std::sync::Arc;
 use tips_audit::KafkaBundleEventPublisher;
 use tips_datastore::PostgresDatastore;
@@ -19,8 +20,8 @@ use url::Url;
 #[derive(Parser, Clone)]
 #[command(author, version, about, long_about = None)]
 pub struct Args {
-    #[arg(long, env = "TIPS_MAINTENANCE_KAFKA_BROKERS")]
-    pub kafka_brokers: String,
+    #[arg(long, env = "TIPS_MAINTENANCE_KAFKA_PROPERTIES_FILE")]
+    pub kafka_properties_file: String,
 
     #[arg(
         long,
@@ -100,10 +101,8 @@ async fn main() -> Result<()> {
 
     let datastore = PostgresDatastore::connect(args.database_url.clone()).await?;
 
-    let kafka_producer: FutureProducer = ClientConfig::new()
-        .set("bootstrap.servers", &args.kafka_brokers)
-        .set("message.timeout.ms", "5000")
-        .create()?;
+    let client_config = load_kafka_config_from_file(&args.kafka_properties_file)?;
+    let kafka_producer: FutureProducer = client_config.create()?;
 
     let publisher = KafkaBundleEventPublisher::new(kafka_producer, args.kafka_topic.clone());
 
@@ -125,3 +124,22 @@ async fn main() -> Result<()> {
 
     Ok(())
 }
+
+fn load_kafka_config_from_file(properties_file_path: &str) -> Result<ClientConfig> {
+    let kafka_properties = fs::read_to_string(properties_file_path)?;
+    info!("Kafka properties:\n{}", kafka_properties);
+
+    let mut client_config = ClientConfig::new();
+
+    for line in kafka_properties.lines() {
+        let line = line.trim();
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        if let Some((key, value)) = line.split_once('=') {
+            client_config.set(key.trim(), value.trim());
+        }
+    }
+
+    Ok(client_config)
+}

docker-compose.tips.yml

@@ -8,6 +8,8 @@ services:
       - "8080:8080"
     env_file:
       - .env.docker
+    volumes:
+      - ./docker/ingress-kafka-properties:/app/docker/ingress-kafka-properties:ro
     restart: unless-stopped
 
   audit:
@@ -17,6 +19,8 @@ services:
     container_name: tips-audit
     env_file:
       - .env.docker
+    volumes:
+      - ./docker/audit-kafka-properties:/app/docker/audit-kafka-properties:ro
     restart: unless-stopped
 
   maintenance:
@@ -26,6 +30,8 @@ services:
     container_name: tips-maintenance
     env_file:
       - .env.docker
+    volumes:
+      - ./docker/maintenance-kafka-properties:/app/docker/maintenance-kafka-properties:ro
     restart: unless-stopped
 
   ui:
@@ -46,4 +52,6 @@ services:
     container_name: tips-ingress-writer
     env_file:
       - .env.docker
+    volumes:
+      - ./docker/ingress-writer-kafka-properties:/app/docker/ingress-writer-kafka-properties:ro
     restart: unless-stopped