Option to disable TLS for kafka

Author: jakeloo

cmd/root.go

@@ -124,10 +124,10 @@ func init() {
 	rootCmd.PersistentFlags().Int("storage-staging-postgres-maxIdleConns", 25, "PostgreSQL max idle connections for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-postgres-maxConnLifetime", 300, "PostgreSQL max connection lifetime in seconds for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-postgres-connectTimeout", 10, "PostgreSQL connection timeout in seconds for staging storage")
-	// Kafka storage flags - only for main storage (where blockchain data is committed)
 	rootCmd.PersistentFlags().String("storage-main-kafka-brokers", "", "Kafka brokers for main storage")
 	rootCmd.PersistentFlags().String("storage-main-kafka-username", "", "Kafka username for main storage")
 	rootCmd.PersistentFlags().String("storage-main-kafka-password", "", "Kafka password for main storage")
+	rootCmd.PersistentFlags().Bool("storage-main-kafka-enable-tls", true, "Enable TLS for Kafka connection in main storage")
 	rootCmd.PersistentFlags().String("storage-main-kafka-postgres-host", "", "PostgreSQL host for Kafka main storage bookkeeping")
 	rootCmd.PersistentFlags().Int("storage-main-kafka-postgres-port", 5432, "PostgreSQL port for Kafka main storage bookkeeping")
 	rootCmd.PersistentFlags().String("storage-main-kafka-postgres-username", "", "PostgreSQL username for Kafka main storage bookkeeping")
@@ -149,6 +149,9 @@ func init() {
 	rootCmd.PersistentFlags().Bool("publisher-enabled", false, "Toggle publisher")
 	rootCmd.PersistentFlags().String("publisher-mode", "default", "Publisher mode: default or parallel")
 	rootCmd.PersistentFlags().String("publisher-brokers", "", "Kafka brokers")
+	rootCmd.PersistentFlags().String("publisher-username", "", "Kafka username for publisher")
+	rootCmd.PersistentFlags().String("publisher-password", "", "Kafka password for publisher")
+	rootCmd.PersistentFlags().Bool("publisher-enable-tls", true, "Enable TLS for Kafka connection in publisher")
 	rootCmd.PersistentFlags().Bool("publisher-blocks-enabled", false, "Toggle block publisher")
 	rootCmd.PersistentFlags().String("publisher-blocks-topicName", "", "Kafka topic name for blocks")
 	rootCmd.PersistentFlags().Bool("publisher-transactions-enabled", false, "Toggle transaction publisher")
@@ -255,6 +258,7 @@ func init() {
 	viper.BindPFlag("storage.main.kafka.brokers", rootCmd.PersistentFlags().Lookup("storage-main-kafka-brokers"))
 	viper.BindPFlag("storage.main.kafka.username", rootCmd.PersistentFlags().Lookup("storage-main-kafka-username"))
 	viper.BindPFlag("storage.main.kafka.password", rootCmd.PersistentFlags().Lookup("storage-main-kafka-password"))
+	viper.BindPFlag("storage.main.kafka.enable_tls", rootCmd.PersistentFlags().Lookup("storage-main-kafka-enable-tls"))
 	viper.BindPFlag("storage.main.kafka.postgres.host", rootCmd.PersistentFlags().Lookup("storage-main-kafka-postgres-host"))
 	viper.BindPFlag("storage.main.kafka.postgres.port", rootCmd.PersistentFlags().Lookup("storage-main-kafka-postgres-port"))
 	viper.BindPFlag("storage.main.kafka.postgres.username", rootCmd.PersistentFlags().Lookup("storage-main-kafka-postgres-username"))
@@ -276,6 +280,9 @@ func init() {
 	viper.BindPFlag("publisher.enabled", rootCmd.PersistentFlags().Lookup("publisher-enabled"))
 	viper.BindPFlag("publisher.mode", rootCmd.PersistentFlags().Lookup("publisher-mode"))
 	viper.BindPFlag("publisher.brokers", rootCmd.PersistentFlags().Lookup("publisher-brokers"))
+	viper.BindPFlag("publisher.username", rootCmd.PersistentFlags().Lookup("publisher-username"))
+	viper.BindPFlag("publisher.password", rootCmd.PersistentFlags().Lookup("publisher-password"))
+	viper.BindPFlag("publisher.enable_tls", rootCmd.PersistentFlags().Lookup("publisher-enable-tls"))
 	viper.BindPFlag("publisher.blocks.enabled", rootCmd.PersistentFlags().Lookup("publisher-blocks-enabled"))
 	viper.BindPFlag("publisher.blocks.topicName", rootCmd.PersistentFlags().Lookup("publisher-blocks-topicName"))
 	viper.BindPFlag("publisher.transactions.enabled", rootCmd.PersistentFlags().Lookup("publisher-transactions-enabled"))

configs/config.go

@@ -102,10 +102,11 @@ type PostgresConfig struct {
 }
 
 type KafkaConfig struct {
-	Brokers  string          `mapstructure:"brokers"`
-	Username string          `mapstructure:"username"`
-	Password string          `mapstructure:"password"`
-	Postgres *PostgresConfig `mapstructure:"postgres"`
+	Brokers   string          `mapstructure:"brokers"`
+	Username  string          `mapstructure:"username"`
+	Password  string          `mapstructure:"password"`
+	EnableTLS bool            `mapstructure:"enable_tls"`
+	Postgres  *PostgresConfig `mapstructure:"postgres"`
 }
 
 type RPCBatchRequestConfig struct {
@@ -184,6 +185,7 @@ type PublisherConfig struct {
 	Brokers      string                     `mapstructure:"brokers"`
 	Username     string                     `mapstructure:"username"`
 	Password     string                     `mapstructure:"password"`
+	EnableTLS    bool                       `mapstructure:"enable_tls"`
 	Blocks       BlockPublisherConfig       `mapstructure:"blocks"`
 	Transactions TransactionPublisherConfig `mapstructure:"transactions"`
 	Traces       TracePublisherConfig       `mapstructure:"traces"`

internal/publisher/publisher.go

@@ -76,6 +76,9 @@ func (p *Publisher) initialize() error {
 			User: config.Cfg.Publisher.Username,
 			Pass: config.Cfg.Publisher.Password,
 		}.AsMechanism()))
+	}
+
+	if config.Cfg.Publisher.EnableTLS {
 		tlsDialer := &tls.Dialer{NetDialer: &net.Dialer{Timeout: 10 * time.Second}}
 		opts = append(opts, kgo.Dialer(tlsDialer.DialContext))
 	}

internal/storage/kafka_publisher.go

@@ -47,6 +47,9 @@ func NewKafkaPublisher(cfg *config.KafkaConfig) (*KafkaPublisher, error) {
 			User: cfg.Username,
 			Pass: cfg.Password,
 		}.AsMechanism()))
+	}
+
+	if cfg.EnableTLS {
 		tlsDialer := &tls.Dialer{NetDialer: &net.Dialer{Timeout: 10 * time.Second}}
 		opts = append(opts, kgo.Dialer(tlsDialer.DialContext))
 	}