NETOBSERV-1974: use ovnk-lib to get net event messages (#647)

* get net events messages - web reimpl

* add ovnk deps

* Use ovnk-lib to get event string

* Revert "get net events messages - web reimpl"

This reverts commit 8efee4c.

* fix test and feature name

* use released ovnk

* fix test

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,

go.mod

@@ -8,23 +8,32 @@ require (
 	github.com/gorilla/mux v1.8.1
 	github.com/json-iterator/go v1.1.12
 	github.com/mitchellh/mapstructure v1.5.0
+	github.com/ovn-org/ovn-kubernetes/go-controller v0.0.0-20241126140656-c95491e46334
 	github.com/prometheus/client_golang v1.20.2
 	github.com/prometheus/common v0.57.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.9.0
 	gopkg.in/yaml.v3 v3.0.1
-	k8s.io/api v0.31.0
-	k8s.io/apimachinery v0.31.0
-	k8s.io/client-go v0.31.0
+	k8s.io/api v0.31.1
+	k8s.io/apimachinery v0.31.1
+	k8s.io/client-go v0.31.1
 )
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
+	github.com/cenkalti/hub v1.0.1 // indirect
+	github.com/cenkalti/rpc2 v0.0.0-20210604223624-c1acbc6ec984 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/containernetworking/cni v1.1.2 // indirect
+	github.com/containernetworking/plugins v1.2.0 // indirect
+	github.com/coreos/go-iptables v0.6.0 // indirect
+	github.com/cpuguy83/go-md2man/v2 v2.0.4 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.4 // indirect
@@ -42,19 +51,29 @@ require (
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f // indirect
+	github.com/ovn-org/libovsdb v0.7.1-0.20240820095311-ce1951614a20 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/russross/blackfriday/v2 v2.1.0 // indirect
+	github.com/safchain/ethtool v0.3.1-0.20231027162144-83e5e0097c91 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
+	github.com/urfave/cli/v2 v2.27.2 // indirect
+	github.com/vishvananda/netlink v1.2.1-beta.2.0.20231024175852-77df5d35f725 // indirect
+	github.com/vishvananda/netns v0.0.4 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
+	github.com/xrash/smetrics v0.0.0-20240312152122-5f08fbb34913 // indirect
 	golang.org/x/net v0.27.0 // indirect
 	golang.org/x/oauth2 v0.21.0 // indirect
 	golang.org/x/sys v0.22.0 // indirect
 	golang.org/x/term v0.22.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
+	gopkg.in/gcfg.v1 v1.2.3 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
+	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
+	gopkg.in/warnings.v0 v0.1.2 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 // indirect

go.sum

@@ -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.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/decoders/network-events_test.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.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": [
+          [
+            "1731930300000000000",
+            "{\"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.Unix(1731930300, 0),
+				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,