add mocks to cli

.mk/dev.mk

@@ -0,0 +1,11 @@
+##@ dev helpers
+
+.PHONY:	flows-mock
+flows-mock: build ## Run flow capture using mocks
+	./build/network-observability-cli get-flows --mock true
+	tput reset
+
+.PHONY:	packets-mock
+packets-mock: build ## Run packet capture using mocks
+	./build/network-observability-cli get-packets --mock true
+	tput reset

Makefile

@@ -239,4 +239,5 @@ endif
 help: ## Display this help.
 	@awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n  make \033[36m<target>\033[0m\n"} /^[a-zA-Z_0-9-]+:.*?##/ { printf "  \033[36m%-15s\033[0m %s\n", $$1, $$2 } /^##@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) } ' $(MAKEFILE_LIST)
 
+include .mk/dev.mk
 include .mk/shortcuts.mk

cmd/flow_capture.go

@@ -103,6 +103,7 @@ func runFlowCaptureOnAddr(port int, filename string) {
 		log.Fatal(err)
 	}
 	log.Trace("Started collector")
+	collectorStarted = true
 
 	go func() {
 		<-utils.ExitChannel()

cmd/mocks.go

@@ -0,0 +1,66 @@
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"time"
+
+	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/write/grpc"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/write/grpc/genericmap"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+func sendMock(ch chan struct{}, client genericmap.CollectorClient) {
+
+	gm := config.GenericMap{
+		"DstAddr":          fmt.Sprintf("10.0.0.%d", rand.Intn(255)),
+		"DstPort":          rand.Intn(9999),
+		"Proto":            rand.Intn(255),
+		"SrcAddr":          fmt.Sprintf("10.0.0.%d", rand.Intn(255)),
+		"SrcPort":          rand.Intn(9999),
+		"SrcK8S_Name":      fmt.Sprintf("Pod-%d", rand.Intn(100)),
+		"SrcK8S_Namespace": fmt.Sprintf("Namespace-%d", rand.Intn(100)),
+		"SrcK8S_Type":      "Pod",
+		"TimeFlowEndMs":    time.Now().UnixMilli(),
+	}
+	value, err := json.Marshal(gm)
+	if err != nil {
+		log.Error(err)
+		return
+	}
+	_, err = client.Send(context.Background(), &genericmap.Flow{
+		GenericMap: &anypb.Any{
+			Value: value,
+		},
+	})
+	if err != nil {
+		log.Error(err)
+		return
+	}
+
+	time.Sleep(10 * time.Millisecond)
+	ch <- struct{}{}
+}
+
+func MockForever() {
+	wait := make(chan struct{})
+
+	for !collectorStarted {
+		log.Info("Waiting for collector to start...")
+		time.Sleep(1 * time.Second)
+	}
+
+	for _, port := range ports {
+		cc, err := grpc.ConnectClient("127.0.0.1", port)
+		if err != nil {
+			log.Fatal(err)
+		}
+		for {
+			go sendMock(wait, cc.Client())
+			<-wait
+		}
+	}
+}

cmd/packet_capture.go

@@ -86,6 +86,7 @@ func runPacketCaptureOnAddr(port int, filename string) {
 		log.Fatal(err)
 	}
 	log.Trace("Started collector")
+	collectorStarted = true
 
 	go func() {
 		<-utils.ExitChannel()

cmd/root.go

@@ -53,11 +53,13 @@ var (
 		},
 	}
 
-	captureType    = "Flow"
-	outputBuffer   *bytes.Buffer
-	captureStarted = false
-	stopReceived   = false
-	keyboardError  = ""
+	captureType      = "Flow"
+	outputBuffer     *bytes.Buffer
+	collectorStarted = false
+	captureStarted   = false
+	stopReceived     = false
+	useMocks         = false
+	keyboardError    = ""
 )
 
 // Execute executes the root command.
@@ -67,13 +69,14 @@ func Execute() error {
 
 // func main() {
 func init() {
-	cobra.OnInitialize(initConfig)
+	cobra.OnInitialize(onInit)
 	rootCmd.PersistentFlags().StringVarP(&logLevel, "loglevel", "l", "info", "Log level")
 	rootCmd.PersistentFlags().IntSliceVarP(&ports, "ports", "", []int{9999}, "TCP ports to listen")
 	rootCmd.PersistentFlags().StringSliceVarP(&nodes, "nodes", "", []string{""}, "Node names per port (optionnal)")
 	rootCmd.PersistentFlags().StringVarP(&filter, "filter", "", "", "Filter(s)")
 	rootCmd.PersistentFlags().DurationVarP(&maxTime, "maxtime", "", 5*time.Minute, "Maximum capture time")
 	rootCmd.PersistentFlags().Int64VarP(&maxBytes, "maxbytes", "", 50000000, "Maximum capture bytes")
+	rootCmd.PersistentFlags().BoolVarP(&useMocks, "mock", "", false, "Use mock")
 
 	c := make(chan os.Signal, 1)
 	signal.Notify(c, syscall.SIGTERM)
@@ -92,7 +95,7 @@ func init() {
 	rootCmd.AddCommand(pktCmd)
 }
 
-func initConfig() {
+func onInit() {
 	lvl, _ := logrus.ParseLevel(logLevel)
 	log.SetLevel(lvl)
 
@@ -102,6 +105,11 @@ func initConfig() {
 
 	log.Infof("Running network-observability-cli\nLog level: %s\nFilter(s): %s", logLevel, filter)
 	showKernelVersion()
+
+	if useMocks {
+		log.Info("Using mocks...")
+		go MockForever()
+	}
 }
 
 func showKernelVersion() {

go.mod

@@ -80,7 +80,7 @@ require (
 	golang.org/x/time v0.5.0 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240528184218-531527333157 // indirect
 	google.golang.org/grpc v1.65.0 // indirect
-	google.golang.org/protobuf v1.34.2 // indirect
+	google.golang.org/protobuf v1.34.2
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect