NETOBSERV-1471 capture using gRPC

.gitignore

@@ -1,2 +1,3 @@
 build
 output
+tmp

.mk/shortcuts.mk

@@ -0,0 +1,19 @@
+##@ shortcuts helpers
+
+.PHONY: build
+build: prereqs fmt lint vendors compile ## Test and Build cli
+
+.PHONY: build-image
+build-image: image-build ## Build MULTIARCH_TARGETS images
+
+.PHONY: push-image
+push-image: image-push ## Push MULTIARCH_TARGETS images
+
+.PHONY: build-manifest
+build-manifest: manifest-build ## Build MULTIARCH_TARGETS manifest
+
+.PHONY: push-manifest
+push-manifest: manifest-push ## Push MULTIARCH_TARGETS manifest
+
+.PHONY: images
+images: image-build image-push manifest-build manifest-push ## Build and push MULTIARCH_TARGETS images and related manifest

Dockerfile

@@ -0,0 +1,34 @@
+# We do not use --platform feature to auto fill this ARG because of incompatibility between podman and docker
+ARG TARGETPLATFORM=linux/amd64
+ARG BUILDPLATFORM=linux/amd64
+ARG TARGETARCH=amd64
+
+# Build the manager binary
+FROM --platform=$BUILDPLATFORM docker.io/library/golang:1.21 as builder
+
+ARG TARGETARCH
+ARG TARGETPLATFORM
+ARG VERSION="unknown"
+
+WORKDIR /opt/app-root
+
+COPY cmd cmd
+COPY main.go main.go
+COPY go.mod go.mod
+COPY go.sum go.sum
+COPY vendor/ vendor/
+COPY Makefile Makefile
+COPY .mk/ .mk/
+
+# Build
+RUN GOARCH=$TARGETARCH make compile
+
+# Create final image from ubi + built binary
+FROM --platform=$TARGETPLATFORM registry.access.redhat.com/ubi9/ubi:9.3
+WORKDIR /
+COPY --from=builder /opt/app-root/build .
+RUN mkdir output
+RUN chown 65532 output
+USER 65532:65532
+
+ENTRYPOINT ["/network-observability-cli"]

Makefile

@@ -1,20 +1,71 @@
+# VERSION defines the project version for the bundle.
+# Update this value when you upgrade the version of your project.
+# To re-generate a bundle for another specific version without changing the standard setup, you can:
+# - use the VERSION as arg of the bundle target (e.g make bundle VERSION=0.0.2)
+# - use environment variables to overwrite this value (e.g export VERSION=0.0.2)
+VERSION ?= main
+BUILD_DATE := $(shell date +%Y-%m-%d\ %H:%M)
+TAG_COMMIT := $(shell git rev-list --abbrev-commit --tags --max-count=1)
+TAG := $(shell git describe --abbrev=0 --tags ${TAG_COMMIT} 2>/dev/null || true)
+BUILD_SHA := $(shell git rev-parse --short HEAD)
+BUILD_VERSION := $(TAG:v%=%)
+ifneq ($(COMMIT), $(TAG_COMMIT))
+	BUILD_VERSION := $(BUILD_VERSION)-$(BUILD_SHA)
+endif
+ifneq ($(shell git status --porcelain),)
+	BUILD_VERSION := $(BUILD_VERSION)-dirty
+endif
+
+# Go architecture and targets images to build
+GOARCH ?= amd64
+MULTIARCH_TARGETS ?= amd64
+
+# In CI, to be replaced by `netobserv`
+IMAGE_ORG ?= $(USER)
+
+# Build output 
 NAME := network-observability-cli
 DIST_DIR ?= build
 FILES_OUTPUT_DIR ?= output
 OUTPUT := $(DIST_DIR)/$(NAME)
 
+# IMAGE_TAG_BASE defines the namespace and part of the image name for remote images.
+IMAGE_TAG_BASE ?= quay.io/$(IMAGE_ORG)/$(NAME)
+
+# Image URL to use all building/pushing image targets
+IMAGE ?= $(IMAGE_TAG_BASE):$(VERSION)
+OCI_BUILD_OPTS ?=
+
 # Image building tool (docker / podman) - docker is preferred in CI
-OCI_BIN_PATH = $(shell which docker 2>/dev/null || which podman)
+OCI_BIN_PATH := $(shell which docker 2>/dev/null || which podman)
 OCI_BIN ?= $(shell basename ${OCI_BIN_PATH})
 
 GOLANGCI_LINT_VERSION = v1.53.3
 
-.PHONY: all
-all: build
+# build a single arch target provided as argument
+define build_target
+	echo 'building image for arch $(1)'; \
+	DOCKER_BUILDKIT=1 $(OCI_BIN) buildx build --load --build-arg TARGETPLATFORM=linux/$(1) --build-arg TARGETARCH=$(1) --build-arg BUILDPLATFORM=linux/amd64 ${OCI_BUILD_OPTS} -t ${IMAGE}-$(1) -f Dockerfile .;
+endef
+
+# push a single arch target image
+define push_target
+	echo 'pushing image ${IMAGE}-$(1)'; \
+	DOCKER_BUILDKIT=1 $(OCI_BIN) push ${IMAGE}-$(1);
+endef
+
+# manifest create a single arch target provided as argument
+define manifest_add_target
+	echo 'manifest add target $(1)'; \
+	DOCKER_BUILDKIT=1 $(OCI_BIN) manifest add ${IMAGE} ${IMAGE}-$(1);
+endef
+
+##@ General
 
 .PHONY: prepare
 prepare:
 	@mkdir -p $(DIST_DIR)
+	mkdir -p tmp
 
 .PHONY: prereqs
 prereqs: ## Test if prerequisites are met, and installing missing dependencies
@@ -26,15 +77,17 @@ vendors: ## Refresh vendors directory.
 	@echo "### Checking vendors"
 	go mod tidy && go mod vendor
 
-.PHONY: build
-build: prepare lint ## Build the binary and run Lint
-	@go build -mod vendor -o $(OUTPUT)
-	cp -a ./oc/. ./$(DIST_DIR)
-	cp -a ./res/. ./$(DIST_DIR)/network-observability-cli-resources
+##@ Develop
 
-.PHONY: image
-image: ## Build the docker image
-	$(OCI_BIN) build -t network-observability-cli .
+.PHONY: compile
+compile: ## Build the binary
+	@echo "### Compiling project"
+	GOARCH=${GOARCH} go build -ldflags "-X main.version=${VERSION} -X 'main.buildVersion=${BUILD_VERSION}' -X 'main.buildDate=${BUILD_DATE}'" -mod vendor -a -o $(OUTPUT)
+
+.PHONY: fmt
+fmt: ## Run go fmt against code.
+	@echo "### Formatting code"
+	go fmt ./...
 
 .PHONY: lint
 lint: prereqs ## Lint code
@@ -44,7 +97,7 @@ ifeq (, $(shell which shellcheck))
 	@echo "### shellcheck could not be found, skipping shell lint"
 else
 	@echo "### Run shellcheck against bash scripts"
-	shellcheck ./res/*.sh
+	find . -name '*.sh' -not -path "./vendor/*" | xargs shellcheck
 endif
 
 .PHONY: clean
@@ -53,9 +106,41 @@ clean: ## Clean up build directory
 	@rm -rf $(FILES_OUTPUT_DIR)
 
 .PHONY: oc-commands
-oc-commands: build
+oc-commands: ## Generate oc plugins and add them to /usr/bin/
+	@echo "### Generating oc-commands"
+	./scripts/inject.sh $(DIST_DIR)
 	sudo cp -a ./build/. /usr/bin/
 
+##@ Images
+
+# note: to build and push custom image tag use: IMAGE_ORG=myuser VERSION=dev s
+.PHONY: image-build
+image-build: ## Build MULTIARCH_TARGETS images
+	trap 'exit' INT; \
+	$(foreach target,$(MULTIARCH_TARGETS),$(call build_target,$(target)))
+
+.PHONY: image-push
+image-push: ## Push MULTIARCH_TARGETS images
+	trap 'exit' INT; \
+	$(foreach target,$(MULTIARCH_TARGETS),$(call push_target,$(target)))
+
+.PHONY: manifest-build
+manifest-build: ## Build MULTIARCH_TARGETS manifest
+	echo 'building manifest $(IMAGE)'
+	DOCKER_BUILDKIT=1 $(OCI_BIN) rmi ${IMAGE} -f
+	DOCKER_BUILDKIT=1 $(OCI_BIN) manifest create ${IMAGE} $(foreach target,$(MULTIARCH_TARGETS), --amend ${IMAGE}-$(target));
+
+.PHONY: manifest-push
+manifest-push: ## Push MULTIARCH_TARGETS manifest
+	@echo 'publish manifest $(IMAGE)'
+ifeq (${OCI_BIN}, docker)
+	DOCKER_BUILDKIT=1 $(OCI_BIN) manifest push ${IMAGE};
+else
+	DOCKER_BUILDKIT=1 $(OCI_BIN) manifest push ${IMAGE} docker://${IMAGE};
+endif
+
 .PHONY: help
 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/shortcuts.mk

cmd/flow-capture.go

@@ -1,13 +1,9 @@
 package cmd
 
 import (
-	"bufio"
 	"encoding/json"
 	"fmt"
-	"net"
-	"net/textproto"
 	"os"
-	"os/exec"
 	"regexp"
 	"slices"
 	"sort"
@@ -17,6 +13,9 @@ import (
 
 	"github.com/eiannone/keyboard"
 	"github.com/netobserv/flowlogs-pipeline/pkg/config"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/utils"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/write/grpc"
+	"github.com/netobserv/flowlogs-pipeline/pkg/pipeline/write/grpc/genericmap"
 	"github.com/rodaine/table"
 	"github.com/spf13/cobra"
 
@@ -31,7 +30,7 @@ var flowCmd = &cobra.Command{
 }
 
 var (
-	flowsToShow = 40
+	flowsToShow = 35
 	regexes     = []string{}
 	lastFlows   = []config.GenericMap{}
 
@@ -54,56 +53,56 @@ func runFlowCapture(cmd *cobra.Command, args []string) {
 	go scanner()
 
 	wg := sync.WaitGroup{}
-	wg.Add(len(addresses))
-	for i := range addresses {
+	wg.Add(len(ports))
+	for i := range ports {
 		go func(idx int) {
 			defer wg.Done()
-			runFlowCaptureOnAddr(addresses[idx], nodes[idx])
+			runFlowCaptureOnAddr(ports[idx], nodes[idx])
 		}(i)
 	}
 	wg.Wait()
 }
 
-func runFlowCaptureOnAddr(addr string, filename string) {
-	log.Infof("Starting Flow Capture for %s...", filename)
-
-	tcpServer, err := net.ResolveTCPAddr("tcp", addr)
-
-	if err != nil {
-		log.Error("ResolveTCPAddr failed:", err.Error())
-		log.Fatal(err)
-	}
-	conn, err := net.DialTCP("tcp", nil, tcpServer)
-	if err != nil {
-		log.Error("Dial failed:", err.Error())
-		log.Fatal(err)
+func runFlowCaptureOnAddr(port int, filename string) {
+	if len(filename) > 0 {
+		log.Infof("Starting Flow Capture for %s...", filename)
+	} else {
+		log.Infof("Starting Flow Capture...")
+		filename = strings.Replace(
+			time.Now().UTC().Format(time.RFC3339),
+			":", "", -1) // get rid of offensive colons
 	}
-	reader := bufio.NewReader(conn)
-	tp := textproto.NewReader(reader)
+
 	var f *os.File
-	err = os.MkdirAll("./output/flow/", 0700)
+	err := os.MkdirAll("./output/flow/", 0700)
 	if err != nil {
 		log.Errorf("Create directory failed: %v", err.Error())
 		log.Fatal(err)
 	}
-	f, err = os.Create("./output/flow/" + filename)
+	f, err = os.Create("./output/flow/" + filename + ".json")
 	if err != nil {
 		log.Errorf("Create file %s failed: %v", filename, err.Error())
 		log.Fatal(err)
 	}
-	defer CleanupCapture(conn, f)
-	for {
-		// read one line (ended with \n or \r\n)
-		line, err := tp.ReadLineBytes()
+	defer f.Close()
+
+	flowPackets := make(chan *genericmap.Flow, 100)
+	collector, err := grpc.StartCollector(port, flowPackets)
+	if err != nil {
+		log.Error("StartCollector failed:", err.Error())
+		log.Fatal(err)
+	}
+	go func() {
+		<-utils.ExitChannel()
+		close(flowPackets)
+		collector.Close()
+	}()
+	for fp := range flowPackets {
+		go manageFlowsDisplay(fp.GenericMap.Value)
+		// append new line between each record to read file easilly
+		_, err = f.Write(append(fp.GenericMap.Value, []byte(",\n")...))
 		if err != nil {
-			log.Error("Read line failed:", err.Error())
-		} else {
-			// append new line between each record to read file easilly
-			_, err = f.Write(append(line, []byte("\n")...))
-			if err != nil {
-				log.Fatal(err)
-			}
-			go manageFlowsDisplay(line)
+			log.Fatal(err)
 		}
 	}
 }
@@ -180,12 +179,9 @@ func updateTable() {
 		lastRefresh = now
 
 		// clear terminal to render table properly
-		c := exec.Command("clear")
-		c.Stdout = os.Stdout
-		err := c.Run()
-		if err != nil {
-			log.Fatal(err)
-		}
+		fmt.Print("\x1bc")
+		// no wrap
+		fmt.Print("\033[?7l")
 
 		fmt.Print("Running network-observability-cli as Flow Capture\n")
 		fmt.Printf("Log level: %s\n", logLevel)