Add config, logging for healthcheck

cmd/buildkitd/config/config.go

@@ -1,6 +1,8 @@
 package config
 
 import (
+	"time"
+
 	resolverconfig "github.com/moby/buildkit/util/resolver/config"
 )
 
@@ -24,6 +26,8 @@ type Config struct {
 	Registries map[string]resolverconfig.RegistryConfig `toml:"registry"`
 
 	DNS *DNSConfig `toml:"dns"`
+
+	Health HealthConfig `toml:"health"`
 }
 
 type GRPCConfig struct {
@@ -37,6 +41,12 @@ type GRPCConfig struct {
 	// MaxSendMsgSize int    `toml:"max_send_message_size"`
 }
 
+type HealthConfig struct {
+	Frequency       time.Duration `toml:"frequency"`
+	Timeout         time.Duration `toml:"timeout"`
+	AllowedFailures int           `toml:"allowedFailures"`
+}
+
 type TLSConfig struct {
 	Cert string `toml:"cert"`
 	Key  string `toml:"key"`

cmd/buildkitd/main.go

@@ -402,6 +402,18 @@ func setDefaultConfig(cfg *config.Config) {
 		cfg.GRPC.Address = []string{appdefaults.Address}
 	}
 
+	if cfg.Health.Frequency == 0 {
+		cfg.Health.Frequency = appdefaults.HealthFrequency
+	}
+
+	if cfg.Health.Timeout == 0 {
+		cfg.Health.Timeout = appdefaults.HealthTimeout
+	}
+
+	if cfg.Health.AllowedFailures == 0 {
+		cfg.Health.AllowedFailures = appdefaults.HealthAllowedFailures
+	}
+
 	if cfg.Workers.OCI.Platforms == nil {
 		cfg.Workers.OCI.Platforms = formatPlatforms(archutil.SupportedPlatforms(false))
 	}
@@ -612,7 +624,11 @@ func serverCredentials(cfg config.TLSConfig) (*tls.Config, error) {
 }
 
 func newController(c *cli.Context, cfg *config.Config) (*control.Controller, error) {
-	sessionManager, err := session.NewManager()
+	sessionManager, err := session.NewManager(&session.ManagerOpt{
+		HealthFrequency:       cfg.Health.Frequency,
+		HealthAllowedFailures: cfg.Health.AllowedFailures,
+		HealthTimeout:         cfg.Health.Timeout,
+	})
 	if err != nil {
 		return nil, err
 	}

docs/buildkitd.toml.md

@@ -38,6 +38,14 @@ insecure-entitlements = [ "network.host", "security.insecure" ]
     key = "/etc/buildkit/tls.key"
     ca = "/etc/buildkit/tlsca.crt"
 
+[health]
+  # How often to run the healthcheck
+  frequency = "10s"
+  # How long a healthcheck can take to respond
+  timeout = "1m"
+  # How many failures can be tolerated before cancelling
+  allowedFailures = 3
+
 [worker.oci]
   enabled = true
   # platforms is manually configure platforms, detected automatically if unset.

session/grpc.go

@@ -10,6 +10,7 @@ import (
 	"github.com/moby/buildkit/util/bklog"
 	"github.com/moby/buildkit/util/grpcerrors"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/net/http2"
@@ -27,7 +28,7 @@ func serve(ctx context.Context, grpcServer *grpc.Server, conn net.Conn) {
 	(&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{Handler: grpcServer})
 }
 
-func grpcClientConn(ctx context.Context, conn net.Conn) (context.Context, *grpc.ClientConn, error) {
+func grpcClientConn(ctx context.Context, conn net.Conn, healthCfg ManagerHealthCfg) (context.Context, *grpc.ClientConn, error) {
 	var unary []grpc.UnaryClientInterceptor
 	var stream []grpc.StreamClientInterceptor
 
@@ -70,29 +71,51 @@ func grpcClientConn(ctx context.Context, conn net.Conn) (context.Context, *grpc.
 	}
 
 	ctx, cancel := context.WithCancel(ctx)
-	go monitorHealth(ctx, cc, cancel)
+	go monitorHealth(ctx, cc, cancel, healthCfg)
 
 	return ctx, cc, nil
 }
 
-func monitorHealth(ctx context.Context, cc *grpc.ClientConn, cancelConn func()) {
+func monitorHealth(ctx context.Context, cc *grpc.ClientConn, cancelConn func(), healthCfg ManagerHealthCfg) {
 	defer cancelConn()
 	defer cc.Close()
 
-	ticker := time.NewTicker(1 * time.Second)
+	ticker := time.NewTicker(healthCfg.frequency)
 	defer ticker.Stop()
 	healthClient := grpc_health_v1.NewHealthClient(cc)
 
+	consecutiveFailures := 0
+
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			timeoutStart := time.Now().UTC()
+
+			ctx, cancel := context.WithTimeout(ctx, healthCfg.timeout)
 			_, err := healthClient.Check(ctx, &grpc_health_v1.HealthCheckRequest{})
 			cancel()
+
+			logFields := logrus.Fields{
+				"timeout":        healthCfg.timeout,
+				"actualDuration": time.Since(timeoutStart),
+			}
+
 			if err != nil {
-				return
+				consecutiveFailures++
+
+				logFields["allowedFailures"] = healthCfg.allowedFailures
+				logFields["consecutiveFailures"] = consecutiveFailures
+				bklog.G(ctx).WithFields(logFields).Warn("healthcheck failed")
+
+				if consecutiveFailures >= healthCfg.allowedFailures {
+					bklog.G(ctx).Error("healthcheck failed too many times")
+					return
+				}
+			} else {
+				bklog.G(ctx).WithFields(logFields).Debug("healthcheck completed")
+				consecutiveFailures = 0
 			}
 		}
 	}

session/manager.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/pkg/errors"
 	"google.golang.org/grpc"
@@ -31,12 +32,31 @@ type Manager struct {
 	sessions        map[string]*client
 	mu              sync.Mutex
 	updateCondition *sync.Cond
+	healthCfg       ManagerHealthCfg
+}
+
+// ManagerHealthCfg is the healthcheck configuration for gRPC healthchecks
+type ManagerHealthCfg struct {
+	frequency       time.Duration
+	timeout         time.Duration
+	allowedFailures int
+}
+
+type ManagerOpt struct {
+	HealthFrequency       time.Duration
+	HealthTimeout         time.Duration
+	HealthAllowedFailures int
 }
 
 // NewManager returns a new Manager
-func NewManager() (*Manager, error) {
+func NewManager(opt *ManagerOpt) (*Manager, error) {
 	sm := &Manager{
 		sessions: make(map[string]*client),
+		healthCfg: ManagerHealthCfg{
+			frequency:       opt.HealthFrequency,
+			timeout:         opt.HealthTimeout,
+			allowedFailures: opt.HealthAllowedFailures,
+		},
 	}
 	sm.updateCondition = sync.NewCond(&sm.mu)
 	return sm, nil
@@ -109,7 +129,7 @@ func (sm *Manager) handleConn(ctx context.Context, conn net.Conn, opts map[strin
 	name := h.Get(headerSessionName)
 	sharedKey := h.Get(headerSessionSharedKey)
 
-	ctx, cc, err := grpcClientConn(ctx, conn)
+	ctx, cc, err := grpcClientConn(ctx, conn, sm.healthCfg)
 	if err != nil {
 		sm.mu.Unlock()
 		return err

util/appdefaults/appdefaults_all.go

@@ -0,0 +1,9 @@
+package appdefaults
+
+import "time"
+
+const (
+	HealthAllowedFailures = 1
+	HealthFrequency       = 1 * time.Second
+	HealthTimeout         = 10 * time.Second
+)