Merge pull request #3734 from buildkite/improve-auto-mode-logging

Validate ping mode flag, tweak log levels

Author: DrJosh9000

agent/agent_worker.go

@@ -20,6 +20,12 @@ import (
 	"github.com/buildkite/roko"
 )
 
+const (
+	PingModeAuto       = "auto" // empty string can be used from tests
+	PingModeStreamOnly = "stream-only"
+	PingModePollOnly   = "poll-only"
+)
+
 type AgentWorkerConfig struct {
 	// Whether to set debug in the job
 	Debug bool
@@ -304,15 +310,23 @@ func (a *AgentWorker) Start(ctx context.Context, idleMon *idleMonitor) (startErr
 	streamingLoop := func() {
 		defer wg.Done()
 		err := a.runStreamingPingLoop(ctx, fromStreamingLoopCh)
-		if a.agentConfiguration.PingMode != "streaming-only" {
-			// If the ping mode is streaming-only, then an unrecoverable failure
-			// in the streaming loop should be reported.
-			// Otherwise, it should fall back to the ping loop and carry on,
-			// and if that also has an unrecoverable failure we can report that.
-			// Said another way:
-			// Streaming is best-effort but preferred, unless in streaming-only
-			// mode (it's the only available option).
-			err = nil
+		if err != nil {
+			switch a.agentConfiguration.PingMode {
+			case PingModeStreamOnly:
+				// In streaming-only mode, an unrecoverable failure
+				// in the streaming loop should be reported and should
+				// terminate the agent worker.
+				a.logger.Error("Streaming ping mode failed due to an unrecoverable error: %v", err)
+			default:
+				// In auto mode, the worker should fall back to the ping loop
+				// and carry on. The user might find that interesting (especially if
+				// they are expecting streaming to work).
+				a.logger.Info("Streaming ping mode is unavailable, permanently falling back to polling-based ping mode (the underlying error was: %v)", err)
+				// If the ping loop then has its own unrecoverable error, then
+				// *that* will terminate the worker. But the streaming loop shouldn't.
+				// So treat the error from the streaming loop as "business as usual".
+				err = nil
+			}
 		}
 		errCh <- err
 	}
@@ -323,18 +337,21 @@ func (a *AgentWorker) Start(ctx context.Context, idleMon *idleMonitor) (startErr
 
 	var loops []func()
 	switch a.agentConfiguration.PingMode {
-	case "", "auto":
+	case "", PingModeAuto: // note: "" can happen in some tests
 		loops = []func(){pingLoop, streamingLoop, debouncerLoop}
 		bat.Acquire(actorDebouncer)
 
-	case "ping-only":
+	case PingModePollOnly:
 		loops = []func(){pingLoop}
 		fromDebouncerCh = nil // prevent action loop listening to streaming side
 
-	case "stream-only":
+	case PingModeStreamOnly:
 		loops = []func(){streamingLoop, debouncerLoop}
 		fromPingLoopCh = nil // prevent action loop listening to ping side
 		bat.Acquire(actorDebouncer)
+
+	default:
+		return fmt.Errorf("unknown ping mode %q", a.agentConfiguration.PingMode)
 	}
 
 	// There's always an action handler.

agent/agent_worker_streaming.go

@@ -135,9 +135,11 @@ func (a *streamLoopState) startStream(ctx, streamCtx context.Context) error {
 	a.logger.Debug("[runStreamingPingLoop] Connecting (attempt %d)", a.attempts)
 	stream, err := a.apiClient.StreamPings(streamCtx, a.agent.UUID)
 	if err != nil {
-		a.logger.Error("Connection to ping stream failed: %v", err)
+		// TODO: after we've made streaming endpoints generally available,
+		// think about making some of these logs error or warning level.
+		a.logger.Debug("[runStreamingPingLoop] Connection to ping stream failed: %v", err)
 		if isUnrecoverable(err) {
-			a.logger.Error("Stopping ping stream because the error is unrecoverable")
+			a.logger.Debug("[runStreamingPingLoop] Stopping because the error is unrecoverable")
 			return err
 		}
 		// Fast fallback to the ping loop
@@ -181,9 +183,11 @@ func (a *streamLoopState) handle(ctx context.Context, msg *agentedgev1.StreamPin
 	var amsg actionMessage
 	switch {
 	case streamErr != nil:
+		// TODO: after we've made streaming endpoints generally available,
+		// think about making some of these logs error or warning level.
 		a.logger.Debug("[runStreamingPingLoop] Connection to ping stream failed or ended: %v", streamErr)
 		if isUnrecoverable(streamErr) {
-			a.logger.Error("Stopping ping stream loop because the error is unrecoverable: %v", streamErr)
+			a.logger.Debug("[runStreamingPingLoop] Stopping because the error is unrecoverable")
 			return streamErr
 		}
 		// Stay healthy if the error is deadline-exceeded.
@@ -197,7 +201,7 @@ func (a *streamLoopState) handle(ctx context.Context, msg *agentedgev1.StreamPin
 		amsg.unhealthy = true
 
 	case msg == nil:
-		a.logger.Error("Ping stream yielded a nil message, so assuming the stream is broken")
+		a.logger.Debug("[runStreamingPingLoop] Ping stream yielded a nil message, so assuming the stream is broken")
 		a.logger.Debug("[runStreamingPingLoop] Becoming unhealthy")
 		amsg.unhealthy = true
 

clicommand/agent_start.go

@@ -60,9 +60,18 @@ Example:
 
     $ buildkite-agent start --token xxx`
 
+const pingModePingOnly = "ping-only"
+
 var (
 	verificationFailureBehaviors = []string{agent.VerificationBehaviourBlock, agent.VerificationBehaviourWarn}
 
+	pingModes = []string{
+		agent.PingModeAuto,
+		agent.PingModePollOnly,
+		pingModePingOnly, // canonicalises to agent.PingModePollOnly
+		agent.PingModeStreamOnly,
+	}
+
 	buildkiteSetEnvironmentVariables = []*regexp.Regexp{
 		regexp.MustCompile("^BUILDKITE$"),
 		regexp.MustCompile("^BUILDKITE_.*$"),
@@ -764,8 +773,8 @@ var AgentStartCommand = cli.Command{
 		// API + agent behaviour
 		cli.StringFlag{
 			Name:   "ping-mode",
-			Usage:  "Selects available protocols for dispatching work to this agent. One of ping-only (default), auto (prefer streaming, but fall back to polling when necessary) or stream-only.",
-			Value:  "ping-only",
+			Usage:  "Selects available protocols for dispatching work to this agent. One of auto (default, prefer streaming, but fall back to polling when necessary), poll-only, or stream-only.",
+			Value:  "auto",
 			EnvVar: "BUILDKITE_AGENT_PING_MODE",
 		},
 
@@ -852,6 +861,15 @@ var AgentStartCommand = cli.Command{
 			return fmt.Errorf("failed to unset config from environment: %w", err)
 		}
 
+		if !slices.Contains(pingModes, cfg.PingMode) {
+			return fmt.Errorf("invalid ping mode %q, must be one of %v", cfg.PingMode, pingModes)
+		}
+		// Calling it "ping-only" was a mistake, so canonicalise it to "poll-only"
+		// on the very remote chance someone is using that.
+		if cfg.PingMode == pingModePingOnly {
+			cfg.PingMode = agent.PingModePollOnly
+		}
+
 		if cfg.VerificationJWKSFile != "" {
 			if !slices.Contains(verificationFailureBehaviors, cfg.VerificationFailureBehavior) {
 				return fmt.Errorf(

internal/e2e/basic_test.go

@@ -31,11 +31,11 @@ func TestBasicE2E(t *testing.T) {
 	}
 }
 
-func TestBasicE2E_PingOnly(t *testing.T) {
+func TestBasicE2E_PollOnly(t *testing.T) {
 	ctx := t.Context()
 	tc := newTestCase(t, "basic_e2e.yaml")
 
-	tc.startAgent("--ping-mode=ping-only")
+	tc.startAgent("--ping-mode=poll-only")
 	build := tc.triggerBuild()
 
 	// It should take much less time than 1 minute to successfully run the job.