add ovn model utilities (#481)

pkg/agent/agent.go

@@ -178,6 +178,8 @@ func FlowsAgent(cfg *Config) (*Flows, error) {
 			if s, err = ovnobserv.NewSampleDecoderWithDefaultCollector(context.Background(), networkEventsDBPath,
 				networkEventsOwnerName, cfg.NetworkEventsMonitoringGroupID); err != nil {
 				alog.Warnf("failed to create Network Events sample decoder: %v for id: %d", err, cfg.NetworkEventsMonitoringGroupID)
+			} else {
+				alog.Info("Network Events sample decoder successfully created")
 			}
 		} else {
 			alog.Warn("old kernel doesn't support network events monitoring skip")

pkg/model/record.go

@@ -9,8 +9,8 @@ import (
 	"time"
 
 	"github.com/netobserv/netobserv-ebpf-agent/pkg/ebpf"
+	"github.com/netobserv/netobserv-ebpf-agent/pkg/utils"
 
-	ovnmodel "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/model"
 	ovnobserv "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/sampledecoder"
 	"github.com/sirupsen/logrus"
 )
@@ -114,26 +114,11 @@ func NewRecord(
 		record.NetworkMonitorEventsMD = make([]map[string]string, 0)
 		for _, metadata := range metrics.AdditionalMetrics.NetworkEvents {
 			if !AllZerosMetaData(metadata) {
-				var cm map[string]string
 				if md, err := s.DecodeCookie8Bytes(metadata); err == nil {
-					acl, ok := md.(*ovnmodel.ACLEvent)
 					mdStr := md.String()
 					if !seen[mdStr] {
-						if ok {
-							cm = map[string]string{
-								"Action":    acl.Action,
-								"Type":      acl.Actor,
-								"Feature":   "acl",
-								"Name":      acl.Name,
-								"Namespace": acl.Namespace,
-								"Direction": acl.Direction,
-							}
-						} else {
-							cm = map[string]string{
-								"Message": mdStr,
-							}
-						}
-						record.NetworkMonitorEventsMD = append(record.NetworkMonitorEventsMD, cm)
+						asMap := utils.NetworkEventToMap(md)
+						record.NetworkMonitorEventsMD = append(record.NetworkMonitorEventsMD, asMap)
 						seen[mdStr] = true
 					}
 				}

pkg/utils/network_events.go

+package utils
+
+import (
+	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	ovnmodel "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/model"
+)
+
+func NetworkEventToMap(netev ovnmodel.NetworkEvent) map[string]string {
+	if acl, ok := netev.(*ovnmodel.ACLEvent); ok {
+		return map[string]string{
+			"Action":    acl.Action,
+			"Type":      acl.Actor,
+			"Feature":   "acl",
+			"Name":      acl.Name,
+			"Namespace": acl.Namespace,
+			"Direction": acl.Direction,
+		}
+	}
+	return map[string]string{
+		"Message": netev.String(),
+	}
+}
+
+func NetworkEventsToStrings(flow config.GenericMap) []string {
+	if ne, found := flow["NetworkEvents"]; found {
+		if neList, isList := ne.([]any); isList {
+			var messages []string
+			for _, item := range neList {
+				if neItem, isMap := item.(map[string]any); isMap {
+					messages = append(messages, networkEventItemToString(neItem))
+				}
+			}
+			return messages
+		}
+	}
+	return nil
+}
+
+func networkEventItemToString(in map[string]any) string {
+	if msg := getAsString(in, "Message"); msg != "" {
+		return msg
+	}
+	if feat := getAsString(in, "Feature"); feat == "acl" {
+		aclObj := ovnmodel.ACLEvent{
+			Action:    getAsString(in, "Action"),
+			Actor:     getAsString(in, "Type"),
+			Name:      getAsString(in, "Name"),
+			Namespace: getAsString(in, "Namespace"),
+			Direction: getAsString(in, "Direction"),
+		}
+		return aclObj.String()
+	}
+	return ""
+}
+
+func getAsString(in map[string]any, key string) string {
+	if anyV, hasKey := in[key]; hasKey {
+		if v, isStr := anyV.(string); isStr {
+			return v
+		}
+	}
+	return ""
+}