k8s: Let controller-runtime manage the pod informer

- Use a pod informer from controller-runtime.
- Add a ControllerManager Implementation of the PodAccessor interface.
- Add integration tests for PodAccessor methods.

Signed-off-by: Michi Mutsuzaki <[email protected]>

Author: michi-covalent

.github/workflows/integration-test.yaml

@@ -41,8 +41,9 @@ jobs:
         # renovate: datasource=golang-version depName=go
         go-version: '1.24.2'
 
-    - name: Install Kind and create cluster
-      uses: helm/kind-action@a1b0e391336a6ee6713a0583f8c6240d70863de3 # v1.12.0
+    - name: Create a Kind cluster
+      run: |
+        make kind
 
     - name: Pull Tetragon Images
       uses: nick-fields/retry@7152eba30c6575329ac0576536151aca5a72780e # v3.0.0

cmd/tetragon/main.go

@@ -66,8 +66,6 @@ import (
 	"github.com/spf13/viper"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/durationpb"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 )
 
 var (
@@ -389,9 +387,8 @@ func tetragonExecuteCtx(ctx context.Context, cancel context.CancelFunc, ready fu
 	// Initialize a k8s watcher used to retrieve process metadata. This should
 	// happen before the sensors are loaded, otherwise events will be stuck
 	// waiting for metadata.
-	var k8sClient *kubernetes.Clientset
 	var controllerManager *manager.ControllerManager
-	var k8sWatcher watcher.K8sResourceWatcher
+	var podAccessor watcher.PodAccessor
 	if option.Config.EnableK8s {
 		log.Info("Enabling Kubernetes API")
 		// Start controller-runtime manager.
@@ -411,37 +408,18 @@ func tetragonExecuteCtx(ctx context.Context, cancel context.CancelFunc, ready fu
 				return err
 			}
 		}
-		// retrieve k8s clients
-		k8sClient, _, err = watcher.GetK8sClients(func(_ *rest.Config) error { return nil })
-		if err != nil {
-			return err
-		}
-
-		// create k8s watcher
-		k8sWatcher = watcher.NewK8sWatcher(k8sClient, nil, 60*time.Second)
-
-		// add informers for all resources
-		// NB(anna): To add a pod informer, we need to pass the underlying
-		// struct, not just the interface, because the function initializes
-		// a cache inside this struct. This can be refactored if needed.
-		realK8sWatcher := k8sWatcher.(*watcher.K8sWatcher)
-		err = watcher.AddPodInformer(realK8sWatcher, true)
-		if err != nil {
-			return err
-		}
+		podAccessor = controllerManager
 	} else {
 		log.Info("Disabling Kubernetes API")
-		k8sWatcher = watcher.NewFakeK8sWatcher(nil)
+		podAccessor = watcher.NewFakeK8sWatcher(nil)
 	}
-	// start k8s watcher
-	k8sWatcher.Start()
 
 	pcGCInterval := option.Config.ProcessCacheGCInterval
 	if pcGCInterval <= 0 {
 		pcGCInterval = defaults.DefaultProcessCacheGCInterval
 	}
 
-	if err := process.InitCache(k8sWatcher, option.Config.ProcessCacheSize, pcGCInterval); err != nil {
+	if err := process.InitCache(podAccessor, option.Config.ProcessCacheSize, pcGCInterval); err != nil {
 		return err
 	}
 
@@ -460,7 +438,7 @@ func tetragonExecuteCtx(ctx context.Context, cancel context.CancelFunc, ready fu
 	ctx, cancel2 := context.WithCancel(ctx)
 	defer cancel2()
 
-	hookRunner := rthooks.GlobalRunner().WithWatcher(k8sWatcher)
+	hookRunner := rthooks.GlobalRunner().WithWatcher(podAccessor)
 
 	err = setRedactionFilters()
 	if err != nil {

pkg/manager/manager.go

@@ -5,53 +5,91 @@ package manager
 
 import (
 	"context"
+	"fmt"
 	"sync"
 
 	"github.com/bombsimon/logrusr/v4"
 	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
 	"github.com/cilium/tetragon/pkg/logger"
+	"github.com/cilium/tetragon/pkg/podhooks"
+	"github.com/cilium/tetragon/pkg/reader/node"
+	"github.com/cilium/tetragon/pkg/watcher"
 	corev1 "k8s.io/api/core/v1"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/cache"
 	ctrl "sigs.k8s.io/controller-runtime"
+	cmCache "sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	ctrlManager "sigs.k8s.io/controller-runtime/pkg/manager"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 )
 
 var (
 	initOnce, startOnce sync.Once
 	manager             *ControllerManager
+	_                   watcher.PodAccessor = (*ControllerManager)(nil)
 )
 
 // ControllerManager is responsible for running controller-runtime controllers,
 // and interacting with Kubernetes API server in general. If you need to interact
 // with Kubernetes API server, this is the place to start.
 type ControllerManager struct {
-	Manager ctrlManager.Manager
+	Manager         ctrlManager.Manager
+	deletedPodCache *watcher.DeletedPodCache
+	podInformer     cache.SharedIndexInformer
 }
 
 func Get() *ControllerManager {
+	var err error
 	initOnce.Do(func() {
-		ctrl.SetLogger(logrusr.New(logger.GetLogger()))
-		scheme := runtime.NewScheme()
-		utilruntime.Must(clientgoscheme.AddToScheme(scheme))
-		utilruntime.Must(v1alpha1.AddToScheme(scheme))
-		utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
-		controllerManager, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{Scheme: scheme})
+		manager, err = newControllerManager(true)
 		if err != nil {
 			panic(err)
 		}
-		manager = &ControllerManager{
-			Manager: controllerManager,
-		}
 	})
 	return manager
 }
 
+// newControllerManager creates a new controller manager. The enableMetrics flag
+// is for unit tests so that we can instantiate multiple ControllerManager instances
+// without them trying to bind to the same port 8080.
+func newControllerManager(enableMetrics bool) (*ControllerManager, error) {
+	ctrl.SetLogger(logrusr.New(logger.GetLogger()))
+	scheme := runtime.NewScheme()
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+	utilruntime.Must(v1alpha1.AddToScheme(scheme))
+	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
+	cacheOptions := cmCache.Options{
+		ByObject: map[client.Object]cmCache.ByObject{
+			&corev1.Pod{}: {
+				Field: fields.OneTermEqualSelector("spec.nodeName", node.GetKubernetesNodeName()),
+			},
+		},
+	}
+	metricsOptions := metricsserver.Options{}
+	if !enableMetrics {
+		metricsOptions.BindAddress = "0"
+	}
+	controllerOptions := ctrl.Options{Scheme: scheme, Cache: cacheOptions, Metrics: metricsOptions}
+	controllerManager, err := ctrl.NewManager(ctrl.GetConfigOrDie(), controllerOptions)
+	if err != nil {
+		return nil, err
+	}
+	manager = &ControllerManager{
+		Manager: controllerManager,
+	}
+	err = manager.addPodInformer()
+	if err != nil {
+		return nil, err
+	}
+	return manager, nil
+}
+
 func (cm *ControllerManager) Start(ctx context.Context) {
 	startOnce.Do(func() {
 		go func() {
@@ -116,3 +154,46 @@ func (cm *ControllerManager) WaitCRDs(ctx context.Context, crds map[string]struc
 	}
 	return nil
 }
+
+func (cm *ControllerManager) addPodInformer() error {
+	// initialize deleted pod cache
+	var err error
+	deletedPodCache, err := watcher.NewDeletedPodCache()
+	if err != nil {
+		return fmt.Errorf("failed to initialize deleted pod cache: %w", err)
+	}
+	cm.deletedPodCache = deletedPodCache
+
+	// Initialize a pod informer.
+	podInformer, err := cm.Manager.GetCache().GetInformer(context.Background(), &corev1.Pod{})
+	if err != nil {
+		return err
+	}
+	cm.podInformer = podInformer.(cache.SharedIndexInformer)
+	err = cm.podInformer.AddIndexers(cache.Indexers{
+		watcher.ContainerIdx: watcher.ContainerIndexFunc,
+		watcher.PodIdx:       watcher.PodIndexFunc,
+	})
+	if err != nil {
+		return err
+	}
+	// add event handlers to the informer
+	_, err = cm.podInformer.AddEventHandler(cm.deletedPodCache.EventHandler())
+	if err != nil {
+		return nil
+	}
+	podhooks.InstallHooks(cm.podInformer)
+	return nil
+}
+
+func (cm *ControllerManager) FindContainer(containerID string) (*corev1.Pod, *corev1.ContainerStatus, bool) {
+	return watcher.FindContainer(containerID, cm.podInformer, cm.deletedPodCache)
+}
+
+func (cm *ControllerManager) FindPod(podID string) (*corev1.Pod, error) {
+	return watcher.FindPod(podID, cm.podInformer)
+}
+
+func (cm *ControllerManager) FindMirrorPod(hash string) (*corev1.Pod, error) {
+	return watcher.FindMirrorPod(hash, cm.podInformer)
+}

pkg/manager/manager_integration_test.go

@@ -7,25 +7,43 @@ package manager
 
 import (
 	"context"
+	"os"
 	"testing"
+	"time"
 
+	"github.com/cilium/tetragon/pkg/reader/node"
+	"github.com/cilium/tetragon/pkg/watcher"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
 )
 
+const (
+	nodeName = "tetragon-dev-control-plane"
+)
+
+// ManagerTestSuite is a test suite for the ControllerManager. It assumes that
+// a Kind cluster created with "make kind" is running. More specifically, it
+// assumes that the cluster has a single node named "tetragon-dev-control-plane".
 type ManagerTestSuite struct {
 	suite.Suite
 	testEnv *envtest.Environment
 	manager *ControllerManager
 }
 
 func (suite *ManagerTestSuite) SetupSuite() {
+	err := os.Setenv("NODE_NAME", nodeName)
+	assert.NoError(suite.T(), err)
+	node.SetKubernetesNodeName()
 	useExistingCluster := true
 	suite.testEnv = &envtest.Environment{
 		UseExistingCluster: &useExistingCluster,
 	}
-	_, err := suite.testEnv.Start()
+	_, err = suite.testEnv.Start()
 	assert.NoError(suite.T(), err)
 	suite.manager = Get()
 	suite.manager.Start(context.Background())
@@ -43,6 +61,85 @@ func (suite *ManagerTestSuite) TestListNamespaces() {
 	assert.Equal(suite.T(), namespaces[0].Name, namespace.Name)
 }
 
+func (suite *ManagerTestSuite) TestFindPod() {
+	var pods corev1.PodList
+	err := suite.manager.Manager.GetCache().List(context.Background(), &pods, client.InNamespace("kube-system"))
+	assert.NoError(suite.T(), err)
+	assert.NotEmpty(suite.T(), pods.Items)
+	pod, err := suite.manager.FindPod(string(pods.Items[0].UID))
+	assert.NoError(suite.T(), err)
+	assert.Equal(suite.T(), pods.Items[0].UID, pod.UID)
+}
+
+func (suite *ManagerTestSuite) TestFindContainer() {
+	// Create a pod.
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "nginx",
+			Namespace: "kube-system",
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{Name: "nginx", Image: "nginx"}},
+		},
+	}
+	k8sClient := suite.manager.Manager.GetClient()
+	_ = k8sClient.Create(context.Background(), pod)
+
+	// Get the container ID of the pod.
+	podFromClient := corev1.Pod{}
+	containerID := ""
+	assert.Eventually(suite.T(), func() bool {
+		err := k8sClient.Get(context.Background(), client.ObjectKey{Namespace: pod.Namespace, Name: pod.Name}, &podFromClient)
+		if err != nil {
+			return false
+		}
+		containerID, err = watcher.ContainerIDKey(podFromClient.Status.ContainerStatuses[0].ContainerID)
+		if err != nil {
+			return false
+		}
+		return true
+	}, 10*time.Second, 1*time.Second)
+
+	// FindContainer should return the pod and container.
+	podFromCache, container, found := suite.manager.FindContainer(containerID)
+	assert.True(suite.T(), found)
+	assert.Equal(suite.T(), pod.Name, podFromCache.Name)
+	assert.Equal(suite.T(), pod.Spec.Containers[0].Name, container.Name)
+
+	// Delete the pod.
+	err := k8sClient.Delete(context.Background(), pod)
+	assert.NoError(suite.T(), err)
+	assert.Eventually(suite.T(), func() bool {
+		err = k8sClient.Get(context.Background(), client.ObjectKey{Namespace: pod.Namespace, Name: pod.Name}, &podFromClient)
+		return errors.IsNotFound(err)
+	}, 10*time.Second, 1*time.Second)
+
+	// FindContainer should still return the pod and container from the deleted pod cache.
+	podFromCache, container, found = suite.manager.FindContainer(containerID)
+	assert.True(suite.T(), found)
+	assert.Equal(suite.T(), pod.Name, podFromCache.Name)
+	assert.Equal(suite.T(), pod.Spec.Containers[0].Name, container.Name)
+}
+
+func (suite *ManagerTestSuite) TestLocalPods() {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	err := os.Setenv("NODE_NAME", "nonexistent-node")
+	assert.NoError(suite.T(), err)
+	node.SetKubernetesNodeName()
+	controllerManager, err := newControllerManager(false)
+	assert.NoError(suite.T(), err)
+	go func() {
+		assert.NoError(suite.T(), controllerManager.Manager.Start(ctx))
+	}()
+	controllerManager.Manager.GetCache().WaitForCacheSync(ctx)
+	pods := corev1.PodList{}
+	err = controllerManager.Manager.GetCache().List(context.Background(), &pods)
+	assert.NoError(suite.T(), err)
+	// Pod cache should be empty because the node name is set to a nonexistent node.
+	assert.Empty(suite.T(), pods.Items)
+}
+
 func (suite *ManagerTestSuite) TearDownSuite() {
 	assert.NoError(suite.T(), suite.testEnv.Stop())
 }

pkg/watcher/deleted_pod_cache.go

@@ -55,7 +55,7 @@ func (c *DeletedPodCache) EventHandler() cache.ResourceEventHandler {
 						continue
 					}
 
-					key, err := containerIDKey(contID)
+					key, err := ContainerIDKey(contID)
 					if err != nil {
 						logger.GetLogger().Warn("failed to crate container key for id '%s': %w", contID, err)
 						continue