tetragon: Remove sensors on exit not programs

cmd/tetragon/main.go

@@ -224,7 +224,6 @@ func tetragonExecute() error {
 	obs := observer.NewObserver(option.Config.TracingPolicy)
 	defer func() {
 		obs.PrintStats()
-		obs.RemovePrograms()
 	}()
 
 	defaultLevel := logger.GetLogLevel()
@@ -277,6 +276,9 @@ func tetragonExecute() error {
 	if err := obs.InitSensorManager(sensorMgWait); err != nil {
 		return err
 	}
+	defer func() {
+		observer.RemoveSensors(ctx)
+	}()
 
 	/* Remove any stale programs, otherwise feature set change can cause
 	 * old programs to linger resulting in undefined behavior. And because
@@ -338,7 +340,7 @@ func tetragonExecute() error {
 	pm, err := tetragonGrpc.NewProcessManager(
 		ctx,
 		&cleanupWg,
-		observer.SensorManager,
+		observer.GetSensorManager(),
 		hookRunner)
 	if err != nil {
 		return err
@@ -356,20 +358,24 @@ func tetragonExecute() error {
 	obs.AddListener(pm)
 	saveInitInfo()
 	if option.Config.EnableK8s {
-		go crd.WatchTracePolicy(ctx, observer.SensorManager)
+		go crd.WatchTracePolicy(ctx, observer.GetSensorManager())
 	}
 
 	obs.LogPinnedBpf(observerDir)
 
 	// load base sensor
-	if err := base.GetInitialSensor().Load(observerDir, observerDir); err != nil {
+	base := base.GetInitialSensor()
+	if err := base.Load(observerDir, observerDir); err != nil {
 		return err
 	}
+	defer func() {
+		base.Unload()
+	}()
 
 	// now that the base sensor was loaded, we can start the sensor manager
 	close(sensorMgWait)
 	sensorMgWait = nil
-	observer.SensorManager.LogSensorsAndProbes(ctx)
+	observer.GetSensorManager().LogSensorsAndProbes(ctx)
 
 	err = loadTpFromDir(ctx, option.Config.TracingPolicyDir)
 	if err != nil {
@@ -444,7 +450,7 @@ func addTracingPolicy(ctx context.Context, file string) error {
 		return err
 	}
 
-	err = observer.SensorManager.AddTracingPolicy(ctx, tp)
+	err = observer.GetSensorManager().AddTracingPolicy(ctx, tp)
 	if err != nil {
 		return err
 	}

pkg/bench/bench.go

@@ -227,7 +227,7 @@ func startBenchmarkExporter(ctx context.Context, obs *observer.Observer, summary
 	processManager, err := grpc.NewProcessManager(
 		ctx,
 		&wg,
-		observer.SensorManager,
+		observer.GetSensorManager(),
 		hookRunner)
 	if err != nil {
 		return err

pkg/observer/observer.go

@@ -40,9 +40,6 @@ var (
 	eventHandler = make(map[uint8]func(r *bytes.Reader) ([]Event, error))
 
 	observerList []*Observer
-
-	/* SensorManager handles dynamic sensors loading / unloading. */
-	SensorManager *sensors.Manager
 )
 
 type Event notify.Message
@@ -342,9 +339,11 @@ func (k *Observer) Start(ctx context.Context) error {
 
 // InitSensorManager starts the sensor controller
 func (k *Observer) InitSensorManager(waitChan chan struct{}) error {
-	var err error
-	SensorManager, err = sensors.StartSensorManager(option.Config.BpfDir, option.Config.MapDir, waitChan)
-	return err
+	mgr, err := sensors.StartSensorManager(option.Config.BpfDir, option.Config.MapDir, waitChan)
+	if err != nil {
+		return err
+	}
+	return SetSensorManager(mgr)
 }
 
 func NewObserver(configFile string) *Observer {
@@ -401,6 +400,12 @@ func (k *Observer) RemovePrograms() {
 	RemovePrograms(option.Config.BpfDir, option.Config.MapDir)
 }
 
+func RemoveSensors(ctx context.Context) {
+	if mgr := GetSensorManager(); mgr != nil {
+		mgr.RemoveAllSensors(ctx)
+	}
+}
+
 // Log Active pinned BPF resources
 func (k *Observer) LogPinnedBpf(observerDir string) {
 	finfo, err := os.Stat(observerDir)

pkg/observer/observertesthelper/observer_test_helper.go

@@ -134,7 +134,6 @@ func testDone(tb testing.TB, obs *observer.Observer) {
 		}
 	}
 
-	obs.RemovePrograms()
 	obs.PrintStats()
 	obs.Remove()
 }
@@ -213,8 +212,6 @@ func newDefaultObserver(oo *testObserverOptions) *observer.Observer {
 }
 
 func getDefaultObserver(tb testing.TB, ctx context.Context, base *sensors.Sensor, opts ...TestOption) (*observer.Observer, error) {
-	var cnfSensor *sensors.Sensor
-
 	testutils.CaptureLog(tb, logger.GetLogger().(*logrus.Logger))
 
 	o := newDefaultTestOptions(opts...)
@@ -245,15 +242,8 @@ func getDefaultObserver(tb testing.TB, ctx context.Context, base *sensors.Sensor
 			return nil, fmt.Errorf("failed to parse tracingpolicy: %w", err)
 		}
 	}
-	if tp != nil {
-		var err error
-		cnfSensor, err = sensors.GetMergedSensorFromParserPolicy(tp)
-		if err != nil {
-			return nil, err
-		}
-	}
 
-	if err := loadSensor(tb, base, cnfSensor); err != nil {
+	if err := loadObserver(tb, ctx, base, tp); err != nil {
 		return nil, err
 	}
 
@@ -379,9 +369,10 @@ func loadExporter(tb testing.TB, ctx context.Context, obs *observer.Observer, op
 
 	// NB(kkourt): we use the global that was set up by InitSensorManager(). We should clean
 	// this up and remove/hide the global variable.
-	sensorManager := observer.SensorManager
+	sensorManager := observer.GetSensorManager()
 	tb.Cleanup(func() {
-		sensorManager.StopSensorManager(context.TODO())
+		sensorManager.StopSensorManager(ctx)
+		observer.ResetSensorManager()
 	})
 
 	if oo.crd {
@@ -438,6 +429,26 @@ func loadExporter(tb testing.TB, ctx context.Context, obs *observer.Observer, op
 	return nil
 }
 
+func loadObserver(tb testing.TB, ctx context.Context, base *sensors.Sensor,
+	tp tracingpolicy.TracingPolicy) error {
+
+	if err := base.Load(option.Config.BpfDir, option.Config.MapDir); err != nil {
+		tb.Fatalf("Load base error: %s\n", err)
+	}
+
+	if tp != nil {
+		if err := observer.GetSensorManager().AddTracingPolicy(ctx, tp); err != nil {
+			tb.Fatalf("SensorManager.AddTracingPolicy error: %s\n", err)
+		}
+	}
+
+	tb.Cleanup(func() {
+		observer.RemoveSensors(ctx)
+		base.Unload()
+	})
+	return nil
+}
+
 func loadSensor(tb testing.TB, base *sensors.Sensor, sens *sensors.Sensor) error {
 	if err := base.Load(option.Config.BpfDir, option.Config.MapDir); err != nil {
 		tb.Fatalf("Load base error: %s\n", err)

pkg/observer/sensor_manager.go

@@ -0,0 +1,40 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Tetragon
+
+package observer
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/cilium/tetragon/pkg/sensors"
+)
+
+var (
+	// SensorManager handles dynamic sensors loading / unloading, and is a global variable for
+	// now
+	sensorManager   *sensors.Manager
+	sensorManagerMu sync.Mutex
+)
+
+// ResetSensorManager resets the global sensorManager variable to nil. Intended only for testing.
+func ResetSensorManager() {
+	sensorManager = nil
+}
+
+func SetSensorManager(sm *sensors.Manager) error {
+	sensorManagerMu.Lock()
+	defer sensorManagerMu.Unlock()
+
+	if sensorManager != nil {
+		return fmt.Errorf("observer sensorManager already set")
+	}
+	sensorManager = sm
+	return nil
+}
+
+func GetSensorManager() *sensors.Manager {
+	sensorManagerMu.Lock()
+	defer sensorManagerMu.Unlock()
+	return sensorManager
+}

pkg/sensors/base/base.go

@@ -59,6 +59,12 @@ var (
 	/* Internal statistics for debugging */
 	ExecveStats        = program.MapBuilder("execve_map_stats", Execve)
 	ExecveJoinMapStats = program.MapBuilder("tg_execve_joined_info_map_stats", ExecveBprmCommit)
+
+	sensor = sensors.Sensor{
+		Name:  "__base__",
+		Progs: GetDefaultPrograms(),
+		Maps:  GetDefaultMaps(),
+	}
 )
 
 func GetExecveMap() *program.Map {
@@ -100,11 +106,7 @@ func GetDefaultMaps() []*program.Map {
 
 // GetInitialSensor returns the base sensor
 func GetInitialSensor() *sensors.Sensor {
-	return &sensors.Sensor{
-		Name:  "__base__",
-		Progs: GetDefaultPrograms(),
-		Maps:  GetDefaultMaps(),
-	}
+	return &sensor
 }
 
 // ExecObj returns the exec object based on the kernel version

pkg/sensors/exec/cgroups_test.go

@@ -589,9 +589,7 @@ func setupTgRuntimeConf(t *testing.T, trackingCgrpLevel, logLevel, hierarchyId,
 }
 
 func setupObserver(ctx context.Context, t *testing.T) *tus.TestSensorManager {
-	testManager := tus.StartTestSensorManager(ctx, t)
-	observer.SensorManager = testManager.Manager
-
+	testManager := tus.GetTestSensorManager(ctx, t)
 	if err := observer.InitDataCache(1024); err != nil {
 		t.Fatalf("failed to call observer.InitDataCache %s", err)
 	}

pkg/sensors/exec/exec_test.go

@@ -502,7 +502,7 @@ func TestLoadInitialSensor(t *testing.T) {
 
 	tus.CheckSensorLoad([]*sensors.Sensor{sensor}, sensorMaps, sensorProgs, t)
 
-	sensors.UnloadAll()
+	sensor.Unload()
 }
 
 func TestDocker(t *testing.T) {

pkg/sensors/handler.go

@@ -4,6 +4,7 @@
 package sensors
 
 import (
+	"errors"
 	"fmt"
 
 	slimv1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/apis/meta/v1"
@@ -191,7 +192,25 @@ func (h *handler) addSensor(op *sensorAdd) error {
 	return nil
 }
 
+func removeAllSensors(h *handler) error {
+	var errs error
+	for _, col := range h.collections {
+		if err := col.unload(); err != nil {
+			errs = errors.Join(errs, err)
+		}
+		delete(h.collections, col.name)
+	}
+	return errs
+}
+
 func (h *handler) removeSensor(op *sensorRemove) error {
+	if op.all {
+		if op.name != "" {
+			return fmt.Errorf("removeSensor called with all flag and sensor name %s",
+				op.name)
+		}
+		return removeAllSensors(h)
+	}
 	col, exists := h.collections[op.name]
 	if !exists {
 		return fmt.Errorf("sensor %s does not exist", op.name)

pkg/sensors/load.go

@@ -369,3 +369,11 @@ func UnloadAll() {
 	AllPrograms = []*program.Program{}
 	AllMaps = []*program.Map{}
 }
+
+func UnloadSensors(sens []*Sensor) {
+	for i := range sens {
+		if err := sens[i].Unload(); err != nil {
+			logger.GetLogger().Warnf("Failed to unload sensor: %s", err)
+		}
+	}
+}

pkg/sensors/manager.go

@@ -235,6 +235,20 @@ func (h *Manager) RemoveSensor(ctx context.Context, sensorName string) error {
 	return err
 }
 
+func (h *Manager) RemoveAllSensors(ctx context.Context) error {
+	retc := make(chan error)
+	op := &sensorRemove{
+		ctx:     ctx,
+		all:     true,
+		retChan: retc,
+	}
+
+	h.sensorCtl <- op
+	err := <-retc
+
+	return err
+}
+
 func (h *Manager) StopSensorManager(ctx context.Context) error {
 	retc := make(chan error)
 	op := &sensorCtlStop{
@@ -325,6 +339,7 @@ type sensorAdd struct {
 type sensorRemove struct {
 	ctx     context.Context
 	name    string
+	all     bool
 	retChan chan error
 }
 

pkg/sensors/test/lseek_test.go

@@ -11,7 +11,6 @@ import (
 
 	ec "github.com/cilium/tetragon/api/v1/tetragon/codegen/eventchecker"
 	"github.com/cilium/tetragon/pkg/jsonchecker"
-	"github.com/cilium/tetragon/pkg/observer"
 	"github.com/cilium/tetragon/pkg/observer/observertesthelper"
 	_ "github.com/cilium/tetragon/pkg/sensors/exec"
 	tus "github.com/cilium/tetragon/pkg/testutils/sensors"
@@ -77,8 +76,7 @@ func TestSensorLseekEnable(t *testing.T) {
 
 	sensor := GetTestSensor()
 
-	smanager := tus.StartTestSensorManager(ctx, t)
-	observer.SensorManager = smanager.Manager
+	smanager := tus.GetTestSensorManager(ctx, t)
 	smanager.AddAndEnableSensor(ctx, t, sensor, sensor.Name)
 
 	observertesthelper.LoopEvents(ctx, t, &doneWG, &readyWG, obs)

pkg/sensors/tracing/kprobe_test.go

@@ -3616,10 +3616,6 @@ spec:
  - call: "sys_write"
    syscall: true
 `
-	b := base.GetInitialSensor()
-	if err := b.Load(option.Config.BpfDir, option.Config.MapDir); err != nil {
-		return fmt.Errorf("load base sensor failed: %w", err)
-	}
 
 	tp, _ := tracingpolicy.FromYAML(testHook)
 	if tp == nil {
@@ -3630,7 +3626,11 @@ spec:
 	if err != nil {
 		return err
 	}
-	return sens.Load(option.Config.BpfDir, option.Config.MapDir)
+	err = sens.Load(option.Config.BpfDir, option.Config.MapDir)
+	if err != nil {
+		return err
+	}
+	return sens.Unload()
 }
 
 func TestKprobeBpfAttr(t *testing.T) {
@@ -3777,7 +3777,7 @@ spec:
 
 	tus.CheckSensorLoad(sens, sensorMaps, sensorProgs, t)
 
-	sensors.UnloadAll()
+	sensors.UnloadSensors(sens)
 }
 
 func TestFakeSyscallError(t *testing.T) {