podinfo: add a deleted pod cache

cmd/tetragon/main.go

@@ -411,7 +411,10 @@ func tetragonExecute() error {
 		}
 
 		k8sClient := kubernetes.NewForConfigOrDie(config)
-		k8sWatcher = watcher.NewK8sWatcher(k8sClient, 60*time.Second)
+		k8sWatcher, err = watcher.NewK8sWatcher(k8sClient, 60*time.Second)
+		if err != nil {
+			return err
+		}
 	} else {
 		log.Info("Disabling Kubernetes API")
 		k8sWatcher = watcher.NewFakeK8sWatcher(nil)

pkg/metrics/watchermetrics/watchermetrics.go

@@ -44,17 +44,26 @@ var (
 		Help:        "The total number of events for a given watcher type.",
 		ConstLabels: nil,
 	}, []string{"watcher"})
+
+	WatcherDeletedPodCacheHits = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace:   consts.MetricsNamespace,
+		Name:        "watcher_delete_pod_cache_hits",
+		Help:        "The total hits for pod information in the deleted pod cache.",
+		ConstLabels: nil,
+	})
 )
 
 func RegisterMetrics(group metrics.Group) {
 	group.MustRegister(WatcherErrors)
 	group.MustRegister(WatcherEvents)
+	group.MustRegister(WatcherDeletedPodCacheHits)
 }
 
 func InitMetrics() {
 	// Initialize metrics with labels
 	GetWatcherEvents(K8sWatcher).Add(0)
 	GetWatcherErrors(K8sWatcher, FailedToGetPodError).Add(0)
+	GetWatcherDeletedPodCacheHits().Add(0)
 }
 
 // Get a new handle on an WatcherEvents metric for a watcher type
@@ -66,3 +75,7 @@ func GetWatcherEvents(watcherType Watcher) prometheus.Counter {
 func GetWatcherErrors(watcherType Watcher, watcherError ErrorType) prometheus.Counter {
 	return WatcherErrors.WithLabelValues(watcherType.String(), string(watcherError))
 }
+
+func GetWatcherDeletedPodCacheHits() prometheus.Counter {
+	return WatcherDeletedPodCacheHits
+}

pkg/process/podinfo_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cilium/tetragon/api/v1/tetragon"
 	"github.com/cilium/tetragon/pkg/watcher"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 	v1 "k8s.io/api/core/v1"
@@ -49,7 +50,8 @@ func TestK8sWatcher_GetPodInfo(t *testing.T) {
 	}
 
 	k8sClient := fake.NewSimpleClientset(&pod)
-	watcher := watcher.NewK8sWatcher(k8sClient, time.Hour)
+	watcher, err := watcher.NewK8sWatcher(k8sClient, time.Hour)
+	require.NoError(t, err)
 	watcher.Start()
 	pid := uint32(1)
 	podInfo := getPodInfo(watcher, "abcd1234", "curl", "cilium.io", 1)

pkg/watcher/deleted_pod_cache.go

@@ -0,0 +1,87 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Tetragon
+
+package watcher
+
+import (
+	"github.com/cilium/tetragon/pkg/logger"
+	"github.com/cilium/tetragon/pkg/metrics/watchermetrics"
+	lru "github.com/hashicorp/golang-lru/v2"
+	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/tools/cache"
+)
+
+type deletedPodCacheEntry struct {
+	pod        *v1.Pod
+	contStatus *v1.ContainerStatus
+}
+
+type deletedPodCache struct {
+	*lru.Cache[string, deletedPodCacheEntry]
+}
+
+func newDeletedPodCache() (*deletedPodCache, error) {
+	c, err := lru.New[string, deletedPodCacheEntry](128)
+	if err != nil {
+		return nil, err
+	}
+	return &deletedPodCache{c}, nil
+}
+
+func (c *deletedPodCache) eventHandler() cache.ResourceEventHandler {
+	return cache.ResourceEventHandlerFuncs{
+		DeleteFunc: func(obj interface{}) {
+			var pod *corev1.Pod
+			switch concreteObj := obj.(type) {
+			case *corev1.Pod:
+				pod = concreteObj
+			case cache.DeletedFinalStateUnknown:
+				// Handle the case when the watcher missed the deletion event
+				// (e.g. due to a lost apiserver connection).
+				deletedObj, ok := concreteObj.Obj.(*corev1.Pod)
+				if !ok {
+					return
+				}
+				pod = deletedObj
+			default:
+				return
+			}
+
+			run := func(s []v1.ContainerStatus) {
+				for i := range s {
+					contStatus := &s[i]
+					contID := contStatus.ContainerID
+					if contID == "" {
+						continue
+					}
+
+					key, err := containerIDKey(contID)
+					if err != nil {
+						logger.GetLogger().Warn("failed to crate container key for id '%s': %w", contID, err)
+						continue
+					}
+
+					c.Add(key, deletedPodCacheEntry{
+						pod:        pod,
+						contStatus: contStatus,
+					})
+				}
+			}
+
+			run(pod.Status.InitContainerStatuses)
+			run(pod.Status.ContainerStatuses)
+			run(pod.Status.EphemeralContainerStatuses)
+		},
+	}
+}
+
+func (c *deletedPodCache) findContainer(containerID string) (*corev1.Pod, *corev1.ContainerStatus, bool) {
+	v, ok := c.Get(containerID)
+	if !ok {
+		return nil, nil, false
+	}
+
+	watchermetrics.GetWatcherDeletedPodCacheHits().Inc()
+	return v.pod, v.contStatus, true
+}

pkg/watcher/watcher.go

@@ -51,8 +51,9 @@ type K8sResourceWatcher interface {
 
 // K8sWatcher maintains a local cache of k8s resources.
 type K8sWatcher struct {
-	informers map[string]cache.SharedIndexInformer
-	startFunc func()
+	informers       map[string]cache.SharedIndexInformer
+	startFunc       func()
+	deletedPodCache *deletedPodCache
 }
 
 type InternalSharedInformerFactory interface {
@@ -74,6 +75,19 @@ func podIndexFunc(obj interface{}) ([]string, error) {
 	return nil, fmt.Errorf("podIndexFunc: %w - found %T", errNoPod, obj)
 }
 
+func containerIDKey(contID string) (string, error) {
+	parts := strings.Split(contID, "//")
+	if len(parts) != 2 {
+		return "", fmt.Errorf("unexpected containerID format, expecting 'docker://<name>', got %q", contID)
+	}
+	cid := parts[1]
+	if len(cid) > containerIDLen {
+		cid = cid[:containerIDLen]
+	}
+	return cid, nil
+
+}
+
 // containerIndexFunc index pod by container IDs.
 func containerIndexFunc(obj interface{}) ([]string, error) {
 	var containerIDs []string
@@ -84,13 +98,9 @@ func containerIndexFunc(obj interface{}) ([]string, error) {
 			// be patient.
 			return nil
 		}
-		parts := strings.Split(fullContainerID, "//")
-		if len(parts) != 2 {
-			return fmt.Errorf("unexpected containerID format, expecting 'docker://<name>', got %q", fullContainerID)
-		}
-		cid := parts[1]
-		if len(cid) > containerIDLen {
-			cid = cid[:containerIDLen]
+		cid, err := containerIDKey(fullContainerID)
+		if err != nil {
+			return err
 		}
 		containerIDs = append(containerIDs, cid)
 		return nil
@@ -121,36 +131,50 @@ func containerIndexFunc(obj interface{}) ([]string, error) {
 	return nil, fmt.Errorf("%w - found %T", errNoPod, obj)
 }
 
-// NewK8sWatcher returns a pointer to an initialized K8sWatcher struct.
-func NewK8sWatcher(k8sClient kubernetes.Interface, stateSyncIntervalSec time.Duration) *K8sWatcher {
-	nodeName := node.GetNodeNameForExport()
-	if nodeName == "" {
-		logger.GetLogger().Warn("env var NODE_NAME not specified, K8s watcher will not work as expected")
+func newK8sWatcher(
+	informerFactory informers.SharedInformerFactory,
+) (*K8sWatcher, error) {
+
+	deletedPodCache, err := newDeletedPodCache()
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize deleted pod cache: %w", err)
 	}
 
 	k8sWatcher := &K8sWatcher{
-		informers: make(map[string]cache.SharedIndexInformer),
-		startFunc: func() {},
+		informers:       make(map[string]cache.SharedIndexInformer),
+		startFunc:       func() {},
+		deletedPodCache: deletedPodCache,
 	}
 
-	k8sInformerFactory := informers.NewSharedInformerFactoryWithOptions(k8sClient, stateSyncIntervalSec,
-		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
-			// Watch local pods only.
-			options.FieldSelector = "spec.nodeName=" + nodeName
-		}))
-	podInformer := k8sInformerFactory.Core().V1().Pods().Informer()
-	k8sWatcher.AddInformers(k8sInformerFactory, &InternalInformer{
+	podInformer := informerFactory.Core().V1().Pods().Informer()
+	k8sWatcher.AddInformers(informerFactory, &InternalInformer{
 		Name:     podInformerName,
 		Informer: podInformer,
 		Indexers: map[string]cache.IndexFunc{
 			containerIdx: containerIndexFunc,
 			podIdx:       podIndexFunc,
 		},
 	})
-
+	podInformer.AddEventHandler(k8sWatcher.deletedPodCache.eventHandler())
 	podhooks.InstallHooks(podInformer)
 
-	return k8sWatcher
+	return k8sWatcher, nil
+}
+
+// NewK8sWatcher returns a pointer to an initialized K8sWatcher struct.
+func NewK8sWatcher(k8sClient kubernetes.Interface, stateSyncIntervalSec time.Duration) (*K8sWatcher, error) {
+	nodeName := node.GetNodeNameForExport()
+	if nodeName == "" {
+		logger.GetLogger().Warn("env var NODE_NAME not specified, K8s watcher will not work as expected")
+	}
+
+	informerFactory := informers.NewSharedInformerFactoryWithOptions(k8sClient, stateSyncIntervalSec,
+		informers.WithTweakListOptions(func(options *metav1.ListOptions) {
+			// Watch local pods only.
+			options.FieldSelector = "spec.nodeName=" + nodeName
+		}))
+
+	return newK8sWatcher(informerFactory)
 }
 
 func (watcher *K8sWatcher) AddInformers(factory InternalSharedInformerFactory, infs ...*InternalInformer) {
@@ -210,9 +234,14 @@ func (watcher *K8sWatcher) FindContainer(containerID string) (*corev1.Pod, *core
 	// If we can't find any pod indexed then fall back to the entire pod list.
 	// If we find more than 1 pods indexed also fall back to the entire pod list.
 	if len(objs) != 1 {
-		return findContainer(containerID, podInformer.GetStore().List())
+		objs = podInformer.GetStore().List()
 	}
-	return findContainer(containerID, objs)
+	pod, cont, found := findContainer(containerID, objs)
+	if found {
+		return pod, cont, found
+	}
+
+	return watcher.deletedPodCache.findContainer(indexedContainerID)
 }
 
 // FindMirrorPod finds the mirror pod of a static pod based on the hash