Add BBS-connectivity health check to auctioneer

jobs/auctioneer/spec

@@ -89,6 +89,27 @@ properties:
     description: "Timeout in seconds to receive a response to the keepalive ping. If a response is not received within this time, the locket client will reconnect to another server."
     default: 22
 
+  # BBS-connectivity health check for the auctioneer. Mirrors the
+  # semantics of diego.bbs.enable_db_health_check but for the auction
+  # path. When enabled, the auctioneer exits on N consecutive failed
+  # BBS Ping()s so monit can restart it and its Locket lock can be
+  # picked up by a healthy standby. Fixes the pathology where a
+  # DNS-partitioned auctioneer retains its Locket lock indefinitely
+  # because renewals travel a persistent gRPC connection while auction
+  # work needs fresh DNS. See bbs_health_check_runner.go.
+  diego.auctioneer.enable_bbs_health_check:
+    description: "Enable a runner that probes BBS and exits the auctioneer on repeated failures, releasing the Locket lock cleanly for standby pickup."
+    default: false
+  diego.auctioneer.bbs_health_check_interval:
+    description: "Interval between BBS health-check probes."
+    default: "10s"
+  diego.auctioneer.bbs_health_check_timeout:
+    description: "Per-probe timeout for BBS health-check pings."
+    default: "5s"
+  diego.auctioneer.bbs_health_check_failure_threshold:
+    description: "Number of consecutive failed probes before the auctioneer exits."
+    default: 3
+
   loggregator.v2_api_port:
     description: "Local metron agent gRPC port"
     default: 3458

jobs/auctioneer/templates/auctioneer.json.erb

@@ -77,6 +77,14 @@
     raise "The locket client keepalive time property should not be larger than the timeout"
   end
 
+  # BBS health check.
+  if p("diego.auctioneer.enable_bbs_health_check")
+    config[:enable_bbs_health_check] = true
+    config[:bbs_health_check_interval] = p("diego.auctioneer.bbs_health_check_interval")
+    config[:bbs_health_check_timeout] = p("diego.auctioneer.bbs_health_check_timeout")
+    config[:bbs_health_check_failure_threshold] = p("diego.auctioneer.bbs_health_check_failure_threshold")
+  end
+
   config[:loggregator]={}
   config[:loggregator][:loggregator_api_port] = p("loggregator.v2_api_port")
   config[:loggregator][:loggregator_ca_path] = "#{conf_dir}/certs/loggregator/ca.crt"

src/code.cloudfoundry.org/auctioneer/cmd/auctioneer/bbs_health_check_runner.go

@@ -0,0 +1,130 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"os"
+	"time"
+
+	"code.cloudfoundry.org/bbs"
+	"code.cloudfoundry.org/clock"
+	"code.cloudfoundry.org/lager/v3"
+)
+
+// bbsHealthCheckRunner periodically probes BBS connectivity from the
+// auctioneer host and exits (via ifrit error) when N consecutive checks
+// fail. This lets monit restart the auctioneer, whose normal SIGTERM
+// path cleanly releases the Locket lock — allowing a healthy-AZ
+// standby to acquire leadership.
+//
+// The auctioneer's Locket lock is renewed
+// over a persistent gRPC connection (DNS-free) while auction work
+// (fetching cell reps from BBS) resolves bbs.service.cf.internal on
+// every call. Under a DNS-only partition, renewals succeed and the
+// leader retains the lock indefinitely despite being functionally
+// dead. A dedicated health check breaks the tie by making
+// functional-path degradation observable to the process supervisor.
+type bbsHealthCheckRunner struct {
+	logger           lager.Logger
+	bbsClient        bbs.InternalClient
+	clock            clock.Clock
+	interval         time.Duration
+	timeout          time.Duration
+	failureThreshold int
+}
+
+func newBBSHealthCheckRunner(
+	logger lager.Logger,
+	bbsClient bbs.InternalClient,
+	clk clock.Clock,
+	interval, timeout time.Duration,
+	failureThreshold int,
+) *bbsHealthCheckRunner {
+	return &bbsHealthCheckRunner{
+		logger:           logger.Session("bbs-health-check-runner"),
+		bbsClient:        bbsClient,
+		clock:            clk,
+		interval:         interval,
+		timeout:          timeout,
+		failureThreshold: failureThreshold,
+	}
+}
+
+// Run implements ifrit.Runner. Signals a ready channel immediately;
+// on each tick, probes BBS. On threshold consecutive failures, returns
+// a non-nil error so the ifrit group tears down and monit restarts.
+func (r *bbsHealthCheckRunner) Run(signals <-chan os.Signal, ready chan<- struct{}) error {
+	logger := r.logger.Session("run")
+	logger.Info("starting", lager.Data{
+		"interval":          r.interval.String(),
+		"timeout":           r.timeout.String(),
+		"failure_threshold": r.failureThreshold,
+	})
+
+	close(ready)
+
+	ticker := r.clock.NewTicker(r.interval)
+	defer ticker.Stop()
+
+	failures := 0
+	for {
+		select {
+		case <-signals:
+			logger.Info("received-signal")
+			return nil
+
+		case <-ticker.C():
+			if r.probe(logger) {
+				if failures > 0 {
+					logger.Info("health-check-recovered", lager.Data{
+						"consecutive_failures": failures,
+					})
+				}
+				failures = 0
+				continue
+			}
+
+			failures++
+			logger.Error("health-check-failed", nil, lager.Data{
+				"failures":  failures,
+				"threshold": r.failureThreshold,
+			})
+
+			if failures >= r.failureThreshold {
+				err := errors.New("bbs connectivity degraded")
+				logger.Error("bbs-connectivity-degraded-restarting-auctioneer",
+					err, lager.Data{"failures": failures})
+				return err
+			}
+		}
+	}
+}
+
+// probe performs one BBS Ping with an enforced timeout. Returns true
+// on success. The ping uses the same bbs.Client the auctioneer uses
+// for real work, so it exercises the same DNS/TLS/HTTP path — any
+// degradation visible to auction requests is visible here.
+func (r *bbsHealthCheckRunner) probe(logger lager.Logger) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), r.timeout)
+	defer cancel()
+
+	done := make(chan bool, 1)
+	go func() {
+		// bbs.InternalClient.Ping does not accept a context directly,
+		// but the underlying HTTP client honors the per-request
+		// timeout configured on the auctioneer's cfhttp client. The
+		// context.WithTimeout above enforces an outer bound so a
+		// truly hung Ping cannot stall the health check goroutine.
+		done <- r.bbsClient.Ping(logger, "auctioneer-bbs-health-check")
+	}()
+
+	select {
+	case ok := <-done:
+		return ok
+	case <-ctx.Done():
+		logger.Error("probe-timeout", ctx.Err(), lager.Data{
+			"timeout": r.timeout.String(),
+		})
+		return false
+	}
+}

src/code.cloudfoundry.org/auctioneer/cmd/auctioneer/bbs_health_check_runner_test.go

@@ -0,0 +1,126 @@
+package main
+
+import (
+	"os"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"code.cloudfoundry.org/bbs"
+	"code.cloudfoundry.org/clock/fakeclock"
+	"code.cloudfoundry.org/lager/v3"
+	"code.cloudfoundry.org/lager/v3/lagertest"
+)
+
+// fakeBBSClient is a minimal stand-in for bbs.InternalClient used to
+// exercise the health-check probe path without spinning up a full BBS.
+// Only Ping() is implemented meaningfully; all other methods panic if
+// called (the runner should not invoke them).
+type fakeBBSClient struct {
+	pingResult atomic.Bool
+	pings      atomic.Int64
+	bbs.InternalClient
+}
+
+func newFakeBBSClient(initialResult bool) *fakeBBSClient {
+	c := &fakeBBSClient{}
+	c.pingResult.Store(initialResult)
+	return c
+}
+
+func (f *fakeBBSClient) Ping(_ lager.Logger, _ string) bool {
+	f.pings.Add(1)
+	return f.pingResult.Load()
+}
+
+// TestBBSHealthCheckRunner_ExitsOnThreshold verifies the runner exits
+// (returns a non-nil error) after `failureThreshold` consecutive
+// failed probes.
+func TestBBSHealthCheckRunner_ExitsOnThreshold(t *testing.T) {
+	logger := lagertest.NewTestLogger("test")
+	bbsClient := newFakeBBSClient(false)
+	fakeClock := fakeclock.NewFakeClock(time.Now())
+
+	r := newBBSHealthCheckRunner(logger, bbsClient, fakeClock,
+		100*time.Millisecond, 50*time.Millisecond, 3)
+
+	signals := make(chan os.Signal)
+	ready := make(chan struct{})
+	errCh := make(chan error, 1)
+	go func() { errCh <- r.Run(signals, ready) }()
+
+	<-ready
+
+	// Drive the fake clock forward. WaitForWatcherAndIncrement blocks
+	// until the goroutine calls NewTicker/Sleep on the fake clock,
+	// then fires the tick. We give each tick a moment to be processed.
+	for i := 0; i < 3; i++ {
+		fakeClock.WaitForWatcherAndIncrement(100 * time.Millisecond)
+		// Poll for the probe to be observed (fakeclock fires the
+		// channel synchronously, but the goroutine still needs a
+		// scheduling slice to consume it and run probe()).
+		deadline := time.Now().Add(500 * time.Millisecond)
+		for bbsClient.pings.Load() <= int64(i) && time.Now().Before(deadline) {
+			time.Sleep(2 * time.Millisecond)
+		}
+	}
+
+	select {
+	case err := <-errCh:
+		if err == nil {
+			t.Fatal("expected non-nil error on threshold breach")
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatalf("runner did not exit within 2s of threshold breach; pings=%d", bbsClient.pings.Load())
+	}
+	if bbsClient.pings.Load() < 3 {
+		t.Fatalf("expected >=3 pings, got %d", bbsClient.pings.Load())
+	}
+}
+
+// TestBBSHealthCheckRunner_RecoversOnSuccess verifies that a
+// successful probe resets the failure counter and the runner
+// continues without exiting.
+func TestBBSHealthCheckRunner_RecoversOnSuccess(t *testing.T) {
+	logger := lagertest.NewTestLogger("test")
+	bbsClient := newFakeBBSClient(false)
+	fakeClock := fakeclock.NewFakeClock(time.Now())
+
+	r := newBBSHealthCheckRunner(logger, bbsClient, fakeClock,
+		100*time.Millisecond, 50*time.Millisecond, 3)
+
+	signals := make(chan os.Signal, 1)
+	ready := make(chan struct{})
+	errCh := make(chan error, 1)
+	go func() { errCh <- r.Run(signals, ready) }()
+	<-ready
+
+	// Two failures then flip to success and tick once more; the runner
+	// must not exit even though the total number of failed probes
+	// (2) is less than the threshold (3).
+	for i := 0; i < 2; i++ {
+		fakeClock.WaitForWatcherAndIncrement(100 * time.Millisecond)
+		deadline := time.Now().Add(500 * time.Millisecond)
+		for bbsClient.pings.Load() <= int64(i) && time.Now().Before(deadline) {
+			time.Sleep(2 * time.Millisecond)
+		}
+	}
+	bbsClient.pingResult.Store(true)
+	fakeClock.WaitForWatcherAndIncrement(100 * time.Millisecond)
+	// Give the success probe a scheduling slice.
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for bbsClient.pings.Load() < 3 && time.Now().Before(deadline) {
+		time.Sleep(2 * time.Millisecond)
+	}
+
+	// Should still be running; send signal to shut down cleanly.
+	signals <- os.Interrupt
+	select {
+	case err := <-errCh:
+		if err != nil {
+			t.Fatalf("expected clean exit on signal, got %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("runner did not exit after signal")
+	}
+}

src/code.cloudfoundry.org/auctioneer/cmd/auctioneer/config/config.go

@@ -37,6 +37,19 @@ type AuctioneerConfig struct {
 	StartingContainerCountMaximum int                   `json:"starting_container_count_maximum,omitempty"`
 	StartingContainerWeight       float64               `json:"starting_container_weight,omitempty"`
 	UUID                          string                `json:"uuid,omitempty"`
+
+	// BBS connectivity health check.
+	// When EnableBBSHealthCheck is true, a runner probes BBS every
+	// BBSHealthCheckInterval with per-probe timeout
+	// BBSHealthCheckTimeout. After BBSHealthCheckFailureThreshold
+	// consecutive failures the auctioneer exits, releasing its
+	// Locket lock cleanly via SIGTERM so a healthy peer can
+	// acquire leadership.
+	EnableBBSHealthCheck           bool                  `json:"enable_bbs_health_check,omitempty"`
+	BBSHealthCheckInterval         durationjson.Duration `json:"bbs_health_check_interval,omitempty"`
+	BBSHealthCheckTimeout          durationjson.Duration `json:"bbs_health_check_timeout,omitempty"`
+	BBSHealthCheckFailureThreshold int                   `json:"bbs_health_check_failure_threshold,omitempty"`
+
 	debugserver.DebugServerConfig
 	lagerflags.LagerConfig
 	locket.ClientLocketConfig

src/code.cloudfoundry.org/auctioneer/cmd/auctioneer/main.go

@@ -129,6 +129,25 @@ func main() {
 		{Name: "auction-server", Runner: auctionServer},
 	}
 
+	// Optional BBS-connectivity health check. When enabled,
+	// the auctioneer exits on N consecutive failed pings to BBS so
+	// monit can restart it and its Locket lock can be picked up by a
+	// healthy-AZ standby. See bbs_health_check_runner.go.
+	if cfg.EnableBBSHealthCheck {
+		bbsHealthCheck := newBBSHealthCheckRunner(
+			logger,
+			initializeBBSClient(logger, cfg),
+			clock,
+			time.Duration(cfg.BBSHealthCheckInterval),
+			time.Duration(cfg.BBSHealthCheckTimeout),
+			cfg.BBSHealthCheckFailureThreshold,
+		)
+		members = append(members, grouper.Member{
+			Name:   "bbs-health-check",
+			Runner: bbsHealthCheck,
+		})
+	}
+
 	if cfg.DebugAddress != "" {
 		members = append(grouper.Members{
 			{Name: "debug-server", Runner: debugserver.Runner(cfg.DebugAddress, reconfigurableSink)},