Use ovnk-lib to get event string

Author: jotak

cmd/plugin-backend.go

@@ -5,10 +5,13 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"slices"
 
 	"github.com/sirupsen/logrus"
 
 	"github.com/netobserv/network-observability-console-plugin/pkg/config"
+	"github.com/netobserv/network-observability-console-plugin/pkg/decoders"
+	"github.com/netobserv/network-observability-console-plugin/pkg/model"
 	"github.com/netobserv/network-observability-console-plugin/pkg/server"
 )
 
@@ -54,6 +57,11 @@ func main() {
 		log.WithError(err).Fatal("auth checker error")
 	}
 
+	if slices.Contains(cfg.Frontend.Features, "NetworkEvents") {
+		// Add decoder hook
+		model.AddFlowLineMapping(decoders.NetworkEventsToString)
+	}
+
 	go server.StartMetrics(&server.MetricsConfig{
 		Port:     cfg.Server.MetricsPort,
 		CertPath: cfg.Server.CertPath,

pkg/decoders/network-events.go

@@ -0,0 +1,63 @@
+package decoders
+
+import (
+	"encoding/json"
+
+	"github.com/netobserv/network-observability-console-plugin/pkg/model/fields"
+	ovnmodel "github.com/ovn-org/ovn-kubernetes/go-controller/observability-lib/model"
+	"github.com/sirupsen/logrus"
+)
+
+var dlog = logrus.WithField("module", "decoders")
+
+func NetworkEventsToString(in string) string {
+	line := make(map[string]any)
+	if err := json.Unmarshal([]byte(in), &line); err != nil {
+		dlog.Errorf("Could not decode NetworkEvent: %v", err)
+		return in
+	}
+	if ne, found := line[fields.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))
+				}
+			}
+			line[fields.NetworkEvents] = messages
+			b, err := json.Marshal(line)
+			if err != nil {
+				dlog.Errorf("Could not reencode NetworkEvent: %v", err)
+				return in
+			}
+			return string(b)
+		}
+	}
+	return in
+}
+
+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 ""
+}

pkg/decoders/network-events_test.go

@@ -0,0 +1,31 @@
+package decoders
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestReencode_NoChange(t *testing.T) {
+	js := `{"SrcK8S_Name":"ip-10-0-1-7.ec2.internal","Bytes":66,"Packets":1,"Interfaces":["br-ex"]}`
+	out := NetworkEventsToString(js)
+	assert.Equal(t, js, out)
+}
+
+func TestReencode_UpdateEvent(t *testing.T) {
+	js := `{"SrcK8S_Name":"ip-10-0-1-7.ec2.internal","Bytes":66,"Packets":1,"Interfaces":["br-ex"],"NetworkEvents":[{"Feature":"acl","Type":"NetpolNode","Action":"allow","Direction":"Ingress"}]}`
+	out := NetworkEventsToString(js)
+	assert.Equal(
+		t,
+		`{"Bytes":66,"Interfaces":["br-ex"],"NetworkEvents":["Allowed by default allow from local node policy, direction ingress"],"Packets":1,"SrcK8S_Name":"ip-10-0-1-7.ec2.internal"}`,
+		out,
+	)
+
+	js = `{"SrcK8S_Name":"ip-10-0-1-7.ec2.internal","Bytes":66,"Packets":1,"Interfaces":["br-ex"],"NetworkEvents":[{"Message":"custom message"}]}`
+	out = NetworkEventsToString(js)
+	assert.Equal(
+		t,
+		`{"Bytes":66,"Interfaces":["br-ex"],"NetworkEvents":["custom message"],"Packets":1,"SrcK8S_Name":"ip-10-0-1-7.ec2.internal"}`,
+		out,
+	)
+}

pkg/model/loki.go

@@ -10,6 +10,16 @@ import (
 	"github.com/prometheus/common/model"
 )
 
+type FlowLineMapping = func(string) string
+
+var (
+	flowLineMappings = []FlowLineMapping{}
+)
+
+func AddFlowLineMapping(f FlowLineMapping) {
+	flowLineMappings = append(flowLineMappings, f)
+}
+
 // QueryResponse represents the http json response to a logQL query
 type QueryResponse struct {
 	Status string            `json:"status"`
@@ -128,6 +138,13 @@ func unmarshalQueryResponseData(data []byte) (ResultType, ResultValue, interface
 	case ResultTypeStream:
 		var s Streams
 		err = json.Unmarshal(unmarshal.Result, &s)
+		for _, mapping := range flowLineMappings {
+			for i := range s {
+				for ii := range s[i].Entries {
+					s[i].Entries[ii].Line = mapping(s[i].Entries[ii].Line)
+				}
+			}
+		}
 		value = s
 	case ResultTypeMatrix:
 		var m Matrix

pkg/model/loki_test.go

@@ -2,7 +2,9 @@ package model
 
 import (
 	"encoding/json"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/netobserv/network-observability-console-plugin/pkg/utils/constants"
 	"github.com/stretchr/testify/assert"
@@ -34,6 +36,57 @@ func TestQueryResponseUnmarshal(t *testing.T) {
 	assert.IsType(t, expType, qr.Data.Result)
 }
 
+func TestQueryResponseUnmarshalLineMapping(t *testing.T) {
+	AddFlowLineMapping(func(in string) string {
+		return strings.ReplaceAll(in, "Bytes", "Bytezz")
+	})
+	js := `
+{
+  "status": "success",
+  "data": {
+    "resultType": "streams",
+    "result": [
+      {
+        "stream": {
+          "app": "netobserv-flowcollector"
+        },
+        "values": [
+          [
+            "1731926700000000000",
+            "{\"SrcK8S_Name\":\"ip-10-0-1-7.ec2.internal\",\"Bytes\":66,\"Packets\":1,\"Interfaces\":[\"br-ex\"]}"
+          ]
+        ]
+      }
+    ],
+    "stats": {
+      "summary": {}
+    }
+  }
+}
+	`
+
+	var qr QueryResponse
+	err := json.Unmarshal([]byte(js), &qr)
+	require.NoError(t, err)
+	assert.Equal(t, ResultTypeStream, string(qr.Data.ResultType))
+	assert.NotNil(t, qr.Data.Result)
+	var expType Streams
+	assert.IsType(t, expType, qr.Data.Result)
+	data := qr.Data.Result.(Streams)
+	assert.Len(t, data, 1)
+	assert.Equal(t, Stream{
+		Labels: map[string]string{
+			"app": "netobserv-flowcollector",
+		},
+		Entries: []Entry{
+			{
+				Timestamp: time.Date(2024, time.November, 18, 11, 45, 0, 0, time.Local),
+				Line:      `{"SrcK8S_Name":"ip-10-0-1-7.ec2.internal","Bytezz":66,"Packets":1,"Interfaces":["br-ex"]}`,
+			},
+		},
+	}, data[0])
+}
+
 func TestAggregatedQueryResponseMarshal(t *testing.T) {
 	qr := AggregatedQueryResponse{
 		ResultType: ResultTypeStream,