prevent startup deadlock when watching many CRDs

go.mod

@@ -21,6 +21,7 @@ require (
 	k8s.io/apiextensions-apiserver v0.0.0-20230112083153-33db789573b1
 	k8s.io/apimachinery v0.28.6
 	k8s.io/client-go v0.28.6
+	k8s.io/kubectl v0.28.6
 )
 
 require (
@@ -35,30 +36,43 @@ require (
 	github.com/dustin/go-humanize v1.0.0 // indirect
 	github.com/emicklei/go-restful/v3 v3.9.0 // indirect
 	github.com/evanphx/json-patch v4.12.0+incompatible // indirect
+	github.com/fatih/camelcase v1.0.0 // indirect
+	github.com/go-errors/errors v1.4.2 // indirect
 	github.com/go-logr/logr v1.2.4 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/snappy v0.0.1 // indirect
+	github.com/google/btree v1.0.1 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
+	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/google/uuid v1.3.0 // indirect
+	github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7 // indirect
 	github.com/imdario/mergo v0.3.8 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/common v0.37.0 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/spf13/cobra v1.7.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/xlab/treeprint v1.2.0 // indirect
+	go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect
 	golang.org/x/oauth2 v0.8.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
 	golang.org/x/sys v0.22.0 // indirect
 	golang.org/x/term v0.22.0 // indirect
 	golang.org/x/text v0.16.0 // indirect
@@ -68,10 +82,13 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
+	k8s.io/cli-runtime v0.28.6 // indirect
 	k8s.io/klog/v2 v2.100.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9 // indirect
 	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
+	sigs.k8s.io/kustomize/api v0.13.5-0.20230601165947-6ce0bf390ce3 // indirect
+	sigs.k8s.io/kustomize/kyaml v0.14.3-0.20230601165947-6ce0bf390ce3 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )

helm/sloop/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 appVersion: "1.0"
 description: Sloop is a kubernetes history visualization tool.
 name: sloop
-version: 0.2.0
+version: 0.3.0
 icon: https://raw.githubusercontent.com/salesforce/sloop/master/other/sloop_logo_color.png

helm/sloop/templates/statefulset.yaml

@@ -31,6 +31,11 @@ spec:
         - --config=/sloopconfig/sloop.json
         command:
         - /sloop
+        {{- if .Values.remoteKubeconfigSecret }}
+        env:
+        - name: KUBECONFIG
+          value: /remote-kube/config
+        {{- end }}
         image: {{ .Values.image.repository }}:{{ .Values.image.tag }}
         imagePullPolicy: {{ .Values.image.pullPolicy }}
         name: sloop
@@ -66,6 +71,11 @@ spec:
           name: sloop-data
         - mountPath: /sloopconfig/
           name: sloopconfig
+        {{- if .Values.remoteKubeconfigSecret }}
+        - mountPath: /remote-kube/
+          name: remote-kube
+          readOnly: true
+        {{- end }}
       {{- with .Values.image.pullSecrets }}
       imagePullSecrets:
         {{- toYaml . | nindent 8 }}
@@ -77,6 +87,11 @@ spec:
       - configMap:
           name: {{ .Values.name }}
         name: sloopconfig
+      {{- if .Values.remoteKubeconfigSecret }}
+      - name: remote-kube
+        secret:
+          secretName: {{ .Values.remoteKubeconfigSecret }}
+      {{- end }}
       serviceAccountName: {{ .Values.serviceAccountName }}
 {{- if .Values.persistentVolume.enabled }}
   volumeClaimTemplates:

helm/sloop/values.yaml

@@ -6,6 +6,12 @@ podLabels: {}
 podAnnotations: {}
 name: sloop
 replicas: 1
+## If set, mounts this Secret (expected to hold a kubeconfig under the key
+## "config") at /remote-kube and points sloop's KUBECONFIG at it, so this
+## instance watches the cluster described by that kubeconfig instead of the
+## cluster it runs on. Used to watch node-less clusters (e.g. a management
+## plane) from a cluster that has nodes.
+remoteKubeconfigSecret: ""
 image:
   tag: latest
   repository: ghcr.io/salesforce/sloop

pkg/sloop/ingress/kubeclient.go

@@ -54,6 +54,14 @@ func MakeKubernetesClient(masterURL string, kubeContext string, privilegedAccess
 	if privilegedAccess {
 		clientConfig := getConfig(masterURL, kubeContext)
 		config, err = ClientConfig(clientConfig)
+		// Check err before touching config: on failure config is nil and the
+		// former unconditional config.Host deref turned every config problem
+		// (missing kubeconfig, unresolvable context) into a panic/restart loop
+		// instead of an error the caller can retry.
+		if err != nil {
+			glog.Errorf("Cannot create config from kubeconfig (context=%v, masterURL=%v): %v", kubeContext, masterURL, err)
+			return nil, err
+		}
 		glog.Infof("Building k8sclient with context=%v, masterURL=%v, configFile=%v.", kubeContext, config.Host, clientConfig.ConfigAccess().GetLoadingPrecedence())
 	} else {
 		glog.Infof("Creating Config using BuildConfigFromFlags")

pkg/sloop/ingress/kubewatcher.go

@@ -81,6 +81,7 @@ var (
 	metricIngressKubewatchbytes         = promauto.NewCounterVec(prometheus.CounterOpts{Name: "sloop_ingress_kubewatchbytes"}, []string{"kind", "watchtype"})
 	metricCrdInformerStarted            = promauto.NewGauge(prometheus.GaugeOpts{Name: "sloop_crd_informer_started"})
 	metricCrdInformerRunning            = promauto.NewGauge(prometheus.GaugeOpts{Name: "sloop_crd_informer_running"})
+	metricCrdInformerWatchErrors        = promauto.NewCounterVec(prometheus.CounterOpts{Name: "sloop_crd_informer_watch_errors"}, []string{"kind", "group", "version"})
 )
 
 // Todo: Add additional parameters for filtering
@@ -93,12 +94,18 @@ func NewKubeWatcherSource(kubeClient kubernetes.Interface, outChan chan typed.Ku
 
 	kw.startWellKnownInformers(kubeClient, enableGranularMetrics)
 	if includeCrds {
-		err := kw.startCustomInformers(masterURL, kubeContext, enableGranularMetrics)
-		if err != nil {
-			return nil, err
-		}
-
+		// Arm the ticker BEFORE the first attempt: startCustomInformers fails
+		// wholesale on a transient CRD List error, and bailing out here meant
+		// the ticker was never created, so CRD watching stayed off for the
+		// life of the process while the well-known informers kept running and
+		// /healthz stayed green. With the ticker armed first, the refresh
+		// loop retries discovery every crdRefreshInterval until it succeeds.
+		// The goroutine is started only after the first attempt returns, so
+		// two startCustomInformers calls never race over the informer map.
 		kw.refreshCrd = time.NewTicker(crdRefreshInterval)
+		if err := kw.startCustomInformers(masterURL, kubeContext, enableGranularMetrics); err != nil {
+			glog.Errorf("Initial CRD discovery failed, retrying every %v: %v", crdRefreshInterval, err)
+		}
 		go kw.refreshCrdInformers(masterURL, kubeContext, enableGranularMetrics)
 	}
 
@@ -122,7 +129,7 @@ func (i *kubeWatcherImpl) startWellKnownInformers(kubeclient kubernetes.Interfac
 	i.informerFactory.Core().V1().PersistentVolumeClaims().Informer().AddEventHandler(i.getEventHandlerForResource("PersistentVolumeClaim", enableGranularMetrics))
 	i.informerFactory.Core().V1().PersistentVolumes().Informer().AddEventHandler(i.getEventHandlerForResource("PersistentVolume", enableGranularMetrics))
 	i.informerFactory.Core().V1().Pods().Informer().AddEventHandler(i.getEventHandlerForResource("Pod", enableGranularMetrics))
-	i.informerFactory.Policy().V1beta1().PodDisruptionBudgets().Informer().AddEventHandler(i.getEventHandlerForResource("PodDisruptionBudget", enableGranularMetrics))
+	i.informerFactory.Policy().V1().PodDisruptionBudgets().Informer().AddEventHandler(i.getEventHandlerForResource("PodDisruptionBudget", enableGranularMetrics))
 	i.informerFactory.Core().V1().Services().Informer().AddEventHandler(i.getEventHandlerForResource("Service", enableGranularMetrics))
 	i.informerFactory.Core().V1().ReplicationControllers().Informer().AddEventHandler(i.getEventHandlerForResource("ReplicationController", enableGranularMetrics))
 	i.informerFactory.Storage().V1().StorageClasses().Informer().AddEventHandler(i.getEventHandlerForResource("StorageClass", enableGranularMetrics))
@@ -197,6 +204,18 @@ func (i *kubeWatcherImpl) startNewCrdInformer(crdInformer *crdInformerInfo, fact
 	kind := crdInformer.crd.kind
 	informer := factory.ForResource(gvr)
 	informer.Informer().AddEventHandler(i.getEventHandlerForResource(kind, enableGranularMetrics))
+	// Reflector list/watch failures are otherwise invisible: Run() never
+	// returns an error and retries forever, and metricCrdInformerRunning is
+	// incremented before Run(), so a permanently failing informer (RBAC
+	// denied, unreachable apiserver) looks healthy from the outside while
+	// recording nothing. Count and log the failures so an empty database can
+	// be traced to its cause.
+	if err := informer.Informer().SetWatchErrorHandler(func(_ *cache.Reflector, err error) {
+		metricCrdInformerWatchErrors.WithLabelValues(kind, gvr.Group, gvr.Version).Inc()
+		glog.Errorf("CRD informer list/watch failed for %s (%v): %v", kind, gvr, err)
+	}); err != nil {
+		glog.Errorf("Failed to install watch error handler for %s (%v): %v", kind, gvr, err)
+	}
 
 	go func() {
 		glog.V(2).Infof("Starting CRD informer for: %s (%v)", kind, gvr)
@@ -227,11 +246,33 @@ func getCrdList(crdClient clientset.Interface) ([]crdGroupVersionResourceKind, e
 
 	var resources []crdGroupVersionResourceKind
 	for _, crd := range crdList.Items {
+		// Watch exactly one served version per CRD. The apiserver returns the
+		// same objects through every served version, so watching all of them
+		// wrote each object N times (observed: 220 watch events for 110
+		// Projects on a v1+v1alpha1 CRD), and an unserved version has no
+		// endpoint at all - its informer 404-loops forever while being
+		// counted as running. Prefer the storage version when it is served,
+		// else fall back to the first served one.
+		chosen := ""
 		for _, version := range crd.Spec.Versions {
-			gvrk := crdGroupVersionResourceKind{group: crd.Spec.Group, version: version.Name, resource: crd.Spec.Names.Plural, kind: crd.Spec.Names.Kind}
-			glog.V(2).Infof("CRD: group: %s, version: %s, kind: %s, plural:%s, singular:%s, short names:%v", crd.Spec.Group, version.Name, crd.Spec.Names.Kind, crd.Spec.Names.Plural, crd.Spec.Names.Singular, crd.Spec.Names.ShortNames)
-			resources = append(resources, gvrk)
+			if !version.Served {
+				continue
+			}
+			if chosen == "" {
+				chosen = version.Name
+			}
+			if version.Storage {
+				chosen = version.Name
+				break
+			}
+		}
+		if chosen == "" {
+			glog.V(2).Infof("CRD %s kind %s has no served versions; skipping", crd.Spec.Group, crd.Spec.Names.Kind)
+			continue
 		}
+		gvrk := crdGroupVersionResourceKind{group: crd.Spec.Group, version: chosen, resource: crd.Spec.Names.Plural, kind: crd.Spec.Names.Kind}
+		glog.V(2).Infof("CRD: group: %s, version: %s, kind: %s, plural:%s, singular:%s, short names:%v", crd.Spec.Group, chosen, crd.Spec.Names.Kind, crd.Spec.Names.Plural, crd.Spec.Names.Singular, crd.Spec.Names.ShortNames)
+		resources = append(resources, gvrk)
 	}
 	return resources, nil
 }
@@ -245,11 +286,28 @@ func getCrdListV1beta1(crdClient clientset.Interface) ([]crdGroupVersionResource
 	// duplicated code (see getCrdList), the types for crdList are different
 	var resources []crdGroupVersionResourceKind
 	for _, crd := range crdList.Items {
+		// See getCrdList: watch exactly one served version per CRD,
+		// preferring the storage version.
+		chosen := ""
 		for _, version := range crd.Spec.Versions {
-			gvrk := crdGroupVersionResourceKind{group: crd.Spec.Group, version: version.Name, resource: crd.Spec.Names.Plural, kind: crd.Spec.Names.Kind}
-			glog.V(2).Infof("CRD: group: %s, version: %s, kind: %s, plural:%s, singular:%s, short names:%v", crd.Spec.Group, version.Name, crd.Spec.Names.Kind, crd.Spec.Names.Plural, crd.Spec.Names.Singular, crd.Spec.Names.ShortNames)
-			resources = append(resources, gvrk)
+			if !version.Served {
+				continue
+			}
+			if chosen == "" {
+				chosen = version.Name
+			}
+			if version.Storage {
+				chosen = version.Name
+				break
+			}
 		}
+		if chosen == "" {
+			glog.V(2).Infof("CRD %s kind %s has no served versions; skipping", crd.Spec.Group, crd.Spec.Names.Kind)
+			continue
+		}
+		gvrk := crdGroupVersionResourceKind{group: crd.Spec.Group, version: chosen, resource: crd.Spec.Names.Plural, kind: crd.Spec.Names.Kind}
+		glog.V(2).Infof("CRD: group: %s, version: %s, kind: %s, plural:%s, singular:%s, short names:%v", crd.Spec.Group, chosen, crd.Spec.Names.Kind, crd.Spec.Names.Plural, crd.Spec.Names.Singular, crd.Spec.Names.ShortNames)
+		resources = append(resources, gvrk)
 	}
 	return resources, nil
 }
@@ -345,15 +403,24 @@ func (i *kubeWatcherImpl) processUpdate(kind string, obj interface{}, watchResul
 }
 
 func (i *kubeWatcherImpl) writeToOutChan(watchResult *typed.KubeWatchResult) {
-	// We need to ensure that no messages are written to outChan after stop is called
-	// Kube watch library has a way to tell it to stop, but no way to know it is complete
-	// Use a lock around output channel for this purpose
+	// We need to ensure that no messages are written to outChan after stop is called.
+	// The lock only guards the stopped check; we must NOT hold it across the channel
+	// send, otherwise a full channel would block here while holding i.protection, which
+	// deadlocks any other path that needs the lock (e.g. starting CRD informers during
+	// the initial sync of a cluster with many CRDs).
 	i.protection.Lock()
-	defer i.protection.Unlock()
-	if i.stopped {
+	stopped := i.stopped
+	i.protection.Unlock()
+	if stopped {
 		return
 	}
-	i.outchan <- *watchResult // WARNING - if this channel gets full, this push will block while holding i.protection in a locked state
+
+	// Send without holding the lock. Select on stopChan so that the send unblocks if the
+	// watcher is stopped while the channel is full, instead of blocking forever.
+	select {
+	case i.outchan <- *watchResult:
+	case <-i.stopChan:
+	}
 }
 
 func (i *kubeWatcherImpl) getResourceAsJsonString(kind string, obj interface{}) (string, error) {

pkg/sloop/ingress/kubewatcher_test.go

@@ -39,14 +39,38 @@ type dummyData struct {
 
 // when fake client tries to list CRDs, return a list with one defined
 func reactionListOfOne(_ k8sTesting.Action) (bool, runtime.Object, error) {
-	versions := []apiextensionsv1.CustomResourceDefinitionVersion{{Name: "v1"}}
+	versions := []apiextensionsv1.CustomResourceDefinitionVersion{{Name: "v1", Served: true, Storage: true}}
 	name := apiextensionsv1.CustomResourceDefinitionNames{Plural: "things", Kind: "k"}
 	spec := apiextensionsv1.CustomResourceDefinitionSpec{Group: "g", Versions: versions, Names: name}
 	crd := apiextensionsv1.CustomResourceDefinition{Spec: spec}
 	list := apiextensionsv1.CustomResourceDefinitionList{Items: []apiextensionsv1.CustomResourceDefinition{crd}}
 	return true, &list, nil
 }
 
+// when fake client tries to list CRDs, return one CRD that serves two
+// versions (storage version listed second) plus one CRD with no served
+// version at all: getCrdList must pick exactly the storage version of the
+// first and skip the second entirely.
+func reactionMultiVersion(_ k8sTesting.Action) (bool, runtime.Object, error) {
+	dual := apiextensionsv1.CustomResourceDefinition{Spec: apiextensionsv1.CustomResourceDefinitionSpec{
+		Group: "g",
+		Names: apiextensionsv1.CustomResourceDefinitionNames{Plural: "things", Kind: "k"},
+		Versions: []apiextensionsv1.CustomResourceDefinitionVersion{
+			{Name: "v1alpha1", Served: true, Storage: false},
+			{Name: "v1", Served: true, Storage: true},
+		},
+	}}
+	unserved := apiextensionsv1.CustomResourceDefinition{Spec: apiextensionsv1.CustomResourceDefinitionSpec{
+		Group: "g2",
+		Names: apiextensionsv1.CustomResourceDefinitionNames{Plural: "others", Kind: "o"},
+		Versions: []apiextensionsv1.CustomResourceDefinitionVersion{
+			{Name: "v1", Served: false, Storage: true},
+		},
+	}}
+	list := apiextensionsv1.CustomResourceDefinitionList{Items: []apiextensionsv1.CustomResourceDefinition{dual, unserved}}
+	return true, &list, nil
+}
+
 // when fake client tries to list CRDs, return an error
 func reactionError(_ k8sTesting.Action) (bool, runtime.Object, error) {
 	return true, nil, fmt.Errorf("failed")
@@ -185,6 +209,15 @@ func Test_getCrdList(t *testing.T) {
 	crdList, err = getCrdList(crdClient)
 	assert.Len(t, crdList, 1)
 	assert.NoError(t, err)
+
+	// One informer per CRD, on the served storage version; CRDs with no
+	// served version are skipped entirely.
+	crdClient, _ = newTestCrdClient(reactionMultiVersion)(&rest.Config{})
+	crdList, err = getCrdList(crdClient)
+	assert.NoError(t, err)
+	assert.Len(t, crdList, 1)
+	assert.Equal(t, "v1", crdList[0].version)
+	assert.Equal(t, "k", crdList[0].kind)
 }
 
 func Test_getEventHandlerForResource(t *testing.T) {