Merge pull request #64946 from liggitt/log-healthz

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Add healthz check to ensure logging is not blocked

When running the apiserver/controllers in pods, we encountered a docker bug that blocked stdout/stderr (https://github.com/moby/moby/issues/31373)

That in turn blocked flushing logs, which in turn eventually blocked any goroutine that logs anything (which is pretty much all the important goroutines)

This adds a healthz check that logging is not blocked so that healthz indicates something is wrong

```release-note
NONE
```

pkg/kubelet/server/server.go

@@ -256,6 +256,7 @@ func (s *Server) InstallAuthFilter() {
 func (s *Server) InstallDefaultHandlers() {
 	healthz.InstallHandler(s.restfulCont,
 		healthz.PingHealthz,
+		healthz.LogHealthz,
 		healthz.NamedCheck("syncloop", s.syncLoopHealthCheck),
 	)
 	ws := new(restful.WebService)

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

@@ -255,7 +255,7 @@ func NewConfig(codecs serializer.CodecFactory) *Config {
 		HandlerChainWaitGroup:        new(utilwaitgroup.SafeWaitGroup),
 		LegacyAPIGroupPrefixes:       sets.NewString(DefaultLegacyAPIPrefix),
 		DisabledPostStartHooks:       sets.NewString(),
-		HealthzChecks:                []healthz.HealthzChecker{healthz.PingHealthz},
+		HealthzChecks:                []healthz.HealthzChecker{healthz.PingHealthz, healthz.LogHealthz},
 		EnableIndex:                  true,
 		EnableDiscovery:              true,
 		EnableProfiling:              true,

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

@@ -101,6 +101,7 @@ func TestNewWithDelegate(t *testing.T) {
     "/foo",
     "/healthz",
     "/healthz/delegate-health",
+    "/healthz/log",
     "/healthz/ping",
     "/healthz/poststarthook/delegate-post-start-hook",
     "/healthz/poststarthook/generic-apiserver-start-informers",
@@ -111,6 +112,7 @@ func TestNewWithDelegate(t *testing.T) {
   ]
 }`, t)
 	checkPath(server.URL+"/healthz", http.StatusInternalServerError, `[+]ping ok
+[+]log ok
 [-]wrapping-health failed: reason withheld
 [-]delegate-health failed: reason withheld
 [+]poststarthook/generic-apiserver-start-informers ok

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

@@ -20,7 +20,10 @@ go_library(
     ],
     importmap = "k8s.io/kubernetes/vendor/k8s.io/apiserver/pkg/server/healthz",
     importpath = "k8s.io/apiserver/pkg/server/healthz",
-    deps = ["//vendor/github.com/golang/glog:go_default_library"],
+    deps = [
+        "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+    ],
 )
 
 filegroup(

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

@@ -22,8 +22,12 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/golang/glog"
+
+	"k8s.io/apimachinery/pkg/util/wait"
 )
 
 // HealthzChecker is a named healthz checker.
@@ -56,6 +60,34 @@ func (ping) Check(_ *http.Request) error {
 	return nil
 }
 
+// LogHealthz returns true if logging is not blocked
+var LogHealthz HealthzChecker = &log{}
+
+type log struct {
+	startOnce    sync.Once
+	lastVerified atomic.Value
+}
+
+func (l *log) Name() string {
+	return "log"
+}
+
+func (l *log) Check(_ *http.Request) error {
+	l.startOnce.Do(func() {
+		l.lastVerified.Store(time.Now())
+		go wait.Forever(func() {
+			glog.Flush()
+			l.lastVerified.Store(time.Now())
+		}, time.Minute)
+	})
+
+	lastVerified := l.lastVerified.Load().(time.Time)
+	if time.Since(lastVerified) < (2 * time.Minute) {
+		return nil
+	}
+	return fmt.Errorf("logging blocked")
+}
+
 // NamedCheck returns a healthz checker for the given name and function.
 func NamedCheck(name string, check func(r *http.Request) error) HealthzChecker {
 	return &healthzCheck{name, check}