Increase testing coverage (and some related refactors)

.github/workflows/pull_request.yml

@@ -33,6 +33,6 @@ jobs:
       - name: Report coverage
         uses: codecov/codecov-action@v3
         with:
-          files: ./cover.out
+          files: ./testoutput/cover.txt
           flags: unittests
           token: ${{ secrets.CODECOV_TOKEN }}

.github/workflows/pull_request_integration_tests.yml

@@ -26,4 +26,10 @@ jobs:
         if: always()
         with:
           name: Test Logs
-          path: "*.log"
+          path: "testoutput/*.log"
+      - name: Report coverage
+        uses: codecov/codecov-action@v3
+        with:
+          files: ./testoutput/itest-covdata.txt
+          flags: integration-test
+          token: ${{ secrets.CODECOV_TOKEN }}

.gitignore

@@ -12,6 +12,7 @@
 
 # Output of the go coverage tool, specifically when used with LiteIDE
 *.out
+testoutput
 
 # Dependency directories (remove the comment below to include it)
 # vendor/

.golangci.yml

@@ -37,5 +37,5 @@ linters-settings:
       - indexAlloc
       - deprecatedComment
   cyclop:
-    max-complexity: 20
+    max-complexity: 10
 

Dockerfile

@@ -1,4 +1,4 @@
-# Build the manager binary
+# Build the autoinstrumenter binary
 FROM golang:1.20 as builder
 
 # TODO: embed software version in executable

Makefile

@@ -4,6 +4,8 @@ MAIN_GO_FILE ?= cmd/$(CMD)/main.go
 GOOS ?= linux
 GOARCH ?= amd64
 
+TEST_OUTPUT ?= ./testoutput
+
 IMG_REGISTRY ?= docker.io
 # Set your registry username. CI will set 'grafana' but you mustn't use it for manual pushing.
 IMG_ORG ?=
@@ -27,7 +29,7 @@ CLANG ?= clang
 CFLAGS := -O2 -g -Wall -Werror $(CFLAGS)
 
 # regular expressions for excluded file patterns
-EXCLUDE_COVERAGE_FILES="(/cmd/)|(bpf_bpfe)|(/pingserver/)|(/test/collector/)"
+EXCLUDE_COVERAGE_FILES="(bpf_)|(/pingserver/)|(/test/collector/)"
 
 .DEFAULT_GOAL := all
 
@@ -71,6 +73,7 @@ GO_OFFSETS_TRACKER = $(TOOLS_DIR)/go-offsets-tracker
 .PHONY: prereqs
 prereqs:
 	@echo "### Check if prerequisites are met, and installing missing dependencies"
+	mkdir -p $(TEST_OUTPUT)
 	$(call go-install-tool,$(GOLANGCI_LINT),github.com/golangci/golangci-lint/cmd/[email protected])
 	$(call go-install-tool,$(BPF2GO),github.com/cilium/ebpf/cmd/[email protected])
 	$(call go-install-tool,$(GO_OFFSETS_TRACKER),github.com/grafana/go-offsets-tracker/cmd/[email protected])
@@ -114,24 +117,30 @@ compile:
 	@echo "### Compiling project"
 	GOOS=$(GOOS) GOARCH=$(GOARCH) go build -mod vendor -ldflags -a -o bin/$(CMD) $(MAIN_GO_FILE)
 
+# Generated binary can provide coverage stats according to https://go.dev/blog/integration-test-coverage
+.PHONY: compile-for-coverage
+compile-for-coverage:
+	@echo "### Compiling project to generate coverage profiles"
+	GOOS=$(GOOS) GOARCH=$(GOARCH) go build -mod vendor -cover -a -o bin/$(CMD) $(MAIN_GO_FILE)
+
 .PHONY: test
 test:
 	@echo "### Testing code"
-	go test -mod vendor -a ./... -coverpkg=./... -coverprofile cover.all.out
+	go test -mod vendor -a ./... -coverpkg=./... -coverprofile $(TEST_OUTPUT)/cover.all.txt
 
 .PHONY: cov-exclude-generated
 cov-exclude-generated:
-	grep -vE $(EXCLUDE_COVERAGE_FILES) cover.all.out > cover.out
+	grep -vE $(EXCLUDE_COVERAGE_FILES) $(TEST_OUTPUT)/cover.all.txt > $(TEST_OUTPUT)/cover.txt
 
 .PHONY: coverage-report
 coverage-report: cov-exclude-generated
 	@echo "### Generating coverage report"
-	go tool cover --func=./cover.out
+	go tool cover --func=$(TEST_OUTPUT)/cover.txt
 
 .PHONY: coverage-report-html
 coverage-report-html: cov-exclude-generated
 	@echo "### Generating HTML coverage report"
-	go tool cover --html=./cover.out
+	go tool cover --html=$(TEST_OUTPUT)/cover.txt
 
 .PHONY: image-build-push
 image-build-push:
@@ -158,18 +167,29 @@ cleanup-integration-test:
 	$(OCI_BIN) compose $(COMPOSE_ARGS) rm -f
 	$(OCI_BIN) rmi -f $(shell $(OCI_BIN) images --format '{{.Repository}}:{{.Tag}}' | grep 'hatest-') || true
 
-# TODO: provide coverage info for integration testing https://go.dev/blog/integration-test-coverage
 .PHONY: run-integration-test
 run-integration-test:
 	@echo "### Running integration tests"
 	go clean -testcache
 	go test -mod vendor -a ./test/integration/... --tags=integration
 
 .PHONY: integration-test
-integration-test: prepare-integration-test
+integration-test: prereqs prepare-integration-test
 	$(MAKE) run-integration-test || (ret=$$?; $(MAKE) cleanup-integration-test && exit $$ret)
+	$(MAKE) itest-coverage-data
 	$(MAKE) cleanup-integration-test
 
+.PHONY: itest-coverage-data
+itest-coverage-data:
+	# merge coverage data from all the integration tests
+	mkdir -p $(TEST_OUTPUT)/merge
+	go tool covdata merge -i=$(TEST_OUTPUT) -o $(TEST_OUTPUT)/merge
+	go tool covdata textfmt -i=$(TEST_OUTPUT)/merge -o $(TEST_OUTPUT)/itest-covdata.raw.txt
+	# replace the unexpected /src/cmd/otelauto/main.go file by the module path
+	sed 's/^\/src\/cmd\//github.com\/grafana\/ebpf-autoinstrument\/cmd\//' $(TEST_OUTPUT)/itest-covdata.raw.txt > $(TEST_OUTPUT)/itest-covdata.all.txt
+	# exclude generated files from coverage data
+	grep -vE $(EXCLUDE_COVERAGE_FILES) $(TEST_OUTPUT)/itest-covdata.all.txt > $(TEST_OUTPUT)/itest-covdata.txt
+
 .PHONY: drone
 drone:
 	@echo "### Regenerating and signing .drone/drone.yml"

cmd/otelauto/main.go

@@ -7,6 +7,8 @@ import (
 	"io"
 	"net/http"
 	"os"
+	"os/signal"
+	"syscall"
 
 	"golang.org/x/exp/slog"
 
@@ -49,8 +51,21 @@ func main() {
 	}
 
 	slog.Info("Starting main node")
-	// TODO: add shutdown hook for graceful stop
-	bp.Run(context.TODO())
+
+	// Adding shutdown hook for graceful stop
+	ctx, cancel := context.WithCancel(context.Background())
+	exit := make(chan os.Signal, 1)
+	signal.Notify(exit, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		sig := <-exit
+		slog.Debug("Received termination signal", "signal", sig.String())
+		cancel()
+	}()
+
+	go bp.Run(ctx)
+
+	<-ctx.Done()
+	slog.Info("stopping auto-instrumenter")
 }
 
 func loadConfig(configPath *string) *pipe.Config {

pkg/ebpf/common/ringbuf.go

@@ -10,11 +10,36 @@ import (
 	"time"
 
 	"github.com/cilium/ebpf"
+
 	"github.com/cilium/ebpf/ringbuf"
 	"github.com/mariomac/pipes/pkg/node"
 	"golang.org/x/exp/slog"
 )
 
+// ringBufReader interface extracts the used methods from ringbuf.Reader for proper
+// dependency injection during tests
+type ringBufReader interface {
+	io.Closer
+	Read() (ringbuf.Record, error)
+}
+
+// readerFactory instantiates a ringBufReader from a ring buffer. In unit tests, we can
+// replace this function by a mock/dummy.
+var readerFactory = func(rb *ebpf.Map) (ringBufReader, error) {
+	return ringbuf.NewReader(rb)
+}
+
+type ringBufForwarder struct {
+	cfg        *TracerConfig
+	logger     *slog.Logger
+	ringbuffer *ebpf.Map
+	closers    []io.Closer
+	events     []HTTPRequestTrace
+	evLen      int
+	access     sync.Mutex
+	ticker     *time.Ticker
+}
+
 // ForwardRingbuf returns a function reads HTTPRequestTraces from an input ring buffer, accumulates them into an
 // internal buffer, and forwards them to an output events channel, previously converted to transform.HTTPRequestSpan
 // instances
@@ -24,73 +49,100 @@ func ForwardRingbuf(
 	ringbuffer *ebpf.Map,
 	closers ...io.Closer,
 ) node.StartFuncCtx[[]HTTPRequestTrace] {
-	// TODO: make use of context to cancel process
-	return func(_ context.Context, eventsChan chan<- []HTTPRequestTrace) {
-		// BPF will send each measured trace via Ring Buffer, so we listen for them from the
-		// user space.
-		eventsReader, err := ringbuf.NewReader(ringbuffer)
+	rbf := ringBufForwarder{
+		cfg: cfg, logger: logger, ringbuffer: ringbuffer, closers: closers,
+	}
+	return rbf.readAndForward
+}
+
+func (rbf *ringBufForwarder) readAndForward(ctx context.Context, eventsChan chan<- []HTTPRequestTrace) {
+	// BPF will send each measured trace via Ring Buffer, so we listen for them from the
+	// user space.
+	eventsReader, err := readerFactory(rbf.ringbuffer)
+	if err != nil {
+		rbf.logger.Error("creating perf reader. Exiting", err)
+		return
+	}
+	rbf.closers = append(rbf.closers, eventsReader)
+	defer rbf.closeAllResources()
+
+	rbf.events = make([]HTTPRequestTrace, rbf.cfg.BatchLength)
+	rbf.evLen = 0
+
+	// If the underlying context is closed, it closes the events reader
+	// so the function can exit.
+	go rbf.bgListenContextCancelation(ctx, eventsReader)
+
+	// Forwards periodically on timeout, if the batch is not full
+	if rbf.cfg.BatchTimeout > 0 {
+		rbf.ticker = time.NewTicker(rbf.cfg.BatchTimeout)
+		go rbf.bgFlushOnTimeout(eventsChan)
+	}
+
+	// Main loop:
+	// 1. Listen for content in the ring buffer
+	// 2. Decode binary data into HTTPRequestTrace instance
+	// 3. Accumulate the HTTPRequestTrace into a batch slice
+	// 4. When the length of the batch slice reaches cfg.BatchLength,
+	//    submit it to the next stage of the pipeline
+	for {
+		rbf.logger.Debug("starting to read perf buffer")
+		record, err := eventsReader.Read()
+		rbf.logger.Debug("received event")
 		if err != nil {
-			logger.Error("creating perf reader. Exiting", err)
-			return
-		}
-		defer func() {
-			logger.Debug("closing eBPF resources")
-			for _, c := range closers {
-				_ = c.Close()
-			}
-			_ = eventsReader.Close()
-		}()
-
-		events := make([]HTTPRequestTrace, cfg.BatchLength)
-		ev := 0
-		ticker := time.NewTicker(cfg.BatchTimeout)
-		access := sync.Mutex{}
-		go func() {
-			if cfg.BatchTimeout == 0 {
+			if errors.Is(err, ringbuf.ErrClosed) {
+				rbf.logger.Debug("ring buffer is closed")
 				return
 			}
-			// submit periodically on timeout, if the batch is not full
-			for {
-				<-ticker.C
-				access.Lock()
-				if ev > 0 {
-					logger.Debug("submitting traces on timeout", "len", ev)
-					eventsChan <- events[:ev]
-					events = make([]HTTPRequestTrace, cfg.BatchLength)
-					ev = 0
-				}
-				access.Unlock()
-			}
-		}()
-		for {
-			logger.Debug("starting to read perf buffer")
-			record, err := eventsReader.Read()
-			logger.Debug("received event")
-			if err != nil {
-				if errors.Is(err, ringbuf.ErrClosed) {
-					logger.Debug("ring buffer is closed")
-					return
-				}
-				logger.Error("error reading from perf reader", err)
-				continue
-			}
+			rbf.logger.Error("error reading from perf reader", err)
+			continue
+		}
 
-			access.Lock()
-			err = binary.Read(bytes.NewBuffer(record.RawSample), binary.LittleEndian, &events[ev])
-			if err != nil {
-				logger.Error("error parsing perf event", err)
-				access.Unlock()
-				continue
-			}
-			ev++
-			if ev == cfg.BatchLength {
-				logger.Debug("submitting traces after batch is full", "len", ev)
-				eventsChan <- events
-				events = make([]HTTPRequestTrace, cfg.BatchLength)
-				ev = 0
-				ticker.Reset(cfg.BatchTimeout)
+		rbf.access.Lock()
+		err = binary.Read(bytes.NewBuffer(record.RawSample), binary.LittleEndian, &rbf.events[rbf.evLen])
+		if err != nil {
+			rbf.logger.Error("error parsing perf event", err)
+			rbf.access.Unlock()
+			continue
+		}
+		rbf.evLen++
+		if rbf.evLen == rbf.cfg.BatchLength {
+			rbf.logger.Debug("submitting traces after batch is full", "len", rbf.evLen)
+			rbf.flushEvents(eventsChan)
+			if rbf.ticker != nil {
+				rbf.ticker.Reset(rbf.cfg.BatchTimeout)
 			}
-			access.Unlock()
 		}
+		rbf.access.Unlock()
+	}
+}
+
+func (rbf *ringBufForwarder) flushEvents(eventsChan chan<- []HTTPRequestTrace) {
+	eventsChan <- rbf.events[:rbf.evLen]
+	rbf.events = make([]HTTPRequestTrace, rbf.cfg.BatchLength)
+	rbf.evLen = 0
+}
+
+func (rbf *ringBufForwarder) bgFlushOnTimeout(eventsChan chan<- []HTTPRequestTrace) {
+	for {
+		<-rbf.ticker.C
+		rbf.access.Lock()
+		if rbf.evLen > 0 {
+			rbf.logger.Debug("submitting traces on timeout", "len", rbf.evLen)
+			rbf.flushEvents(eventsChan)
+		}
+		rbf.access.Unlock()
+	}
+}
+
+func (rbf *ringBufForwarder) bgListenContextCancelation(ctx context.Context, eventsReader ringBufReader) {
+	<-ctx.Done()
+	_ = eventsReader.Close()
+}
+
+func (rbf *ringBufForwarder) closeAllResources() {
+	rbf.logger.Debug("closing eBPF resources")
+	for _, c := range rbf.closers {
+		_ = c.Close()
 	}
 }