migrate other services to kafka properties file

Author: danyalprout

.env.example

@@ -4,14 +4,13 @@ TIPS_INGRESS_PORT=8080
 TIPS_INGRESS_RPC_MEMPOOL=http://localhost:2222
 TIPS_INGRESS_DUAL_WRITE_MEMPOOL=false
 TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE=/app/docker/ingress-kafka-properties
-TIPS_INGRESS_KAFKA_INGRESS_TOPIC=tips-ingress-rpc
+TIPS_INGRESS_KAFKA_INGRESS_TOPIC=tips-ingress
 TIPS_INGRESS_LOG_LEVEL=info
 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
@@ -24,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
@@ -41,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

@@ -43,7 +43,7 @@ struct Config {
     #[arg(
         long,
         env = "TIPS_INGRESS_KAFKA_INGRESS_TOPIC",
-        default_value = "tips-ingress-rpc"
+        default_value = "tips-ingress"
     )]
     ingress_topic: String,
 
@@ -99,18 +99,7 @@ async fn main() -> anyhow::Result<()> {
         .network::<Optimism>()
         .connect_http(config.mempool_url);
 
-    let kafka_properties = fs::read_to_string(&config.ingress_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());
-        }
-    }
+    let client_config = load_kafka_config_from_file(&config.ingress_kafka_properties)?;
 
     let queue_producer: FutureProducer = client_config.create()?;
 
@@ -136,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,22 @@ 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_WRITER_KAFKA_TOPIC",
-        default_value = "tips-ingress-rpc"
+        env = "TIPS_INGRESS_KAFKA_TOPIC",
+        default_value = "tips-ingress"
     )]
-    kafka_topic: String,
+    ingress_topic: String,
 
-    #[arg(long, env = "TIPS_INGRESS_WRITER_KAFKA_GROUP_ID")]
-    kafka_group_id: String,
+    #[arg(
+        long,
+        env = "TIPS_INGRESS_WRITER_AUDIT_TOPIC",
+        default_value = "tips-audit"
+    )]
+    audit_topic: String,
 
     #[arg(long, env = "TIPS_INGRESS_WRITER_LOG_LEVEL", default_value = "info")]
     log_level: String,
@@ -129,31 +134,20 @@ 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 +161,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

@@ -19,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:
@@ -28,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:
@@ -48,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

docker-compose.yml

@@ -56,7 +56,7 @@ services:
     command: |
       sh -c "
         kafka-topics --create --if-not-exists --topic tips-audit --bootstrap-server kafka:29092 --partitions 3 --replication-factor 1
-        kafka-topics --create --if-not-exists --topic tips-ingress-rpc --bootstrap-server kafka:29092 --partitions 3 --replication-factor 1
+        kafka-topics --create --if-not-exists --topic tips-ingress --bootstrap-server kafka:29092 --partitions 3 --replication-factor 1
         kafka-topics --list --bootstrap-server kafka:29092
       "
 

docker/audit-kafka-properties

@@ -0,0 +1,10 @@
+# Kafka configuration properties for audit service
+bootstrap.servers=host.docker.internal:9094
+message.timeout.ms=5000
+group.id=local-audit
+enable.partition.eof=false
+session.timeout.ms=6000
+enable.auto.commit=false
+auto.offset.reset=earliest
+fetch.wait.max.ms=100
+fetch.min.bytes=1

docker/ingress-kafka-properties

@@ -1,3 +1,3 @@
 # Kafka configuration properties for ingress service
-bootstrap.servers=host.docker.internal:9092
+bootstrap.servers=host.docker.internal:9094
 message.timeout.ms=5000