watcher: Implement FindPodInfoByIP

Implement FindPodInfoByIP to look up PodInfo by IP address.

Signed-off-by: Michi Mutsuzaki <michi@isovalent.com>

pkg/observer/observertesthelper/observer_test_helper.go

@@ -35,6 +35,7 @@ import (
 	"github.com/cilium/tetragon/pkg/cilium"
 	"github.com/cilium/tetragon/pkg/exporter"
 	tetragonGrpc "github.com/cilium/tetragon/pkg/grpc"
+	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
 	"github.com/cilium/tetragon/pkg/logger"
 	"github.com/cilium/tetragon/pkg/option"
 	"github.com/cilium/tetragon/pkg/process"
@@ -539,6 +540,10 @@ func (f *fakeK8sWatcher) FindServiceByIP(ip string) ([]*corev1.Service, error) {
 	return nil, fmt.Errorf("service with IP %s not found", ip)
 }
 
+func (f *fakeK8sWatcher) FindPodInfoByIP(ip string) ([]*v1alpha1.PodInfo, error) {
+	return nil, fmt.Errorf("PodInfo with IP %s not found", ip)
+}
+
 // Used to wait for a process to start, we do a lookup on PROCFS because this
 // may be called before obs is created.
 func WaitForProcess(process string) error {

pkg/watcher/fake.go

@@ -6,6 +6,7 @@ package watcher
 import (
 	"fmt"
 
+	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 )
 
@@ -71,3 +72,7 @@ func (watcher *FakeK8sWatcher) AddService(service *corev1.Service) {
 func (watcher *FakeK8sWatcher) ClearAllServices() {
 	watcher.services = nil
 }
+
+func (watcher *FakeK8sWatcher) FindPodInfoByIP(ip string) ([]*v1alpha1.PodInfo, error) {
+	return nil, fmt.Errorf("PodInfo with IP %q not found", ip)
+}

pkg/watcher/watcher.go

@@ -10,27 +10,32 @@ import (
 	"strings"
 	"time"
 
+	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
+	"github.com/cilium/tetragon/pkg/k8s/client/clientset/versioned"
+	"github.com/cilium/tetragon/pkg/k8s/client/clientset/versioned/fake"
+	"github.com/cilium/tetragon/pkg/k8s/client/informers/externalversions"
 	"github.com/cilium/tetragon/pkg/logger"
 	"github.com/cilium/tetragon/pkg/podhooks"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
-
-	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 const (
 	containerIDLen = 15
 	containerIdx   = "containers-ids"
 	podIdx         = "pod-ids"
 	serviceIPsIdx  = "service-ips"
+	podInfoIPsIdx  = "pod-info-ips"
 )
 
 var (
 	errNoPod     = errors.New("object is not a *corev1.Pod")
 	errNoService = errors.New("object is not a *corev1.Service")
+	errNoPodInfo = errors.New("object is not a *PodInfo")
 )
 
 // K8sResourceWatcher defines an interface for accessing various resources from Kubernetes API.
@@ -43,12 +48,16 @@ type K8sResourceWatcher interface {
 
 	// FindServiceByIP finds a service given the IP address.
 	FindServiceByIP(ip string) ([]*corev1.Service, error)
+
+	// FindPodInfoByIP finds a service given the IP address.
+	FindPodInfoByIP(ip string) ([]*v1alpha1.PodInfo, error)
 }
 
 // K8sWatcher maintains a local cache of k8s resources.
 type K8sWatcher struct {
 	podInformer     cache.SharedIndexInformer
 	serviceInformer cache.SharedIndexInformer
+	podInfoInformer cache.SharedIndexInformer
 }
 
 func podIndexFunc(obj interface{}) ([]string, error) {
@@ -115,8 +124,26 @@ func serviceIPIndexFunc(obj interface{}) ([]string, error) {
 	return nil, fmt.Errorf("%w - found %T", errNoService, obj)
 }
 
+// podInfoIPIndexFunc indexes services by their IP addresses
+func podInfoIPIndexFunc(obj interface{}) ([]string, error) {
+	switch t := obj.(type) {
+	case *v1alpha1.PodInfo:
+		var ips []string
+		for _, ip := range t.Status.PodIPs {
+			ips = append(ips, ip.IP)
+		}
+		return ips, nil
+	}
+	return nil, fmt.Errorf("%w - found %T", errNoPodInfo, obj)
+}
+
 // NewK8sWatcher returns a pointer to an initialized K8sWatcher struct.
 func NewK8sWatcher(k8sClient kubernetes.Interface, stateSyncIntervalSec time.Duration) *K8sWatcher {
+	return NewK8sWatcherWithTetragonClient(k8sClient, fake.NewSimpleClientset(), stateSyncIntervalSec)
+}
+
+// NewK8sWatcherWithTetragonClient returns a pointer to an initialized K8sWatcher struct.
+func NewK8sWatcherWithTetragonClient(k8sClient kubernetes.Interface, tetragonClient versioned.Interface, stateSyncIntervalSec time.Duration) *K8sWatcher {
 	nodeName := os.Getenv("NODE_NAME")
 	if nodeName == "" {
 		logger.GetLogger().Warn("env var NODE_NAME not specified, K8s watcher will not work as expected")
@@ -150,15 +177,29 @@ func NewK8sWatcher(k8sClient kubernetes.Interface, stateSyncIntervalSec time.Dur
 		panic(err)
 	}
 
+	podInfoInformerFactory := externalversions.NewSharedInformerFactory(tetragonClient, stateSyncIntervalSec)
+	podInfoInformer := podInfoInformerFactory.Cilium().V1alpha1().PodInfo().Informer()
+	err = podInfoInformer.AddIndexers(map[string]cache.IndexFunc{
+		podInfoIPsIdx: podInfoIPIndexFunc,
+	})
+	if err != nil {
+		// Panic during setup since this should never fail, if it fails is a
+		// developer mistake.
+		panic(err)
+	}
+
 	podhooks.InstallHooks(podInformer)
 
 	k8sInformerFactory.Start(wait.NeverStop)
 	k8sInformerFactory.WaitForCacheSync(wait.NeverStop)
 	serviceInformerFactory.Start(wait.NeverStop)
 	serviceInformerFactory.WaitForCacheSync(wait.NeverStop)
+	podInfoInformerFactory.Start(wait.NeverStop)
+	podInfoInformerFactory.WaitForCacheSync(wait.NeverStop)
 	logger.GetLogger().WithField("num_pods", len(podInformer.GetStore().ListKeys())).Info("Initialized pod cache")
 	logger.GetLogger().WithField("num_services", len(serviceInformer.GetStore().ListKeys())).Info("Initialized service cache")
-	return &K8sWatcher{podInformer, serviceInformer}
+	logger.GetLogger().WithField("num_pod_info", len(podInfoInformer.GetStore().ListKeys())).Info("Initialized pod info cache")
+	return &K8sWatcher{podInformer, serviceInformer, podInfoInformer}
 }
 
 // FindContainer implements K8sResourceWatcher.FindContainer.
@@ -260,3 +301,22 @@ func (watcher *K8sWatcher) FindServiceByIP(ip string) ([]*corev1.Service, error)
 	}
 	return services, nil
 }
+
+func (watcher *K8sWatcher) FindPodInfoByIP(ip string) ([]*v1alpha1.PodInfo, error) {
+	objs, err := watcher.podInfoInformer.GetIndexer().ByIndex(podInfoIPsIdx, ip)
+	if err != nil {
+		return nil, fmt.Errorf("pod info watcher returned: %w", err)
+	}
+	if len(objs) == 0 {
+		return nil, fmt.Errorf("PodInfo with IP %s not found", ip)
+	}
+	var podInfo []*v1alpha1.PodInfo
+	for _, obj := range objs {
+		service, ok := obj.(*v1alpha1.PodInfo)
+		if !ok {
+			return nil, fmt.Errorf("unexpected type %t", objs[0])
+		}
+		podInfo = append(podInfo, service)
+	}
+	return podInfo, nil
+}

pkg/watcher/watcher_test.go

@@ -8,6 +8,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
+	fakeTetragon "github.com/cilium/tetragon/pkg/k8s/client/clientset/versioned/fake"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -46,3 +48,39 @@ func TestFindServiceByIP(t *testing.T) {
 	assert.Len(t, res, 1)
 	assert.Equal(t, "svc3", res[0].Name)
 }
+
+func TestPodInfoByIP(t *testing.T) {
+	ctx := context.Background()
+	k8sClient := fake.NewSimpleClientset()
+	tetragonClient := fakeTetragon.NewSimpleClientset()
+	watcher := NewK8sWatcherWithTetragonClient(k8sClient, tetragonClient, 0)
+	pod1 := v1alpha1.PodInfo{
+		ObjectMeta: metav1.ObjectMeta{Name: "pod1"},
+		Status:     v1alpha1.PodInfoStatus{PodIPs: []v1alpha1.PodIP{{IP: "1.1.1.1"}}},
+	}
+	pod2 := v1alpha1.PodInfo{
+		ObjectMeta: metav1.ObjectMeta{Name: "pod2"},
+		Status:     v1alpha1.PodInfoStatus{PodIPs: []v1alpha1.PodIP{{IP: "2.2.2.2"}}},
+	}
+	pod3 := v1alpha1.PodInfo{
+		ObjectMeta: metav1.ObjectMeta{Name: "pod3"},
+		Status:     v1alpha1.PodInfoStatus{PodIPs: []v1alpha1.PodIP{{IP: "3.3.3.3"}, {IP: "4.4.4.4"}}},
+	}
+	_, err := tetragonClient.CiliumV1alpha1().PodInfo("my-ns").Create(ctx, &pod1, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	_, err = tetragonClient.CiliumV1alpha1().PodInfo("my-ns").Create(ctx, &pod2, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	_, err = tetragonClient.CiliumV1alpha1().PodInfo("my-ns").Create(ctx, &pod3, metav1.CreateOptions{})
+	assert.NoError(t, err)
+	assert.Eventually(t, func() bool { return len(watcher.podInfoInformer.GetStore().List()) == 3 }, 10*time.Second, 1*time.Second)
+	res, err := watcher.FindPodInfoByIP("1.1.1.1")
+	assert.NoError(t, err)
+	assert.Len(t, res, 1)
+	assert.Equal(t, "pod1", res[0].Name)
+	res, err = watcher.FindPodInfoByIP("4.4.4.4")
+	assert.NoError(t, err)
+	assert.Len(t, res, 1)
+	assert.Equal(t, "pod3", res[0].Name)
+	_, err = watcher.FindPodInfoByIP("5.5.5.5")
+	assert.Error(t, err)
+}