Merge pull request kubernetes#92644 from wojtek-t/informers_readyz

Wait for all informers to sync in /readyz.

Author: k8s-ci-robot

staging/src/k8s.io/apiserver/pkg/server/config.go

@@ -605,13 +605,20 @@ func (c completedConfig) New(name string, delegationTarget DelegationTarget) (*G
 	}
 
 	genericApiServerHookName := "generic-apiserver-start-informers"
-	if c.SharedInformerFactory != nil && !s.isPostStartHookRegistered(genericApiServerHookName) {
-		err := s.AddPostStartHook(genericApiServerHookName, func(context PostStartHookContext) error {
-			c.SharedInformerFactory.Start(context.StopCh)
-			return nil
-		})
-		if err != nil {
-			return nil, err
+	if c.SharedInformerFactory != nil {
+		if !s.isPostStartHookRegistered(genericApiServerHookName) {
+			err := s.AddPostStartHook(genericApiServerHookName, func(context PostStartHookContext) error {
+				c.SharedInformerFactory.Start(context.StopCh)
+				return nil
+			})
+			if err != nil {
+				return nil, err
+			}
+			// TODO: Once we get rid of /healthz consider changing this to post-start-hook.
+			err = s.addReadyzChecks(healthz.NewInformerSyncHealthz(c.SharedInformerFactory))
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 

staging/src/k8s.io/apiserver/pkg/server/config_test.go

@@ -167,6 +167,7 @@ func TestNewWithDelegate(t *testing.T) {
 		"/metrics",
 		"/readyz",
 		"/readyz/delegate-health",
+		"/readyz/informer-sync",
 		"/readyz/log",
 		"/readyz/ping",
 		"/readyz/poststarthook/delegate-post-start-hook",
@@ -242,10 +243,10 @@ func checkExpectedPathsAtRoot(url string, expectedPaths []string, t *testing.T)
 			pathset.Insert(p.(string))
 		}
 		expectedset := sets.NewString(expectedPaths...)
-		for _, p := range pathset.Difference(expectedset) {
+		for p := range pathset.Difference(expectedset) {
 			t.Errorf("Got %v path, which we did not expect", p)
 		}
-		for _, p := range expectedset.Difference(pathset) {
+		for p := range expectedset.Difference(pathset) {
 			t.Errorf(" Expected %v path which we did not get", p)
 		}
 	})

staging/src/k8s.io/apiserver/pkg/server/healthz/BUILD

@@ -31,6 +31,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/endpoints/metrics:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/server/httplog:go_default_library",
+        "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//vendor/k8s.io/klog/v2:go_default_library",
     ],
 )

staging/src/k8s.io/apiserver/pkg/server/healthz/healthz.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/endpoints/metrics"
 	"k8s.io/apiserver/pkg/server/httplog"
+	"k8s.io/client-go/informers"
 	"k8s.io/klog/v2"
 )
 
@@ -81,6 +82,39 @@ func (l *log) Check(_ *http.Request) error {
 	return fmt.Errorf("logging blocked")
 }
 
+type informerSync struct {
+	sharedInformerFactory informers.SharedInformerFactory
+}
+
+var _ HealthChecker = &informerSync{}
+
+// NewInformerSyncHealthz returns a new HealthChecker that will pass only if all informers in the given sharedInformerFactory sync.
+func NewInformerSyncHealthz(sharedInformerFactory informers.SharedInformerFactory) HealthChecker {
+	return &informerSync{
+		sharedInformerFactory: sharedInformerFactory,
+	}
+}
+
+func (i *informerSync) Name() string {
+	return "informer-sync"
+}
+
+func (i *informerSync) Check(_ *http.Request) error {
+	stopCh := make(chan struct{})
+	// Close stopCh to force checking if informers are synced now.
+	close(stopCh)
+
+	var informersByStarted map[bool][]string
+	for informerType, started := range i.sharedInformerFactory.WaitForCacheSync(stopCh) {
+		informersByStarted[started] = append(informersByStarted[started], informerType.String())
+	}
+
+	if notStarted := informersByStarted[false]; len(notStarted) > 0 {
+		return fmt.Errorf("%d informers not started yet: %v", len(notStarted), notStarted)
+	}
+	return nil
+}
+
 // NamedCheck returns a healthz checker for the given name and function.
 func NamedCheck(name string, check func(r *http.Request) error) HealthChecker {
 	return &healthzCheck{name, check}

test/integration/master/kube_apiserver_test.go

@@ -98,11 +98,15 @@ func TestRun(t *testing.T) {
 	}
 }
 
-func endpointReturnsStatusOK(client *kubernetes.Clientset, path string) bool {
-	res := client.CoreV1().RESTClient().Get().AbsPath(path).Do(context.TODO())
+func endpointReturnsStatusOK(client *kubernetes.Clientset, path string) (bool, error) {
+	res := client.CoreV1().RESTClient().Get().RequestURI(path).Do(context.TODO())
 	var status int
 	res.StatusCode(&status)
-	return status == http.StatusOK
+	_, err := res.Raw()
+	if err != nil {
+		return false, err
+	}
+	return status == http.StatusOK, nil
 }
 
 func TestLivezAndReadyz(t *testing.T) {
@@ -113,11 +117,11 @@ func TestLivezAndReadyz(t *testing.T) {
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
-	if !endpointReturnsStatusOK(client, "/livez") {
-		t.Fatalf("livez should be healthy")
+	if statusOK, err := endpointReturnsStatusOK(client, "/livez"); err != nil || !statusOK {
+		t.Fatalf("livez should be healthy, got %v and error %v", statusOK, err)
 	}
-	if !endpointReturnsStatusOK(client, "/readyz") {
-		t.Fatalf("readyz should be healthy")
+	if statusOK, err := endpointReturnsStatusOK(client, "/readyz"); err != nil || !statusOK {
+		t.Fatalf("readyz should be healthy, got %v and error %v", statusOK, err)
 	}
 }