Dev

.github/values/dev.yml

@@ -1,10 +1,3 @@
-volumes:
-  - name: kafka-cert
-    secret:
-    secretName: chainindexing.kafka.dev.tls.astranet.services
-volumeMounts:
-  - name: kafka-cert
-    mountPath: /certs/chainindexing.kafka.dev
 specAnnotations:
   prometheus.io/scrape: "true"
   prometheus.io/port: "8081"

.github/workflows/dev.yml

@@ -17,4 +17,6 @@ jobs:
     with:
       workloadName: astra-indexing-worker-dev
       replicas: "1"
+      kafkaCert: "true"
+      kafkaEnv: "dev"
     secrets: inherit

bootstrap/config/config.go

@@ -48,7 +48,9 @@ type KafkaService struct {
 	User               string   `yaml:"user" toml:"user" xml:"user" json:"user,omitempty"`
 	Password           string   `yaml:"password" toml:"password" xml:"password" json:"password,omitempty"`
 	AuthenticationType string   `yaml:"authentication_type" toml:"authentication_type" xml:"authentication_type" json:"authentication_type,omitempty"`
-	Env                string   `yaml:"env" toml:"env" xml:"env" json:"env,omitempty"`
+	CaCertPath         string   `yaml:"ca_cert_path" toml:"ca_cert_path" xml:"ca_cert_path" json:"ca_cert_path,omitempty"`
+	TlsCertPath        string   `yaml:"tls_cert_path" toml:"tls_cert_path" xml:"tls_cert_path" json:"tls_cert_path,omitempty"`
+	TlsKeyPath         string   `yaml:"tls_key_path" toml:"tls_key_path" xml:"tls_key_path" json:"tls_key_path,omitempty"`
 }
 
 type Blockchain struct {

cmd/astra-indexing/config.go

@@ -51,7 +51,9 @@ type CLIConfig struct {
 	KafkaUser               string
 	KafkaPassword           string
 	KafkaAuthenticationType string
-	KafkaEnv                string
+	CaCertPath              string
+	TlsCertPath             string
+	TlsKeyPath              string
 }
 
 func OverrideByCLIConfig(config *config.Config, cliConfig *CLIConfig) {
@@ -135,7 +137,13 @@ func OverrideByCLIConfig(config *config.Config, cliConfig *CLIConfig) {
 	if cliConfig.KafkaAuthenticationType != "" {
 		config.KafkaService.AuthenticationType = cliConfig.KafkaAuthenticationType
 	}
-	if cliConfig.KafkaEnv != "" {
-		config.KafkaService.Env = cliConfig.KafkaEnv
+	if cliConfig.CaCertPath != "" {
+		config.KafkaService.CaCertPath = cliConfig.CaCertPath
+	}
+	if cliConfig.TlsCertPath != "" {
+		config.KafkaService.TlsCertPath = cliConfig.TlsCertPath
+	}
+	if cliConfig.TlsKeyPath != "" {
+		config.KafkaService.TlsKeyPath = cliConfig.TlsKeyPath
 	}
 }

cmd/astra-indexing/run.go

@@ -158,9 +158,19 @@ func run(args []string) error {
 				EnvVars: []string{"KAFKA_AUTHEN_TYPE"},
 			},
 			&cli.StringFlag{
-				Name:    "kafkaEnv",
-				Usage:   "Kafka Env",
-				EnvVars: []string{"KAFKA_ENV"},
+				Name:    "caCertPath",
+				Usage:   "Ca Cert Path",
+				EnvVars: []string{"KAFKA_CA_CERT_PATH"},
+			},
+			&cli.StringFlag{
+				Name:    "tlsCertPath",
+				Usage:   "Tls Cert Path",
+				EnvVars: []string{"KAFKA_TLS_CERT_PATH"},
+			},
+			&cli.StringFlag{
+				Name:    "tlsKeyPath",
+				Usage:   "Tls Key Path",
+				EnvVars: []string{"KAFKA_TLS_KEY_PATH"},
 			},
 		},
 		Action: func(ctx *cli.Context) error {
@@ -243,8 +253,14 @@ func run(args []string) error {
 			if ctx.IsSet("kafkaAuthenticationType") {
 				cliConfig.KafkaAuthenticationType = ctx.String("kafkaAuthenticationType")
 			}
-			if ctx.IsSet("kafkaEnv") {
-				cliConfig.KafkaEnv = ctx.String("kafkaEnv")
+			if ctx.IsSet("caCertPath") {
+				cliConfig.CaCertPath = ctx.String("caCertPath")
+			}
+			if ctx.IsSet("tlsCertPath") {
+				cliConfig.TlsCertPath = ctx.String("tlsCertPath")
+			}
+			if ctx.IsSet("tlsKeyPath") {
+				cliConfig.TlsKeyPath = ctx.String("tlsKeyPath")
 			}
 
 			OverrideByCLIConfig(&config, &cliConfig)

infrastructure/kafka/consumer/connection.go

@@ -26,7 +26,9 @@ type Consumer[T any] struct {
 	Password           string
 	AuthenticationType string
 	Sigchan            chan os.Signal
-	Env                string
+	CaCertPath         string
+	TlsCertPath        string
+	TlsKeyPath         string
 }
 
 func (c *Consumer[T]) CreateConnection() error {
@@ -147,14 +149,19 @@ func (c *Consumer[T]) getDialer() (*kafka.Dialer, error) {
 		}
 		return dialer, nil
 	case "SSL":
-		tlsCertPath := utils.TLS_CERT_PATH
-		tlsKeyPath := utils.TLS_KEY_PATH
 		caCertPath := utils.CA_CERT_PATH
+		if c.CaCertPath != "" {
+			caCertPath = c.CaCertPath
+		}
 
-		if c.Env == "dev" {
-			tlsCertPath = utils.TLS_CERT_PATH_DEV
-			tlsKeyPath = utils.TLS_KEY_PATH_DEV
-			caCertPath = utils.CA_CERT_PATH_DEV
+		tlsCertPath := utils.TLS_CERT_PATH
+		if c.TlsCertPath != "" {
+			tlsCertPath = c.TlsCertPath
+		}
+
+		tlsKeyPath := utils.TLS_KEY_PATH
+		if c.TlsKeyPath != "" {
+			tlsKeyPath = c.TlsKeyPath
 		}
 
 		keypair, err := tls.LoadX509KeyPair(

infrastructure/kafka/consumer/worker/evm_txs_consumer.go

@@ -29,7 +29,9 @@ func RunEvmTxsConsumer(rdbHandle *rdb.Handle, config *config.Config, logger appl
 		Password:           config.KafkaService.Password,
 		AuthenticationType: config.KafkaService.AuthenticationType,
 		Sigchan:            sigchan,
-		Env:                config.KafkaService.Env,
+		CaCertPath:         config.KafkaService.CaCertPath,
+		TlsCertPath:        config.KafkaService.TlsCertPath,
+		TlsKeyPath:         config.KafkaService.TlsKeyPath,
 	}
 	errConn := evmTxsConsumer.CreateConnection()
 	if errConn != nil {

infrastructure/kafka/consumer/worker/internal_txs_consumer.go

@@ -46,7 +46,9 @@ func RunInternalTxsConsumer(rdbHandle *rdb.Handle, config *config.Config, logger
 		Password:           config.KafkaService.Password,
 		AuthenticationType: config.KafkaService.AuthenticationType,
 		Sigchan:            sigchan,
-		Env:                config.KafkaService.Env,
+		CaCertPath:         config.KafkaService.CaCertPath,
+		TlsCertPath:        config.KafkaService.TlsCertPath,
+		TlsKeyPath:         config.KafkaService.TlsKeyPath,
 	}
 	errConn := internalTxsConsumer.CreateConnection()
 	if errConn != nil {

infrastructure/kafka/consumer/worker/token_transfers_consumer.go

@@ -39,7 +39,9 @@ func RunTokenTransfersConsumer(rdbHandle *rdb.Handle, config *config.Config, log
 		Password:           config.KafkaService.Password,
 		AuthenticationType: config.KafkaService.AuthenticationType,
 		Sigchan:            sigchan,
-		Env:                config.KafkaService.Env,
+		CaCertPath:         config.KafkaService.CaCertPath,
+		TlsCertPath:        config.KafkaService.TlsCertPath,
+		TlsKeyPath:         config.KafkaService.TlsKeyPath,
 	}
 	errConn := tokenTransfersConsumer.CreateConnection()
 	if errConn != nil {

infrastructure/utils.go

@@ -18,10 +18,6 @@ const CA_CERT_PATH = "/certs/chainindexing.kafka.prod/ca.crt"
 const TLS_CERT_PATH = "/certs/chainindexing.kafka.prod/tls.crt"
 const TLS_KEY_PATH = "/certs/chainindexing.kafka.prod/tls.key"
 
-const CA_CERT_PATH_DEV = "/certs/chainindexing.kafka.dev/ca.crt"
-const TLS_CERT_PATH_DEV = "/certs/chainindexing.kafka.dev/tls.crt"
-const TLS_KEY_PATH_DEV = "/certs/chainindexing.kafka.dev/tls.key"
-
 const EVM_TXS_TOPIC = "evm-txs"
 const INTERNAL_TXS_TOPIC = "internal-txs"
 const TOKEN_TRANSFERS_TOPIC = "token-transfers"