Merge pull request #103437 from p0lyn0mial/readyz-signal

genericapiserver: adds HasBeenReady lifecycle signal

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

@@ -103,7 +103,10 @@ func (s *GenericAPIServer) installReadyz() {
 	s.readyzLock.Lock()
 	defer s.readyzLock.Unlock()
 	s.readyzChecksInstalled = true
-	healthz.InstallReadyzHandler(s.Handler.NonGoRestfulMux, s.readyzChecks...)
+	healthz.InstallReadyzHandlerWithHealthyFunc(s.Handler.NonGoRestfulMux, func() {
+		// note: InstallReadyzHandlerWithHealthyFunc guarantees that this is called only once
+		s.lifecycleSignals.HasBeenReady.Signal()
+	}, s.readyzChecks...)
 }
 
 // installLivez creates the livez endpoint for this server.

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

@@ -140,6 +140,12 @@ func InstallReadyzHandler(mux mux, checks ...HealthChecker) {
 	InstallPathHandler(mux, "/readyz", checks...)
 }
 
+// InstallReadyzHandlerWithHealthyFunc is like InstallReadyzHandler, but in addition call firstTimeReady
+// the first time /readyz succeeds.
+func InstallReadyzHandlerWithHealthyFunc(mux mux, firstTimeReady func(), checks ...HealthChecker) {
+	InstallPathHandlerWithHealthyFunc(mux, "/readyz", firstTimeReady, checks...)
+}
+
 // InstallLivezHandler registers handlers for liveness checking on the path
 // "/livez" to mux. *All handlers* for mux must be specified in
 // exactly one call to InstallHandler. Calling InstallHandler more
@@ -154,6 +160,12 @@ func InstallLivezHandler(mux mux, checks ...HealthChecker) {
 // InstallPathHandler more than once for the same path and mux will
 // result in a panic.
 func InstallPathHandler(mux mux, path string, checks ...HealthChecker) {
+	InstallPathHandlerWithHealthyFunc(mux, path, nil, checks...)
+}
+
+// InstallPathHandlerWithHealthyFunc is like InstallPathHandler, but calls firstTimeHealthy exactly once
+// when the handler succeeds for the first time.
+func InstallPathHandlerWithHealthyFunc(mux mux, path string, firstTimeHealthy func(), checks ...HealthChecker) {
 	if len(checks) == 0 {
 		klog.V(5).Info("No default health checks specified. Installing the ping handler.")
 		checks = []HealthChecker{PingHealthz}
@@ -172,7 +184,7 @@ func InstallPathHandler(mux mux, path string, checks ...HealthChecker) {
 			/* component = */ "",
 			/* deprecated */ false,
 			/* removedRelease */ "",
-			handleRootHealth(name, checks...)))
+			handleRootHealth(name, firstTimeHealthy, checks...)))
 	for _, check := range checks {
 		mux.Handle(fmt.Sprintf("%s/%v", path, check.Name()), adaptCheckToHandler(check.Check))
 	}
@@ -209,8 +221,9 @@ func getExcludedChecks(r *http.Request) sets.String {
 }
 
 // handleRootHealth returns an http.HandlerFunc that serves the provided checks.
-func handleRootHealth(name string, checks ...HealthChecker) http.HandlerFunc {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+func handleRootHealth(name string, firstTimeHealthy func(), checks ...HealthChecker) http.HandlerFunc {
+	var notifyOnce sync.Once
+	return func(w http.ResponseWriter, r *http.Request) {
 		excluded := getExcludedChecks(r)
 		// failedVerboseLogOutput is for output to the log.  It indicates detailed failed output information for the log.
 		var failedVerboseLogOutput bytes.Buffer
@@ -246,6 +259,11 @@ func handleRootHealth(name string, checks ...HealthChecker) http.HandlerFunc {
 			return
 		}
 
+		// signal first time this is healthy
+		if firstTimeHealthy != nil {
+			notifyOnce.Do(firstTimeHealthy)
+		}
+
 		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
 		w.Header().Set("X-Content-Type-Options", "nosniff")
 		if _, found := r.URL.Query()["verbose"]; !found {
@@ -255,7 +273,7 @@ func handleRootHealth(name string, checks ...HealthChecker) http.HandlerFunc {
 
 		individualCheckOutput.WriteTo(w)
 		fmt.Fprintf(w, "%s check passed\n", name)
-	})
+	}
 }
 
 // adaptCheckToHandler returns an http.HandlerFunc that serves the provided checks.

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

@@ -17,6 +17,7 @@ limitations under the License.
 package healthz
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net/http"
@@ -311,3 +312,66 @@ type cacheSyncWaiterStub struct {
 func (s cacheSyncWaiterStub) WaitForCacheSync(_ <-chan struct{}) map[reflect.Type]bool {
 	return s.startedByInformerType
 }
+
+func TestInstallReadyzHandlerWithHealthyFunc(t *testing.T) {
+	mux := http.NewServeMux()
+	readyzCh := make(chan struct{})
+
+	hasBeenReadyCounter := 0
+	hasBeenReadyFn := func() {
+		hasBeenReadyCounter++
+	}
+	InstallReadyzHandlerWithHealthyFunc(mux, hasBeenReadyFn, readyOnChanClose{readyzCh})
+
+	// scenario 1: expect the check to fail since the channel hasn't been closed
+	req, err := http.NewRequest("GET", fmt.Sprintf("http://example.com%s", "/readyz"), nil)
+	if err != nil {
+		t.Errorf("%v", err)
+	}
+	rr := httptest.NewRecorder()
+	mux.ServeHTTP(rr, req)
+	if rr.Code != http.StatusInternalServerError {
+		t.Errorf("scenario 1: unexpected status code returned, expected %d, got %d", http.StatusInternalServerError, rr.Code)
+	}
+
+	// scenario 2: close the channel that will cause the readyz checker to report success,
+	//             verify that hasBeenReadyFn was called
+	close(readyzCh)
+	rr = httptest.NewRecorder()
+	req = req.Clone(context.TODO())
+	mux.ServeHTTP(rr, req)
+	if rr.Code != http.StatusOK {
+		t.Errorf("scenario 2: unexpected status code returned, expected %d, got %d", http.StatusOK, rr.Code)
+	}
+	if hasBeenReadyCounter != 1 {
+		t.Errorf("scenario 2: unexpected value of hasBeenReadyCounter, expected 1, got %d", hasBeenReadyCounter)
+	}
+
+	// scenario 3: checks if hasBeenReadyFn hasn't been called again.
+	rr = httptest.NewRecorder()
+	req = req.Clone(context.TODO())
+	mux.ServeHTTP(rr, req)
+	if rr.Code != http.StatusOK {
+		t.Errorf("scenario 3: unexpected status code returned, expected %d, got %d", http.StatusOK, rr.Code)
+	}
+	if hasBeenReadyCounter != 1 {
+		t.Errorf("scenario 3: unexpected value of hasBeenReadyCounter, expected 1, got %d", hasBeenReadyCounter)
+	}
+}
+
+type readyOnChanClose struct {
+	ch <-chan struct{}
+}
+
+func (readyOnChanClose) Name() string {
+	return "readyOnChanClose"
+}
+
+func (c readyOnChanClose) Check(_ *http.Request) error {
+	select {
+	case <-c.ch:
+		return nil
+	default:
+	}
+	return fmt.Errorf("the provided channel hasn't been closed")
+}

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

@@ -26,6 +26,7 @@ Events:
 - ShutdownInitiated: KILL signal received
 - AfterShutdownDelayDuration: shutdown delay duration has passed
 - InFlightRequestsDrained: all in flight request(s) have been drained
+- HasBeenReady is signaled when the readyz endpoint succeeds for the first time
 
 The following is a sequence of shutdown events that we expect to see during termination:
 T0: ShutdownInitiated: KILL signal received
@@ -95,6 +96,9 @@ type lifecycleSignals struct {
 	// HTTPServerStoppedListening termination event is signaled when the
 	// HTTP Server has stopped listening to the underlying socket.
 	HTTPServerStoppedListening lifecycleSignal
+
+	// HasBeenReady is signaled when the readyz endpoint succeeds for the first time.
+	HasBeenReady lifecycleSignal
 }
 
 // newLifecycleSignals returns an instance of lifecycleSignals interface to be used
@@ -105,6 +109,7 @@ func newLifecycleSignals() lifecycleSignals {
 		AfterShutdownDelayDuration: newNamedChannelWrapper("AfterShutdownDelayDuration"),
 		InFlightRequestsDrained:    newNamedChannelWrapper("InFlightRequestsDrained"),
 		HTTPServerStoppedListening: newNamedChannelWrapper("HTTPServerStoppedListening"),
+		HasBeenReady:               newNamedChannelWrapper("HasBeenReady"),
 	}
 }