RHINENG-22325: configure topic for Inventory Views

Author: MichaelMraka

base/utils/config.go

@@ -66,6 +66,7 @@ type coreConfig struct {
 	RemediationUpdateTopic string
 	NotificationsTopic     string
 	TemplateTopic          string
+	InventoryViewsTopic    string
 
 	// services
 	VmaasAddress                  string
@@ -149,6 +150,11 @@ func initDBFromEnv() {
 }
 
 func initKafkaFromEnv() {
+	overrideKafkaAddress := Getenv("KAFKA_ADDRESS", "")
+	if overrideKafkaAddress != "" {
+		CoreCfg.KafkaAddress = overrideKafkaAddress
+		CoreCfg.KafkaServers = []string{overrideKafkaAddress}
+	}
 	CoreCfg.KafkaSslCert = Getenv("KAFKA_SSL_CERT", CoreCfg.KafkaSslCert)
 	CoreCfg.KafkaSslSkipVerify = GetBoolEnvOrDefault("KAFKA_SSL_SKIP_VERIFY", false)
 	CoreCfg.KafkaUsername = Getenv("KAFKA_USERNAME", CoreCfg.KafkaUsername)
@@ -167,6 +173,7 @@ func initTopicsFromEnv() {
 	CoreCfg.RemediationUpdateTopic = Getenv("REMEDIATIONS_UPDATE_TOPIC", "")
 	CoreCfg.NotificationsTopic = Getenv("NOTIFICATIONS_TOPIC", "")
 	CoreCfg.TemplateTopic = Getenv("TEMPLATE_TOPIC", "")
+	CoreCfg.InventoryViewsTopic = Getenv("INVENTORY_VIEWS_TOPIC", "")
 }
 
 func initServicesFromEnv() {
@@ -248,6 +255,7 @@ func initKafkaFromClowder() {
 		translateTopic(&CoreCfg.RemediationUpdateTopic)
 		translateTopic(&CoreCfg.NotificationsTopic)
 		translateTopic(&CoreCfg.TemplateTopic)
+		translateTopic(&CoreCfg.InventoryViewsTopic)
 	}
 }
 
@@ -399,6 +407,7 @@ func printKafkaParams() {
 	fmt.Printf("REMEDIATIONS_UPDATE_TOPIC=%s\n", CoreCfg.RemediationUpdateTopic)
 	fmt.Printf("NOTIFICATIONS_TOPIC=%s\n", CoreCfg.NotificationsTopic)
 	fmt.Printf("TEMPLATE_TOPIC=%s\n", CoreCfg.TemplateTopic)
+	fmt.Printf("INVENTORY_VIEWS_TOPIC=%s\n", CoreCfg.InventoryViewsTopic)
 }
 
 func printServicesParams() {

conf/common.env

@@ -17,6 +17,7 @@ NOTIFICATIONS_TOPIC=platform.notifications.ingress
 PAYLOAD_TRACKER_TOPIC=platform.payload-status
 REMEDIATIONS_UPDATE_TOPIC=platform.remediation-updates.patch
 TEMPLATE_TOPIC=platform.content-sources.template
+INVENTORY_VIEWS_TOPIC=platform.inventory.host-apps
 
 # If vmaas is running locally, its available here
 #VMAAS_ADDRESS=http://vmaas_webapp:8080

conf/local.env

@@ -16,13 +16,14 @@ DB_SSLROOTCERT=dev/database/secrets/pgca.crt
 VMAAS_ADDRESS=http://localhost:9001
 CANDLEPIN_ADDRESS=http://localhost:9001/candlepin
 
-#KAFKA_ADDRESS=localhost:29092
+KAFKA_ADDRESS=localhost:9092
 KAFKA_GROUP=patchman
 KAFKA_SSL_CERT=dev/kafka/secrets/ca.crt
 PAYLOAD_TRACKER_TOPIC=platform.payload-status
 EVENTS_TOPIC=platform.inventory.events
 EVAL_TOPIC=patchman.evaluator.user-evaluation
 TEMPLATE_TOPIC=platform.content-sources.template
+INVENTORY_VIEWS_TOPIC=platform.inventory.host-apps
 
 RBAC_ADDRESS=http://localhost:9001
 

deploy/clowdapp.yaml

@@ -205,6 +205,7 @@ objects:
         - {name: PAYLOAD_TRACKER_TOPIC, value: platform.payload-status}
         - {name: REMEDIATIONS_UPDATE_TOPIC, value: 'platform.remediation-updates.patch'}
         - {name: NOTIFICATIONS_TOPIC, value: 'platform.notifications.ingress'}
+        - {name: INVENTORY_VIEWS_TOPIC, value: '${INVENTORY_VIEWS_TOPIC}'}
         - {name: SSL_CERT_DIR, value: '${SSL_CERT_DIR}'}
         - {name: GOGC, value: '${GOGC}'}  # set garbage collection limit for go 1.18
         - {name: ENABLE_PROFILER, value: '${ENABLE_PROFILER_EVALUATOR_UPLOAD}'}
@@ -251,6 +252,7 @@ objects:
         - {name: PAYLOAD_TRACKER_TOPIC, value: platform.payload-status}
         - {name: REMEDIATIONS_UPDATE_TOPIC, value: 'platform.remediation-updates.patch'}
         - {name: NOTIFICATIONS_TOPIC, value: 'platform.notifications.ingress'}
+        - {name: INVENTORY_VIEWS_TOPIC, value: '${INVENTORY_VIEWS_TOPIC}'}
         - {name: SSL_CERT_DIR, value: '${SSL_CERT_DIR}'}
         - {name: GOGC, value: '${GOGC}'}  # set garbage collection limit for go 1.18
         - {name: ENABLE_PROFILER, value: '${ENABLE_PROFILER_EVALUATOR_RECALC}'}
@@ -297,6 +299,7 @@ objects:
         - {name: PAYLOAD_TRACKER_TOPIC, value: platform.payload-status}
         - {name: REMEDIATIONS_UPDATE_TOPIC, value: 'platform.remediation-updates.patch'}
         - {name: NOTIFICATIONS_TOPIC, value: 'platform.notifications.ingress'}
+        - {name: INVENTORY_VIEWS_TOPIC, value: '${INVENTORY_VIEWS_TOPIC}'}
         - {name: SSL_CERT_DIR, value: '${SSL_CERT_DIR}'}
         - {name: GOGC, value: '${GOGC}'}  # set garbage collection limit for go 1.18
         - {name: ENABLE_PROFILER, value: '${ENABLE_PROFILER_EVALUATOR_USER_EVALUATION}'}
@@ -518,6 +521,7 @@ objects:
 
     kafkaTopics:
     - {replicas: 3, partitions: 10, topicName: platform.inventory.events}
+    - {replicas: 3, partitions: 10, topicName: platform.inventory.host-apps}
     - {replicas: 3, partitions: 10, topicName: patchman.evaluator.upload}
     - {replicas: 3, partitions: 10, topicName: patchman.evaluator.recalc}
     - {replicas: 3, partitions: 8, topicName: platform.payload-status}
@@ -702,6 +706,7 @@ parameters:
 - {name: GOGC, value: '100'}
 - {name: GOMEMLIMIT_EVALUATOR, value: '2700MiB'} # set to 90% of the default memory limit (don't forget `B`)
 - {name: CONSOLEDOT_HOSTNAME, value: localhost}
+- {name: INVENTORY_VIEWS_TOPIC, value: 'platform.inventory.host-apps'}
 
 # Evaluator - upload
 - {name: REPLICAS_EVALUATOR_UPLOAD, value: '1'}

dev/kafka/setup.sh

@@ -4,11 +4,17 @@
 sleep 5
 
 # create topics with multiple partitions for scaling
-for topic in "platform.inventory.events" "patchman.evaluator.upload" \
-             "patchman.evaluator.recalc" "platform.remediation-updates.patch" "platform.notifications.ingress" \
-             "platform.payload-status" "test" \
-             "platform.content-sources.template" \
-             "patchman.evaluator.user-evaluation"
+for topic in \
+            "patchman.evaluator.recalc" \
+            "patchman.evaluator.upload" \
+            "patchman.evaluator.user-evaluation" \
+            "platform.content-sources.template" \
+            "platform.inventory.events" \
+            "platform.inventory.host-apps" \
+            "platform.notifications.ingress" \
+            "platform.payload-status" \
+            "platform.remediation-updates.patch" \
+            "test"
 do
     until /opt/kafka/bin/kafka-topics.sh --create --if-not-exists --topic $topic \
         --partitions 1 --bootstrap-server kafka:9092 --replication-factor 1; do