Try to remove kafka setup

cmd/insights-ingress/main.go

@@ -54,31 +54,15 @@ func main() {
 	})
 
 	kafkaCfg := kafka.Config{
-		Brokers: cfg.KafkaConfig.KafkaBrokers,
 		GroupID: cfg.KafkaConfig.KafkaGroupID,
 	}
 
 	producerCfg := queue.ProducerConfig{
-		Brokers:              cfg.KafkaConfig.KafkaBrokers,
 		Topic:                cfg.KafkaConfig.KafkaTrackerTopic,
 		Async:                true,
 		KafkaDeliveryReports: cfg.KafkaConfig.KafkaDeliveryReports,
 	}
 
-	// Kafka SSL Config
-	if cfg.KafkaConfig.KafkaSSLConfig != (config.KafkaSSLCfg{}) {
-		kafkaCfg.CA = cfg.KafkaConfig.KafkaSSLConfig.KafkaCA
-		kafkaCfg.Username = cfg.KafkaConfig.KafkaSSLConfig.KafkaUsername
-		kafkaCfg.Password = cfg.KafkaConfig.KafkaSSLConfig.KafkaPassword
-		kafkaCfg.SASLMechanism = cfg.KafkaConfig.KafkaSSLConfig.SASLMechanism
-		kafkaCfg.Protocol = cfg.KafkaConfig.KafkaSSLConfig.Protocol
-		producerCfg.CA = cfg.KafkaConfig.KafkaSSLConfig.KafkaCA
-		producerCfg.Username = cfg.KafkaConfig.KafkaSSLConfig.KafkaUsername
-		producerCfg.Password = cfg.KafkaConfig.KafkaSSLConfig.KafkaPassword
-		producerCfg.SASLMechanism = cfg.KafkaConfig.KafkaSSLConfig.SASLMechanism
-		producerCfg.Protocol = cfg.KafkaConfig.KafkaSSLConfig.Protocol
-	}
-
 	validator := kafka.New(&kafkaCfg, cfg.KafkaConfig.ValidTopics...)
 
 	tracker := announcers.NewStatusAnnouncer(&producerCfg)

go.mod

@@ -4,7 +4,6 @@ go 1.16
 
 require (
 	github.com/aws/aws-sdk-go v1.44.5
-	github.com/confluentinc/confluent-kafka-go v1.8.2 // indirect
 	github.com/fsnotify/fsnotify v1.5.4 // indirect
 	github.com/go-chi/chi v4.1.2+incompatible
 	github.com/jarcoal/httpmock v1.0.8
@@ -19,7 +18,7 @@ require (
 	github.com/pelletier/go-toml/v2 v2.0.0 // indirect
 	github.com/prometheus/client_golang v1.12.1
 	github.com/prometheus/common v0.34.0 // indirect
-	github.com/redhatinsights/app-common-go v1.6.1
+	github.com/redhatinsights/app-common-go v1.6.3-0.20220524114951-d700ed08bcc7
 	github.com/redhatinsights/platform-go-middlewares v0.17.0
 	github.com/sirupsen/logrus v1.8.1
 	github.com/spf13/viper v1.11.0

go.sum

@@ -68,8 +68,6 @@ github.com/armon/go-metrics v0.3.10/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/aws/aws-sdk-go v1.38.51/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
-github.com/aws/aws-sdk-go v1.44.3 h1:GA9bJsWeJdwPtcKK9Uq3jfMaPko0ROWzVuqwFM7BBP0=
-github.com/aws/aws-sdk-go v1.44.3/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX4oIKwKHZo=
 github.com/aws/aws-sdk-go v1.44.5 h1:T4mckpWUfplPG4GA3FDWDCM1QaCzisjGzzeCVBhHKwQ=
 github.com/aws/aws-sdk-go v1.44.5/go.mod h1:y4AeaBuwd2Lk+GepC1E9v0qOiTws0MIWAX4oIKwKHZo=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
@@ -389,16 +387,8 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/procfs v0.7.3 h1:4jVXhlkAyzOScmCkXBTOLRLTz8EeU+eyjrwB/EPq0VU=
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
-github.com/redhatinsights/app-common-go v1.6.1 h1:yYobGMfkprkpUuYxNetzaJUo7m1ikaYMkL0J5Z4lRlY=
-github.com/redhatinsights/app-common-go v1.6.1/go.mod h1:SqgG5JkX/RNlk2d+sXamIFxhOIvWLgCBr8uK6q70ESk=
-github.com/redhatinsights/platform-go-middlewares v0.14.0 h1:J7D/4ScOVknr74bfZaeN8zbNWfDhqdJw0IYAyG9xJhc=
-github.com/redhatinsights/platform-go-middlewares v0.14.0/go.mod h1:i5gVDZJ/quCQhs5AW5CwkRPXlz1HfDBvyNtXHnlXZfM=
-github.com/redhatinsights/platform-go-middlewares v0.15.0 h1:o7VBiX3BRN6WoyAQlKuPqgPqq4qx0Uzd7j+eaMGHz30=
-github.com/redhatinsights/platform-go-middlewares v0.15.0/go.mod h1:i5gVDZJ/quCQhs5AW5CwkRPXlz1HfDBvyNtXHnlXZfM=
-github.com/redhatinsights/platform-go-middlewares v0.16.0 h1:ba8jE/Ust6fnTD6PbM3fVyxL73AH2GTPjOQqIJb35eQ=
-github.com/redhatinsights/platform-go-middlewares v0.16.0/go.mod h1:i5gVDZJ/quCQhs5AW5CwkRPXlz1HfDBvyNtXHnlXZfM=
-github.com/redhatinsights/platform-go-middlewares v0.16.1 h1:ZWVv0CiAH3S8ir736PpeIJAnvngxILvphEvW2nBVOzc=
-github.com/redhatinsights/platform-go-middlewares v0.16.1/go.mod h1:i5gVDZJ/quCQhs5AW5CwkRPXlz1HfDBvyNtXHnlXZfM=
+github.com/redhatinsights/app-common-go v1.6.3-0.20220524114951-d700ed08bcc7 h1:qLexa9yG9dxQKcY6hvXqKGpcSQhapIqF2ylxGCZ0/2M=
+github.com/redhatinsights/app-common-go v1.6.3-0.20220524114951-d700ed08bcc7/go.mod h1:TPWBMiEi552jVKUQiOZ1NNbr/k0q7ZWuseqF0V10GRE=
 github.com/redhatinsights/platform-go-middlewares v0.17.0 h1:upjyS2Fq+yGio0N8TrZWha6ZzhidIkRROM0QnDqpiMk=
 github.com/redhatinsights/platform-go-middlewares v0.17.0/go.mod h1:i5gVDZJ/quCQhs5AW5CwkRPXlz1HfDBvyNtXHnlXZfM=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
@@ -673,8 +663,6 @@ golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220227234510-4e6760a101f9/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220429121018-84afa8d3f7b3 h1:kBsBifDikLCf5sUMbcD8p73OinDtAQWQp8+n7FiyzlA=
-golang.org/x/sys v0.0.0-20220429121018-84afa8d3f7b3/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220502124256-b6088ccd6cba h1:AyHWHCBVlIYI5rgEM3o+1PLd0sLPcIAoaUckGQMaWtw=
 golang.org/x/sys v0.0.0-20220502124256-b6088ccd6cba/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=

internal/config/config.go

@@ -33,21 +33,11 @@ type IngressConfig struct {
 }
 
 type KafkaCfg struct {
-	KafkaBrokers         []string
 	KafkaGroupID         string
 	KafkaTrackerTopic    string
 	KafkaDeliveryReports bool
 	KafkaAnnounceTopic   string
 	ValidTopics          []string
-	KafkaSSLConfig       KafkaSSLCfg
-}
-
-type KafkaSSLCfg struct {
-	KafkaCA       string
-	KafkaUsername string
-	KafkaPassword string
-	SASLMechanism string
-	Protocol      string
 }
 
 type StorageCfg struct {
@@ -182,7 +172,6 @@ func Get() *IngressConfig {
 		Debug:                options.GetBool("Debug"),
 		DebugUserAgent:       regexp.MustCompile(options.GetString("DebugUserAgent")),
 		KafkaConfig: KafkaCfg{
-			KafkaBrokers:         options.GetStringSlice("KafkaBrokers"),
 			KafkaGroupID:         options.GetString("KafkaGroupID"),
 			KafkaTrackerTopic:    options.GetString("KafkaTrackerTopic"),
 			KafkaDeliveryReports: options.GetBool("KafkaDeliveryReports"),
@@ -205,13 +194,5 @@ func Get() *IngressConfig {
 		},
 	}
 
-	if options.IsSet("KafkaUsername") {
-		IngressCfg.KafkaConfig.KafkaSSLConfig.KafkaUsername = options.GetString("KafkaUsername")
-		IngressCfg.KafkaConfig.KafkaSSLConfig.KafkaPassword = options.GetString("KafkaPassword")
-		IngressCfg.KafkaConfig.KafkaSSLConfig.SASLMechanism = options.GetString("SASLMechanism")
-		IngressCfg.KafkaConfig.KafkaSSLConfig.Protocol = options.GetString("Protocol")
-		IngressCfg.KafkaConfig.KafkaSSLConfig.KafkaCA = options.GetString("KafkaCA")
-	}
-
 	return IngressCfg
 }

internal/queue/queue.go

@@ -10,6 +10,7 @@ import (
 
 	prom "github.com/prometheus/client_golang/prometheus"
 	pa "github.com/prometheus/client_golang/prometheus/promauto"
+	clowder "github.com/redhatinsights/app-common-go/pkg/api/v1"
 )
 
 var (
@@ -34,13 +35,7 @@ var (
 // ProducerConfig configures a producer
 type ProducerConfig struct {
 	Topic                string
-	Brokers              []string
 	Async                bool
-	Username             string
-	Password             string
-	CA                   string
-	Protocol             string
-	SASLMechanism        string
 	KafkaDeliveryReports bool
 }
 
@@ -49,26 +44,7 @@ type ProducerConfig struct {
 // buffer has an opportunity to fill.
 func Producer(in chan validators.ValidationMessage, config *ProducerConfig) {
 
-	var configMap kafka.ConfigMap
-
-	if config.SASLMechanism != "" {
-		configMap = kafka.ConfigMap{
-			"bootstrap.servers":   config.Brokers[0],
-			"security.protocol":   config.Protocol,
-			"sasl.mechanism":      config.SASLMechanism,
-			"ssl.ca.location":     config.CA,
-			"sasl.username":       config.Username,
-			"sasl.password":       config.Password,
-			"go.delivery.reports": config.KafkaDeliveryReports,
-		}
-	} else {
-		configMap = kafka.ConfigMap{
-			"bootstrap.servers":   config.Brokers[0],
-			"go.delivery.reports": config.KafkaDeliveryReports,
-		}
-	}
-
-	p, err := kafka.NewProducer(&configMap)
+	p, err := clowder.NewDefaultKafkaProducer()
 
 	if err != nil {
 		l.Log.WithFields(logrus.Fields{"error": err}).Error("Error creating kafka producer")

internal/validators/kafka/kafka.go

@@ -23,49 +23,22 @@ func init() {
 // Validator posts requests to topics for validation
 type Validator struct {
 	ValidationProducerMapping map[string]chan validators.ValidationMessage
-	KafkaBrokers              []string
 	KafkaGroupID              string
-	Username                  string
-	Password                  string
-	CA                        string
-	SASLMechanism             string
-	Protocol                  string
 }
 
 // Config configures a new Kafka Validator
 type Config struct {
-	Brokers         []string
 	GroupID         string
 	ValidationTopic string
-	Username        string
-	Password        string
-	CA              string
-	Protocol        string
-	SASLMechanism   string
 }
 
 // New constructs and initializes a new Kafka Validator
 func New(cfg *Config, topics ...string) *Validator {
 	kv := &Validator{
 		ValidationProducerMapping: make(map[string]chan validators.ValidationMessage),
-		KafkaBrokers:              cfg.Brokers,
 		KafkaGroupID:              cfg.GroupID,
 	}
 
-	if cfg.CA != "" {
-		kv.CA = cfg.CA
-	}
-
-	if cfg.Username != "" {
-		kv.Username = cfg.Username
-		kv.Password = cfg.Password
-	}
-
-	if cfg.SASLMechanism != "" {
-		kv.SASLMechanism = cfg.SASLMechanism
-		kv.Protocol = cfg.Protocol
-	}
-
 	// ensure the announce topic is added and valid
 	topics = append(topics, "announce")
 
@@ -106,13 +79,7 @@ func (kv *Validator) Validate(vr *validators.Request) {
 func (kv *Validator) addProducer(topic string) {
 	ch := make(chan validators.ValidationMessage, 100)
 	go queue.Producer(ch, &queue.ProducerConfig{
-		Brokers:       kv.KafkaBrokers,
-		Topic:         topic,
-		CA:            kv.CA,
-		Username:      kv.Username,
-		Password:      kv.Password,
-		Protocol:      kv.Protocol,
-		SASLMechanism: kv.SASLMechanism,
+		Topic: topic,
 	})
 	kv.ValidationProducerMapping[topic] = ch
 }