Wait for all informers to sync in /readyz.

This commit is contained in:
Maciej Borsz 2020-06-18 15:21:12 +02:00 committed by wojtekt
parent dcdeed97cd
commit 3f68000203
5 changed files with 63 additions and 16 deletions

View File

@ -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
}
}
}

View File

@ -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)
}
})

View File

@ -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",
],
)

View File

@ -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}

View File

@ -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)
}
}