tetragon: Load initial sensor via sensor manager

Loading initial sensor via sensor manager so it stores all the
loaded sensors/programs. So far we kept it separated but having
it in one place is handy for metrics requesting this data.

Moving the load before the server is started so we don't need to
"wait" for that. This way we can remove the sensorMgWait channel.

Signed-off-by: Jiri Olsa <[email protected]>

cmd/tetragon/main.go

@@ -189,6 +189,18 @@ func deleteOldBpfDir(path string) {
 	log.Infof("Removed bpf instance: %s", path)
 }
 
+func loadInitialSensor(ctx context.Context) error {
+	base.ConfigCgroupRate(&option.Config.CgroupRate)
+
+	mgr := observer.GetSensorManager()
+	initialSensor := base.GetInitialSensor()
+
+	if err := mgr.AddSensor(ctx, initialSensor.Name, initialSensor); err != nil {
+		return err
+	}
+	return mgr.EnableSensor(ctx, initialSensor.Name)
+}
+
 func tetragonExecute() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -359,20 +371,7 @@ func tetragonExecute() error {
 		cancel()
 	}()
 
-	// start sensor manager, and have it wait on sensorMgWait until we load
-	// the base sensor. note that this means that calling methods on the
-	// manager will block so they will have to either be executed in a
-	// goroutine or after we close the sensorMgWait channel to avoid
-	// deadlock.
-	sensorMgWait := make(chan struct{})
-	defer func() {
-		// if we fail before closing the channel, close it so that
-		// the sensor manager routine is unblocked.
-		if sensorMgWait != nil {
-			close(sensorMgWait)
-		}
-	}()
-	if err := obs.InitSensorManager(sensorMgWait); err != nil {
+	if err := obs.InitSensorManager(); err != nil {
 		return err
 	}
 
@@ -447,6 +446,16 @@ func tetragonExecute() error {
 		return err
 	}
 
+	// Load initial sensor before we start the server,
+	// so it's there before we allow to load policies.
+	if err = loadInitialSensor(ctx); err != nil {
+		return err
+	}
+	observer.GetSensorManager().LogSensorsAndProbes(ctx)
+	defer func() {
+		observer.RemoveSensors(ctx)
+	}()
+
 	pm, err := tetragonGrpc.NewProcessManager(
 		ctx,
 		&cleanupWg,
@@ -477,28 +486,9 @@ func tetragonExecute() error {
 
 	obs.LogPinnedBpf(observerDir)
 
-	base.ConfigCgroupRate(&option.Config.CgroupRate)
-
-	// load base sensor
-	initialSensor := base.GetInitialSensor()
-	if err := initialSensor.Load(observerDir); err != nil {
-		return err
-	}
-	defer func() {
-		initialSensor.Unload()
-	}()
-
 	cgrouprate.NewCgroupRate(ctx, pm, base.CgroupRateMap, &option.Config.CgroupRate)
 	cgrouprate.Config(base.CgroupRateOptionsMap)
 
-	// now that the base sensor was loaded, we can start the sensor manager
-	close(sensorMgWait)
-	sensorMgWait = nil
-	observer.GetSensorManager().LogSensorsAndProbes(ctx)
-	defer func() {
-		observer.RemoveSensors(ctx)
-	}()
-
 	err = loadTpFromDir(ctx, option.Config.TracingPolicyDir)
 	if err != nil {
 		return err

pkg/bench/bench.go

@@ -87,7 +87,7 @@ func runTetragon(ctx context.Context, configFile string, args *Arguments, summar
 	option.Config.BpfDir = bpf.MapPrefixPath()
 	obs := observer.NewObserver()
 
-	if err := obs.InitSensorManager(nil); err != nil {
+	if err := obs.InitSensorManager(); err != nil {
 		logger.GetLogger().Fatalf("InitSensorManager failed: %v", err)
 	}
 

pkg/observer/observer.go

@@ -341,8 +341,8 @@ func (k *Observer) Start(ctx context.Context) error {
 }
 
 // InitSensorManager starts the sensor controller
-func (k *Observer) InitSensorManager(waitChan chan struct{}) error {
-	mgr, err := sensors.StartSensorManager(option.Config.BpfDir, waitChan)
+func (k *Observer) InitSensorManager() error {
+	mgr, err := sensors.StartSensorManager(option.Config.BpfDir)
 	if err != nil {
 		return err
 	}

pkg/observer/observertesthelper/observer_test_helper.go

@@ -358,7 +358,7 @@ func loadExporter(tb testing.TB, ctx context.Context, obs *observer.Observer, op
 	processCacheSize := 32768
 	dataCacheSize := 1024
 
-	if err := obs.InitSensorManager(nil); err != nil {
+	if err := obs.InitSensorManager(); err != nil {
 		return err
 	}
 

pkg/sensors/manager.go

@@ -27,24 +27,18 @@ type SensorStatus struct {
 // The purpose of this goroutine is to serialize loading and unloading of
 // sensors as requested from different goroutines (e.g., different GRPC
 // clients).
-//
-// if waitChan is not nil, the serving of sensor requests will block until
-// something is received. The intention of this is to allow the main function
-// to first load the base sensor before the sensor manager starts loading other sensors.
 func StartSensorManager(
 	bpfDir string,
-	waitChan chan struct{},
 ) (*Manager, error) {
 	pfState, err := policyfilter.GetState()
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize policy filter state: %w", err)
 	}
-	return StartSensorManagerWithPF(bpfDir, waitChan, pfState)
+	return StartSensorManagerWithPF(bpfDir, pfState)
 }
 
 func StartSensorManagerWithPF(
 	bpfDir string,
-	waitChan chan struct{},
 	pfState policyfilter.State,
 ) (*Manager, error) {
 	colMap := newCollectionMap()

pkg/sensors/manager_test.go

@@ -43,7 +43,7 @@ func TestAddPolicy(t *testing.T) {
 	})
 
 	policy := v1alpha1.TracingPolicy{}
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	assert.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -71,7 +71,7 @@ func TestAddPolicies(t *testing.T) {
 	})
 
 	policy := v1alpha1.TracingPolicy{}
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	assert.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -102,7 +102,7 @@ func TestAddPolicySpecError(t *testing.T) {
 	})
 
 	policy := v1alpha1.TracingPolicy{}
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	assert.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -134,7 +134,7 @@ func TestAddPolicyLoadError(t *testing.T) {
 	})
 
 	policy := v1alpha1.TracingPolicy{}
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	assert.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -157,7 +157,7 @@ func TestPolicyFilterDisabled(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	mgr, err := StartSensorManagerWithPF("", nil, policyfilter.DisabledState())
+	mgr, err := StartSensorManagerWithPF("", policyfilter.DisabledState())
 	assert.NoError(t, err)
 	defer mgr.StopSensorManager(ctx)
 
@@ -209,7 +209,7 @@ func TestPolicyStates(t *testing.T) {
 		})
 
 		policy := v1alpha1.TracingPolicy{}
-		mgr, err := StartSensorManager("", nil)
+		mgr, err := StartSensorManager("")
 		require.NoError(t, err)
 		t.Cleanup(func() {
 			if err := mgr.StopSensorManager(ctx); err != nil {
@@ -234,7 +234,7 @@ func TestPolicyStates(t *testing.T) {
 		})
 
 		policy := v1alpha1.TracingPolicy{}
-		mgr, err := StartSensorManager("", nil)
+		mgr, err := StartSensorManager("")
 		require.NoError(t, err)
 		t.Cleanup(func() {
 			if err := mgr.StopSensorManager(ctx); err != nil {
@@ -274,7 +274,7 @@ func TestPolicyLoadErrorOverride(t *testing.T) {
 	})
 
 	policy := v1alpha1.TracingPolicy{}
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -313,7 +313,7 @@ func TestPolicyListingWhileLoadUnload(t *testing.T) {
 	polName := "test-policy"
 	testSensor := makeTestDelayedSensor(t)
 
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -440,7 +440,7 @@ func TestPolicyKernelMemoryBytes(t *testing.T) {
 	})
 
 	policy := v1alpha1.TracingPolicy{}
-	mgr, err := StartSensorManager("", nil)
+	mgr, err := StartSensorManager("")
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {

pkg/sensors/tracing/generickprobe_test.go

@@ -144,7 +144,7 @@ func Test_DisableEnablePolicy_Kprobe(t *testing.T) {
 
 	tus.LoadSensor(t, base.GetInitialSensor())
 	path := bpf.MapPrefixPath()
-	mgr, err := sensors.StartSensorManager(path, nil)
+	mgr, err := sensors.StartSensorManager(path)
 	assert.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {
@@ -190,7 +190,7 @@ func Test_DisableEnablePolicy_KernelMemoryBytes(t *testing.T) {
 
 	tus.LoadSensor(t, base.GetInitialSensor())
 	path := bpf.MapPrefixPath()
-	mgr, err := sensors.StartSensorManager(path, nil)
+	mgr, err := sensors.StartSensorManager(path)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		if err := mgr.StopSensorManager(ctx); err != nil {

pkg/testutils/sensors/sensors.go

@@ -67,7 +67,7 @@ func getTestSensorManager(ctx context.Context, t *testing.T, pfState policyfilte
 	}
 
 	path := bpf.MapPrefixPath()
-	mgr, err = sensors.StartSensorManagerWithPF(path, nil, pfState)
+	mgr, err = sensors.StartSensorManagerWithPF(path, pfState)
 	if err != nil {
 		t.Fatalf("StartSensorManagerWithPF failed: %s", err)
 	}