Use informers for k8s metadata watcher

Summary:
Switch from using ListWatch + RetryWatchers to the Informer model.
Also drops the use of genny and just inlines the watchers.

Test Plan: skaffold

Reviewers: michelle, philkuz

Reviewed By: michelle, philkuz

Signed-off-by: Vihang Mehta <vihang@pixielabs.ai>

Differential Revision: https://phab.corp.pixielabs.ai/D10036

GitOrigin-RevId: 3af3771

Author: vihangm

src/vizier/services/metadata/controllers/k8smeta/BUILD.bazel

@@ -22,7 +22,6 @@ go_library(
         "k8s_metadata_controller.go",
         "k8s_metadata_handler.go",
         "k8s_metadata_store.go",
-        "k8s_metadata_utils.gen.go",
         "k8s_metadata_utils.go",
         "metadata_topic_listener.go",
     ],
@@ -43,15 +42,11 @@ go_library(
         "@com_github_nats_io_nats_go//:nats_go",
         "@com_github_sirupsen_logrus//:logrus",
         "@io_k8s_api//core/v1:core",
-        "@io_k8s_apimachinery//pkg/apis/meta/internalversion",
-        "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
-        "@io_k8s_apimachinery//pkg/fields",
-        "@io_k8s_apimachinery//pkg/runtime",
         "@io_k8s_apimachinery//pkg/watch",
+        "@io_k8s_client_go//informers",
         "@io_k8s_client_go//kubernetes",
         "@io_k8s_client_go//rest",
         "@io_k8s_client_go//tools/cache",
-        "@io_k8s_client_go//tools/watch",
     ],
 )
 
@@ -60,7 +55,6 @@ go_test(
     srcs = [
         "k8s_metadata_handler_test.go",
         "k8s_metadata_store_test.go",
-        "k8s_metadata_utils_test.go",
         "metadata_topic_listener_test.go",
     ],
     embed = [":k8smeta"],
@@ -77,8 +71,5 @@ go_test(
         "@com_github_nats_io_nats_go//:nats_go",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",
-        "@io_k8s_api//core/v1:core",
-        "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
-        "@io_k8s_apimachinery//pkg/watch",
     ],
 )

src/vizier/services/metadata/controllers/k8smeta/k8s_metadata_controller.go

@@ -22,22 +22,11 @@ import (
 	"sync"
 	"time"
 
-	log "github.com/sirupsen/logrus"
-	v1 "k8s.io/api/core/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/cache"
-
-	"px.dev/pixie/src/vizier/services/metadata/storepb"
 )
 
 const (
-	// refreshUpdateDuration is how frequently we request all k8s updates to
-	// refresh the ones in our DataStore.
-	refreshUpdateDuration = 12 * time.Hour
 	// resourceUpdateTTL is how long the k8s update live in the DataStore.
 	resourceUpdateTTL = 24 * time.Hour
 )
@@ -53,14 +42,7 @@ type Controller struct {
 
 // watcher watches a k8s resource type and forwards the updates to the given update channel.
 type watcher interface {
-	Sync(storedUpdates []*storepb.K8SResource) error
-	StartWatcher(chan struct{}, *sync.WaitGroup)
-}
-
-func listObject(resource string, clientset *kubernetes.Clientset) (runtime.Object, error) {
-	watcher := cache.NewListWatchFromClient(clientset.CoreV1().RESTClient(), resource, v1.NamespaceAll, fields.Everything())
-	opts := metav1.ListOptions{}
-	return watcher.List(opts)
+	StartWatcher(chan struct{})
 }
 
 // NewController creates a new Controller.
@@ -84,103 +66,20 @@ func NewController(mds Store, updateCh chan *K8sResourceMessage) (*Controller, e
 	// for example, nodes and namespaces must be synced before pods, since nodes/namespaces
 	// contain pods.
 	watchers := []watcher{
-		NewNodeWatcher("nodes", updateCh, clientset),
-		NewNamespaceWatcher("namespaces", updateCh, clientset),
-		NewPodWatcher("pods", updateCh, clientset),
-		NewEndpointsWatcher("endpoints", updateCh, clientset),
-		NewServiceWatcher("services", updateCh, clientset),
+		nodeWatcher("nodes", updateCh, clientset),
+		namespaceWatcher("namespaces", updateCh, clientset),
+		podWatcher("pods", updateCh, clientset),
+		endpointsWatcher("endpoints", updateCh, clientset),
+		serviceWatcher("services", updateCh, clientset),
 	}
 
 	mc := &Controller{quitCh: quitCh, updateCh: updateCh, watchers: watchers}
 
-	go mc.Start(mds)
-
-	return mc, nil
-}
-
-// Start starts the k8s watcher. Every 12h, it will resync such that the updates from the
-// last 24h will always contain updates from currently running resources.
-func (mc *Controller) Start(mds Store) {
-	// Loop the sync/watch so that if it ever fails because of an intermittent error, it
-	// will continue trying to collect data.
-	for {
-		select {
-		case <-mc.quitCh:
-			return // Quit signaled, so exit the goroutine.
-		default:
-			err := mc.runSyncWatchLoop(mds)
-			if err == nil {
-				return // Quit signaled, so exit the goroutine.
-			}
-			log.WithError(err).Info("Failed K8s metadata sync/watch... Restarting.")
-			// Wait 5 minutes before retrying, however if stop is called, just return.
-			select {
-			case <-mc.quitCh:
-				return
-			case <-time.After(5 * time.Minute):
-				continue
-			}
-		}
-	}
-}
-
-func (mc *Controller) runSyncWatchLoop(mds Store) error {
-	// Run initial sync and watch.
-	watcherQuitCh := make(chan struct{})
-	var wg sync.WaitGroup
-
-	err := mc.syncAndWatch(mds, watcherQuitCh, &wg)
-	if err != nil {
-		return err
-	}
-
-	ticker := time.NewTicker(refreshUpdateDuration)
-
-	defer func() {
-		close(watcherQuitCh)
-		ticker.Stop()
-	}()
-
-	for {
-		select {
-		case <-ticker.C:
-			close(watcherQuitCh) // Stop previous watcher, resync to send updates for all currently running resources.
-			watcherQuitCh = make(chan struct{})
-			wg.Wait()
-			err = mc.syncAndWatch(mds, watcherQuitCh, &wg)
-			if err != nil {
-				return err
-			}
-		case <-mc.quitCh:
-			return nil
-		}
-	}
-}
-
-// Start syncs the state stored in the datastore with what is currently running in k8s.
-func (mc *Controller) syncAndWatch(mds Store, quitCh chan struct{}, wg *sync.WaitGroup) error {
-	lastUpdate, err := mds.GetUpdateVersion(KelvinUpdateTopic)
-	if err != nil {
-		log.WithError(err).Info("Failed to get latest update version")
-		return err
-	}
-	storedUpdates, err := mds.FetchFullResourceUpdates(0, lastUpdate)
-	if err != nil {
-		log.WithError(err).Info("Failed to fetch resource updates")
-		return err
-	}
-
 	for _, w := range mc.watchers {
-		err := w.Sync(storedUpdates)
-		if err != nil {
-			log.WithError(err).Info("Failed to sync metadata")
-			return err
-		}
-		wg.Add(1)
-		go w.StartWatcher(quitCh, wg)
+		go w.StartWatcher(quitCh)
 	}
 
-	return nil
+	return mc, nil
 }
 
 // Stop stops all K8s watchers.