Backport #2144 and #2167 to v1.0 branch

cmd/tetragon/main.go

@@ -387,6 +387,7 @@ func tetragonExecute() error {
 		log.Info("Disabling Kubernetes API")
 		k8sWatcher = watcher.NewFakeK8sWatcher(nil)
 	}
+	k8sWatcher.Start()
 	_, err := cilium.InitCiliumState(ctx, option.Config.EnableCilium)
 	if err != nil {
 		return err

pkg/observer/observertesthelper/observer_test_helper.go

@@ -36,7 +36,6 @@ 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"
@@ -51,6 +50,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8stypes "k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/tools/cache"
 )
 
 var (
@@ -578,14 +578,15 @@ func (f *fakeK8sWatcher) FindContainer(containerID string) (*corev1.Pod, *corev1
 	return &pod, &container, true
 }
 
-func (f *fakeK8sWatcher) FindServiceByIP(ip string) ([]*corev1.Service, error) {
-	return nil, fmt.Errorf("service with IP %s not found", ip)
+func (f *fakeK8sWatcher) AddInformers(_ watcher.InternalSharedInformerFactory, _ ...*watcher.InternalInformer) {
 }
 
-func (f *fakeK8sWatcher) FindPodInfoByIP(ip string) ([]*v1alpha1.PodInfo, error) {
-	return nil, fmt.Errorf("PodInfo with IP %s not found", ip)
+func (f *fakeK8sWatcher) GetInformer(_ string) cache.SharedIndexInformer {
+	return nil
 }
 
+func (f *fakeK8sWatcher) Start() {}
+
 // 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/process/podinfo_test.go

@@ -54,6 +54,7 @@ func TestK8sWatcher_GetPodInfo(t *testing.T) {
 
 	k8sClient := fake.NewSimpleClientset(&pod)
 	watcher := watcher.NewK8sWatcher(k8sClient, time.Hour)
+	watcher.Start()
 	pid := uint32(1)
 	podInfo := getPodInfo(watcher, "abcd1234", "curl", "cilium.io", 1)
 	assert.True(t, proto.Equal(podInfo, &tetragon.Pod{

pkg/watcher/fake.go

@@ -6,8 +6,8 @@ package watcher
 import (
 	"fmt"
 
-	"github.com/cilium/tetragon/pkg/k8s/apis/cilium.io/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/tools/cache"
 )
 
 // FakeK8sWatcher is used as an "empty" K8sResourceWatcher when --enable-k8s-api flag is not set.
@@ -40,19 +40,6 @@ func (watcher *FakeK8sWatcher) FindPod(podID string) (*corev1.Pod, error) {
 	return nil, fmt.Errorf("podID %s not found (in %d pods)", podID, len(watcher.pods))
 }
 
-func (watcher *FakeK8sWatcher) FindServiceByIP(ip string) ([]*corev1.Service, error) {
-	for i := range watcher.services {
-		if service, ok := watcher.services[i].(*corev1.Service); ok {
-			for _, serviceIP := range service.Spec.ClusterIPs {
-				if serviceIP == ip {
-					return []*corev1.Service{service}, nil
-				}
-			}
-		}
-	}
-	return nil, fmt.Errorf("service with IP %s not found", ip)
-}
-
 // AddPod adds a pod to the fake k8s watcher. This is intended for testing.
 func (watcher *FakeK8sWatcher) AddPod(pod *corev1.Pod) {
 	watcher.pods = append(watcher.pods, pod)
@@ -73,6 +60,11 @@ 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)
+func (watcher *FakeK8sWatcher) AddInformers(_ InternalSharedInformerFactory, _ ...*InternalInformer) {
 }
+
+func (watcher *FakeK8sWatcher) GetInformer(_ string) cache.SharedIndexInformer {
+	return nil
+}
+
+func (watcher *FakeK8sWatcher) Start() {}

pkg/watcher/fake_test.go

@@ -8,37 +8,8 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func TestFakeK8sWatcher_FindServiceByIP(t *testing.T) {
-	services := []interface{}{
-		&corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{Name: "svc-1"},
-			Spec:       corev1.ServiceSpec{ClusterIPs: []string{"1.1.1.1"}},
-		},
-		&corev1.Service{
-			ObjectMeta: metav1.ObjectMeta{Name: "svc-2"},
-			Spec:       corev1.ServiceSpec{ClusterIPs: []string{"2.2.2.2", "3.3.3.3"}},
-		},
-	}
-	fakeWatcher := NewFakeK8sWatcherWithPodsAndServices(nil, services)
-	res, err := fakeWatcher.FindServiceByIP("1.1.1.1")
-	assert.NoError(t, err)
-	assert.Len(t, res, 1)
-	assert.Equal(t, res[0].Name, "svc-1")
-	res, err = fakeWatcher.FindServiceByIP("2.2.2.2")
-	assert.NoError(t, err)
-	assert.Len(t, res, 1)
-	assert.Equal(t, res[0].Name, "svc-2")
-	res, err = fakeWatcher.FindServiceByIP("3.3.3.3")
-	assert.NoError(t, err)
-	assert.Len(t, res, 1)
-	assert.Equal(t, res[0].Name, "svc-2")
-	_, err = fakeWatcher.FindServiceByIP("4.4.4.4")
-	assert.Error(t, err)
-}
-
 func TestFakeK8sWatcher_AddService(t *testing.T) {
 	services := []interface{}{
 		&corev1.Service{},