Stop Beyla if either NetO11y or AppO11y component fails (#1519) (#1520)

pkg/components/beyla.go

@@ -32,11 +32,14 @@ func RunBeyla(ctx context.Context, cfg *beyla.Config) error {
 		wg.Add(1)
 	}
 
+	// of one of both nodes fail, the other should stop
+	ctx, cancel := context.WithCancel(ctx)
 	errs := make(chan error, 2)
 	if app {
 		go func() {
 			defer wg.Done()
 			if err := setupAppO11y(ctx, ctxInfo, cfg); err != nil {
+				cancel()
 				errs <- err
 			}
 		}()
@@ -45,11 +48,13 @@ func RunBeyla(ctx context.Context, cfg *beyla.Config) error {
 		go func() {
 			defer wg.Done()
 			if err := setupNetO11y(ctx, ctxInfo, cfg); err != nil {
+				cancel()
 				errs <- err
 			}
 		}()
 	}
 	wg.Wait()
+	cancel()
 	select {
 	case err := <-errs:
 		return err
@@ -66,9 +71,11 @@ func setupAppO11y(ctx context.Context, ctxInfo *global.ContextInfo, config *beyl
 
 	instr := appolly.New(ctx, ctxInfo, config)
 	if err := instr.FindAndInstrument(&wg); err != nil {
+		slog.Debug("can't find  target process", "error", err)
 		return fmt.Errorf("can't find target process: %w", err)
 	}
 	if err := instr.ReadAndForward(); err != nil {
+		slog.Debug("can't start read and forwarding", "error", err)
 		return fmt.Errorf("can't start read and forwarding: %w", err)
 	}
 	return nil
@@ -82,9 +89,11 @@ func setupNetO11y(ctx context.Context, ctxInfo *global.ContextInfo, cfg *beyla.C
 	slog.Info("starting Beyla in Network metrics mode")
 	flowsAgent, err := agent.FlowsAgent(ctxInfo, cfg)
 	if err != nil {
+		slog.Debug("can't start network metrics capture", "error", err)
 		return fmt.Errorf("can't start network metrics capture: %w", err)
 	}
 	if err := flowsAgent.Run(ctx); err != nil {
+		slog.Debug("can't start network metrics capture", "error", err)
 		return fmt.Errorf("can't start network metrics capture: %w", err)
 	}
 	return nil

pkg/internal/discover/watcher_proc.go

@@ -111,11 +111,14 @@ func (pa *pollAccounter) Run(out chan<- []Event[processAttrs]) {
 	bpfWatchEvents := make(chan watcher.Event, 100)
 	if err := pa.loadBPFWatcher(pa.cfg, bpfWatchEvents); err != nil {
 		log.Error("Unable to load eBPF watcher for process events", "error", err)
+		// will stop pipeline in cascade
+		return
 	}
 
 	if pa.cfg.EBPF.BpfDebug {
 		if err := pa.loadBPFLogger(pa.cfg); err != nil {
 			log.Error("Unable to load eBPF logger for process events", "error", err)
+			// keep running without logs
 		}
 	}