sensors: test intermediate states ({,un}loading)

This patch tests the intermediate states that were introduced in the
previous patches. Specifically, it tests whether ListTracingPolicies
call observes loading/unloading states for the corresponding sensors. It
does so by a TestDelayedSensor that implements the SensorIface
interface.

Signed-off-by: Kornilios Kourtis <kornilios@isovalent.com>

pkg/sensors/delayed_sensor_test.go

@@ -0,0 +1,73 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Tetragon
+
+package sensors
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+// A sensor to test the intermediate policy states (loading / unloading)
+
+func makeTestDelayedSensor(t *testing.T) *TestDelayedSensor {
+	s := &TestDelayedSensor{
+		name:   "test-delayed-sensor",
+		loaded: false,
+		ch:     make(chan struct{}),
+	}
+	RegisterPolicyHandlerAtInit("dummy-policyhandler", &dummyHandler{s: s})
+	t.Cleanup(func() {
+		delete(registeredPolicyHandlers, "dummy-policyhandler")
+	})
+
+	return s
+}
+
+type TestDelayedSensor struct {
+	name   string
+	loaded bool
+	ch     chan struct{}
+}
+
+func (tds *TestDelayedSensor) GetName() string {
+	return tds.name
+}
+
+func (tds *TestDelayedSensor) IsLoaded() bool {
+	return tds.loaded
+}
+
+func (tds *TestDelayedSensor) Load(_ string) error {
+	select {
+	case <-tds.ch:
+	case <-time.After(10 * time.Second):
+		return fmt.Errorf("TestDelayedSensor/Load timeout when waiting for unblocking")
+	}
+	tds.loaded = true
+	return nil
+}
+
+func (tds *TestDelayedSensor) Unload() error {
+	select {
+	case <-tds.ch:
+	case <-time.After(10 * time.Second):
+		return fmt.Errorf("TestDelayedSensor/Unload timeout when waiting for unblocking")
+	}
+	tds.loaded = false
+	return nil
+}
+
+func (tds *TestDelayedSensor) Destroy() {
+	tds.loaded = false
+}
+
+func (tds *TestDelayedSensor) unblock(t *testing.T) {
+	select {
+	case tds.ch <- struct{}{}:
+	default:
+		t.Fatalf("unblocked failed: channel does not seem to be empty")
+	}
+
+}

pkg/sensors/manager_test.go

@@ -7,9 +7,11 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 	"time"
 
+	"github.com/cilium/tetragon/api/v1/tetragon"
 	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
 	"github.com/cilium/tetragon/pkg/policyfilter"
 	"github.com/cilium/tetragon/pkg/sensors/program"
@@ -304,3 +306,116 @@ func TestPolicyLoadErrorOverride(t *testing.T) {
 	assert.Len(t, l.Policies, 1)
 	assert.Equal(t, EnabledState.ToTetragonState(), l.Policies[0].State)
 }
+
+func TestPolicyListingWhileLoadUnload(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	polName := "test-policy"
+	testSensor := makeTestDelayedSensor(t)
+
+	mgr, err := StartSensorManager("", nil)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		if err := mgr.StopSensorManager(ctx); err != nil {
+			panic("failed to stop sensor manager")
+		}
+	})
+
+	checkPolicy := func(t *testing.T, statuses []*tetragon.TracingPolicyStatus, state tetragon.TracingPolicyState) {
+		require.Equal(t, 1, len(statuses))
+		pol := statuses[0]
+		require.Equal(t, pol.Name, polName)
+		require.Equal(t, pol.State, state)
+	}
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		// wait until at least one policy shows up, verify that it's in loading state and
+		// unblock the loading of the policy
+		for {
+			l, err := mgr.ListTracingPolicies(ctx)
+			require.NoError(t, err)
+			if len(l.Policies) > 0 {
+				checkPolicy(t, l.Policies, tetragon.TracingPolicyState_TP_STATE_LOADING)
+				testSensor.unblock(t)
+				break
+			}
+			time.Sleep(1 * time.Millisecond)
+		}
+		wg.Done()
+	}()
+
+	t.Log("adding policy")
+	policy := v1alpha1.TracingPolicy{}
+	policy.ObjectMeta.Name = polName
+	err = mgr.AddTracingPolicy(ctx, &policy)
+	require.NoError(t, err)
+	wg.Wait()
+
+	// check that policy is now enabled
+	l, err := mgr.ListTracingPolicies(ctx)
+	require.NoError(t, err)
+	checkPolicy(t, l.Policies, tetragon.TracingPolicyState_TP_STATE_ENABLED)
+
+	wg.Add(1)
+	go func() {
+		// wait until at least one policy shows up, verify that it's in unloading state and
+		// unblock the unloading of the policy
+		for {
+			l, err := mgr.ListTracingPolicies(ctx)
+			require.NoError(t, err)
+			require.Equal(t, len(l.Policies), 1)
+			if l.Policies[0].State == tetragon.TracingPolicyState_TP_STATE_UNLOADING {
+				testSensor.unblock(t)
+				break
+			}
+			time.Sleep(1 * time.Millisecond)
+		}
+		wg.Done()
+	}()
+
+	t.Log("disabling policy")
+	err = mgr.DisableTracingPolicy(ctx, polName, "")
+	require.NoError(t, err)
+	wg.Wait()
+
+	// check that policy is now disabled
+	l, err = mgr.ListTracingPolicies(ctx)
+	require.NoError(t, err)
+	checkPolicy(t, l.Policies, tetragon.TracingPolicyState_TP_STATE_DISABLED)
+
+	wg.Add(1)
+	go func() {
+		for {
+			l, err := mgr.ListTracingPolicies(ctx)
+			require.NoError(t, err)
+			require.Equal(t, len(l.Policies), 1, "policies:", l.Policies)
+			if l.Policies[0].State == tetragon.TracingPolicyState_TP_STATE_LOADING {
+				testSensor.unblock(t)
+				break
+			}
+			time.Sleep(1000 * time.Millisecond)
+		}
+		wg.Done()
+	}()
+
+	t.Log("re-enabling policy")
+	err = mgr.EnableTracingPolicy(ctx, polName, "")
+	require.NoError(t, err)
+	wg.Wait()
+
+	// check that policy is now diabled
+	l, err = mgr.ListTracingPolicies(ctx)
+	require.NoError(t, err)
+	checkPolicy(t, l.Policies, tetragon.TracingPolicyState_TP_STATE_ENABLED)
+
+	t.Log("deleting policy")
+	err = mgr.DeleteTracingPolicy(ctx, polName, "")
+	require.NoError(t, err)
+	l, err = mgr.ListTracingPolicies(ctx)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(l.Policies))
+}