Fix crash on startup

cmd/flowlogs-pipeline/main_test.go

@@ -18,9 +18,16 @@
 package main
 
 import (
+	"encoding/json"
 	"os"
 	"os/exec"
 	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/transform/kubernetes"
 )
 
 func TestTheMain(t *testing.T) {
@@ -36,3 +43,30 @@ func TestTheMain(t *testing.T) {
 	}
 	t.Fatalf("process ran with err %v, want exit status 1", err)
 }
+
+func TestPipelineConfigSetup(t *testing.T) {
+	// Kube init mock
+	kdata := new(kubernetes.KubeDataMock)
+	kdata.On("InitFromConfig", "").Return(nil)
+	kubernetes.Data = kdata
+
+	js := `{
+    "PipeLine": "[{\"name\":\"grpc\"},{\"follows\":\"grpc\",\"name\":\"enrich\"},{\"follows\":\"enrich\",\"name\":\"loki\"},{\"follows\":\"enrich\",\"name\":\"prometheus\"}]",
+    "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"input\":\"SrcAddr\",\"output\":\"SrcK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstAddr\",\"output\":\"DstK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstPort\",\"output\":\"Service\",\"parameters\":\"Proto\",\"type\":\"add_service\"},{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"parameters\":\"/16\",\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"port\":9102,\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
+    "Health": {
+        "Port": "8080"
+    },
+    "Profile": {
+        "Port": 0
+    }
+}`
+	var opts config.Options
+	err := json.Unmarshal([]byte(js), &opts)
+	require.NoError(t, err)
+	cfg, err := config.ParseConfig(opts)
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+	mainPipeline, err := pipeline.NewPipeline(&cfg)
+	require.NoError(t, err)
+	require.NotNil(t, mainPipeline)
+}

pkg/api/transform_network.go

@@ -24,6 +24,18 @@ type TransformNetwork struct {
 	ProtocolsFile  string                `yaml:"protocolsFile,omitempty" json:"protocolsFile,omitempty" doc:"path to protocols file (optional, default: /etc/protocols)"`
 }
 
+func (tn *TransformNetwork) GetServiceFiles() (string, string) {
+	p := tn.ProtocolsFile
+	if p == "" {
+		p = "/etc/protocols"
+	}
+	s := tn.ServicesFile
+	if s == "" {
+		s = "/etc/services"
+	}
+	return p, s
+}
+
 type TransformNetworkOperationEnum struct {
 	AddRegExIf    string `yaml:"add_regex_if" json:"add_regex_if" doc:"add output field if input field satisfies regex pattern from parameters field"`
 	AddIf         string `yaml:"add_if" json:"add_if" doc:"add output field if input field satisfies criteria from parameters field"`

pkg/pipeline/transform/kubernetes/kubernetes-mock.go

@@ -0,0 +1,13 @@
+package kubernetes
+
+import "github.com/stretchr/testify/mock"
+
+type KubeDataMock struct {
+	mock.Mock
+	kubeDataInterface
+}
+
+func (o *KubeDataMock) InitFromConfig(kubeConfigPath string) error {
+	args := o.Called(kubeConfigPath)
+	return args.Error(0)
+}

pkg/pipeline/transform/kubernetes/kubernetes.go

@@ -36,7 +36,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 )
 
-var Data KubeData
+var Data kubeDataInterface = &KubeData{}
 
 const (
 	kubeConfigEnvVariable = "KUBECONFIG"
@@ -47,7 +47,13 @@ const (
 	typeService           = "Service"
 )
 
+type kubeDataInterface interface {
+	GetInfo(string) (*Info, error)
+	InitFromConfig(string) error
+}
+
 type KubeData struct {
+	kubeDataInterface
 	ipInformers        map[string]cache.SharedIndexInformer
 	replicaSetInformer cache.SharedIndexInformer
 	stopChan           chan struct{}

pkg/pipeline/transform/transform_network.go

@@ -189,15 +189,16 @@ func NewTransformNetwork(params config.StageParam) (Transformer, error) {
 
 	var servicesDB *netdb.ServiceNames
 	if needToInitNetworkServices {
+		pFilename, sFilename := jsonNetworkTransform.GetServiceFiles()
 		var err error
-		protos, err := os.Open(jsonNetworkTransform.ProtocolsFile)
+		protos, err := os.Open(pFilename)
 		if err != nil {
-			return nil, fmt.Errorf("opening %q: %w", jsonNetworkTransform.ProtocolsFile, err)
+			return nil, fmt.Errorf("opening protocols file %q: %w", pFilename, err)
 		}
 		defer protos.Close()
-		services, err := os.Open(jsonNetworkTransform.ServicesFile)
+		services, err := os.Open(sFilename)
 		if err != nil {
-			return nil, fmt.Errorf("opening %q: %w", jsonNetworkTransform.ServicesFile, err)
+			return nil, fmt.Errorf("opening services file %q: %w", sFilename, err)
 		}
 		defer services.Close()
 		servicesDB, err = netdb.LoadServicesDB(protos, services)