RHINENG-22325: sent inventory view event after evaluation

Author: MichaelMraka

base/inventory_views/inventory_views.go

@@ -3,6 +3,7 @@ package inventory_views
 import (
 	"app/base/models"
 	"app/base/utils"
+	"time"
 
 	"gorm.io/gorm"
 )
@@ -35,6 +36,16 @@ type InventoryViewsEvent struct {
 	Hosts     []InventoryViewsHost `json:"hosts"`
 }
 
+func MakeInventoryViewsEvent(tx *gorm.DB, orgID string, systems []models.SystemPlatform) (
+	InventoryViewsEvent, error) {
+	templates, err := FindSystemsTemplates(tx, systems)
+	if err != nil {
+		return InventoryViewsEvent{}, err
+	}
+	hosts := MakeInventoryViewsHosts(systems, templates)
+	return InventoryViewsEvent{OrgID: orgID, Timestamp: time.Now().Format(time.RFC3339), Hosts: hosts}, nil
+}
+
 func MakeInventoryViewsHosts(systems []models.SystemPlatform,
 	templates map[int64]models.TemplateBase) []InventoryViewsHost {
 	hosts := make([]InventoryViewsHost, len(systems))

evaluator/evaluate.go

@@ -59,6 +59,7 @@ var (
 	enableYumUpdatesEval          bool
 	nEvalGoroutines               int
 	enableInstantNotifications    bool
+	enableInventoryViews          bool
 	enableSatelliteFunctionality  bool
 	errVmaasBadRequest            = errors.New("vmaas bad request")
 )
@@ -79,6 +80,7 @@ func configure() {
 	vmaasUpdatesURL = utils.FailIfEmpty(utils.CoreCfg.VmaasAddress, "VMAAS_ADDRESS") + base.VMaaSAPIPrefix + "/updates"
 	configureRemediations()
 	configureNotifications()
+	configureInventoryViews()
 	configureStatus()
 }
 
@@ -126,6 +128,8 @@ func confugureEvaluator() {
 	nEvalGoroutines = utils.PodConfig.GetInt("max_goroutines", 1)
 	// Send advisory notification immediately
 	enableInstantNotifications = utils.PodConfig.GetBool("instant_notifications", true)
+	// Send inventory views events
+	enableInventoryViews = utils.PodConfig.GetBool("inventory_views", false)
 	// Ignore templates for satellite managed systems
 	enableSatelliteFunctionality = utils.PodConfig.GetBool("satellite_functionality", true)
 }
@@ -481,6 +485,15 @@ func evaluateAndStore(system *models.SystemPlatform,
 		return errors.Wrap(err, "Unable to update system")
 	}
 
+	if enableInventoryViews {
+		err = publishInventoryViewsEvent(tx, []models.SystemPlatform{*system}, event)
+		if err != nil {
+			evaluationCnt.WithLabelValues("error-inventory-views-publish").Inc()
+			utils.LogError("orgID", event.GetOrgID(), "inventoryID", system.GetInventoryID(), "err", err.Error(),
+				"publishing inventory views event failed")
+		}
+	}
+
 	// Send instant notification with new advisories
 	if enableInstantNotifications {
 		err = publishNewAdvisoriesNotification(tx, system, event.GetOrgID(), systemAdvisoriesNew)

evaluator/inventory_views.go

@@ -0,0 +1,70 @@
+package evaluator
+
+import (
+	"app/base"
+	"app/base/inventory_views"
+	"app/base/models"
+	"app/base/mqueue"
+	"app/base/utils"
+	"fmt"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/segmentio/kafka-go"
+	"gorm.io/gorm"
+)
+
+var patchAppHeader = kafka.Header{Key: "application", Value: []byte("patch")}
+
+var inventoryViewsPublisher mqueue.Writer
+
+func configureInventoryViews() {
+	if topic := utils.CoreCfg.InventoryViewsTopic; topic != "" {
+		inventoryViewsPublisher = mqueue.NewKafkaWriterFromEnv(topic)
+	}
+}
+
+func publishInventoryViewsEvent(tx *gorm.DB, systems []models.SystemPlatform, origin *mqueue.PlatformEvent) error {
+	if inventoryViewsPublisher == nil {
+		return nil
+	}
+
+	tStart := time.Now()
+	defer utils.ObserveSecondsSince(tStart, evaluationPartDuration.WithLabelValues("inventory-views-publish"))
+
+	orgID := origin.GetOrgID()
+	var requestID string
+	if len(origin.RequestIDs) > 0 {
+		requestID = origin.RequestIDs[0]
+	} else {
+		requestID = fmt.Sprintf("patch-%d", time.Now().UnixNano())
+	}
+
+	value, err := inventory_views.MakeInventoryViewsEvent(tx, orgID, systems)
+	if err != nil {
+		return errors.Wrap(err, "creating inventory views event failed")
+	}
+
+	headers := []kafka.Header{patchAppHeader, {Key: "request_id", Value: []byte(requestID)}}
+
+	msg, err := mqueue.MessageFromJSON(orgID, value, headers)
+	if err != nil {
+		return errors.Wrap(err, "creating inventory views message failed")
+	}
+
+	err = inventoryViewsPublisher.WriteMessages(base.Context, msg)
+	if err != nil {
+		return errors.Wrap(err, "writing message to inventory views publisher failed")
+	}
+
+	// log the event
+	systemIDs := make([]int64, len(systems))
+	for i, s := range systems {
+		systemIDs[i] = s.ID
+	}
+
+	utils.LogInfo("rh_account_ID", systems[0].RhAccountID, "systemIDs", systemIDs,
+		"inventory views event sent successfully")
+
+	return nil
+}