Add to agent exponential backoff for fetching new tasks

agent/rpc/client_grpc.go

@@ -34,16 +34,28 @@ import (
 // set grpc version on compile time to compare against server version response
 const ClientGrpcVersion int32 = proto.Version
 
+// config for exponential backoff on grpc errors
+const (
+	backOffInit = 10 * time.Millisecond
+	backOffMax  = 10 * time.Second
+)
+
 type client struct {
-	client proto.WoodpeckerClient
-	conn   *grpc.ClientConn
+	client              proto.WoodpeckerClient
+	conn                *grpc.ClientConn
+	pollMaxInterval     time.Duration
+	pollInitialInterval time.Duration
 }
 
 // NewGrpcClient returns a new grpc Client.
 func NewGrpcClient(conn *grpc.ClientConn) rpc.Peer {
 	client := new(client)
 	client.client = proto.NewWoodpeckerClient(conn)
 	client.conn = conn
+
+	client.pollInitialInterval = backOffInit
+	client.pollMaxInterval = backOffMax
+
 	return client
 }
 
@@ -53,8 +65,8 @@ func (c *client) Close() error {
 
 func (c *client) newBackOff() backoff.BackOff {
 	b := backoff.NewExponentialBackOff()
-	b.MaxInterval = 10 * time.Second
-	b.InitialInterval = 10 * time.Millisecond
+	b.MaxInterval = c.pollMaxInterval
+	b.InitialInterval = c.pollInitialInterval
 	return b
 }
 

agent/runner.go

@@ -22,8 +22,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/rs/zerolog/log"
-	"github.com/tevino/abool/v2"
 	"google.golang.org/grpc/metadata"
 
 	"go.woodpecker-ci.org/woodpecker/v2/pipeline"
@@ -32,12 +32,24 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v2/shared/utils"
 )
 
+// config for exponential backoff on next task fetch
+const (
+	backOffInit = 10 * time.Millisecond
+	backOffMax  = 10 * time.Second
+)
+
 type Runner struct {
 	client   rpc.Peer
 	filter   rpc.Filter
 	hostname string
 	counter  *State
 	backend  *backend.Backend
+	backOff  backOff
+}
+
+type backOff struct {
+	pollMaxInterval     time.Duration
+	pollInitialInterval time.Duration
 }
 
 func NewRunner(workEngine rpc.Peer, f rpc.Filter, h string, state *State, backend *backend.Backend) Runner {
@@ -47,10 +59,37 @@ func NewRunner(workEngine rpc.Peer, f rpc.Filter, h string, state *State, backen
 		hostname: h,
 		counter:  state,
 		backend:  backend,
+		backOff: backOff{
+			pollInitialInterval: backOffInit,
+			pollMaxInterval:     backOffMax,
+		},
 	}
 }
 
 func (r *Runner) Run(runnerCtx context.Context) error {
+	retry := backoff.NewExponentialBackOff()
+	retry.MaxInterval = r.backOff.pollInitialInterval
+	retry.InitialInterval = r.backOff.pollMaxInterval
+
+	for {
+		select {
+		case <-runnerCtx.Done():
+			return runnerCtx.Err()
+		case <-time.After(retry.NextBackOff()):
+			log.Debug().Msg("polling new steps")
+			gotTask, err := r.runNextTask(runnerCtx)
+			if err != nil {
+				return err
+			}
+			if gotTask {
+				log.Debug().Msg("reset retry time, last task fetch got me work")
+				retry.Reset()
+			}
+		}
+	}
+}
+
+func (r *Runner) runNextTask(runnerCtx context.Context) (bool, error) {
 	log.Debug().Msg("request next execution")
 
 	meta, _ := metadata.FromOutgoingContext(runnerCtx)
@@ -59,10 +98,10 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 	// get the next workflow from the queue
 	work, err := r.client.Next(runnerCtx, r.filter)
 	if err != nil {
-		return err
+		return false, err
 	}
 	if work == nil {
-		return nil
+		return false, nil
 	}
 
 	timeout := time.Hour
@@ -89,8 +128,8 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 
 	logger.Debug().Msg("received execution")
 
-	workflowCtx, cancel := context.WithTimeout(ctxmeta, timeout)
-	defer cancel()
+	workflowCtx, cancelWorkflowCtx := context.WithTimeout(ctxmeta, timeout)
+	defer cancelWorkflowCtx()
 
 	// Add sigterm support for internal context.
 	// Required when the pipeline is terminated by external signals
@@ -99,15 +138,11 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 		logger.Error().Msg("Received sigterm termination signal")
 	})
 
-	canceled := abool.New()
 	go func() {
 		logger.Debug().Msg("listen for cancel signal")
 
 		if werr := r.client.Wait(workflowCtx, work.ID); werr != nil {
-			canceled.SetTo(true)
-			logger.Warn().Err(werr).Msg("cancel signal received")
-
-			cancel()
+			cancelWorkflowCtx()
 		} else {
 			logger.Debug().Msg("stop listening for cancel signal")
 		}
@@ -156,7 +191,7 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 	state.Finished = time.Now().Unix()
 	state.Exited = true
 
-	if canceled.IsSet() {
+	if workflowCtx.Err() != nil {
 		state.Error = ""
 		state.ExitCode = 137
 	} else if err != nil {
@@ -167,7 +202,7 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 		case errors.Is(err, pipeline.ErrCancel):
 			state.Error = ""
 			state.ExitCode = 137
-			canceled.SetTo(true)
+			cancelWorkflowCtx()
 		default:
 			state.ExitCode = 1
 			state.Error = err.Error()
@@ -177,7 +212,7 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 	logger.Debug().
 		Str("error", state.Error).
 		Int("exit_code", state.ExitCode).
-		Bool("canceled", canceled.IsSet()).
+		Bool("canceled", workflowCtx.Err() != nil).
 		Msg("pipeline complete")
 
 	logger.Debug().Msg("uploading logs")
@@ -195,7 +230,7 @@ func (r *Runner) Run(runnerCtx context.Context) error {
 		logger.Debug().Msg("updating pipeline status complete")
 	}
 
-	return nil
+	return true, nil
 }
 
 // extract repository name from the configuration

cmd/agent/agent.go

@@ -49,6 +49,8 @@ import (
 	"go.woodpecker-ci.org/woodpecker/v2/version"
 )
 
+const healthReportTimes = time.Second * 10
+
 func run(c *cli.Context) error {
 	agentConfigPath := c.String("agent-config")
 	hostname := c.String("hostname")
@@ -113,7 +115,6 @@ func run(c *cli.Context) error {
 
 	client := agentRpc.NewGrpcClient(conn)
 
-	sigterm := abool.New()
 	ctx := metadata.NewOutgoingContext(
 		context.Background(),
 		metadata.Pairs("hostname", hostname),
@@ -122,7 +123,6 @@ func run(c *cli.Context) error {
 	agentConfigPersisted := abool.New()
 	ctx = utils.WithContextSigtermCallback(ctx, func() {
 		log.Info().Msg("termination signal is received, shutting down")
-		sigterm.Set()
 
 		// Remove stateless agents from server
 		if agentConfigPersisted.IsNotSet() {
@@ -203,18 +203,23 @@ func run(c *cli.Context) error {
 
 	go func() {
 		for {
-			if sigterm.IsSet() {
-				log.Debug().Msg("terminating health reporting")
+			select {
+			case <-ctx.Done():
+				if err := ctx.Err(); err != nil && !errors.Is(err, &utils.ErrSignalReceived{}) {
+					log.Error().Err(err).Msgf("context closed with unexpected error")
+				} else {
+					log.Debug().Msg("terminating health reporting")
+				}
 				return
-			}
 
-			err := client.ReportHealth(ctx)
-			if err != nil {
-				log.Err(err).Msg("failed to report health")
-				return
-			}
+			default:
+				if err := client.ReportHealth(ctx); err != nil {
+					log.Err(err).Msg("failed to report health")
+					return
+				}
 
-			<-time.After(time.Second * 10)
+				<-time.After(healthReportTimes)
+			}
 		}
 	}()
 
@@ -226,17 +231,9 @@ func run(c *cli.Context) error {
 			r := agent.NewRunner(client, filter, hostname, counter, &backendEngine)
 			log.Debug().Msgf("created new runner %d", i)
 
-			for {
-				if sigterm.IsSet() {
-					log.Debug().Msgf("terminating runner %d", i)
-					return
-				}
-
-				log.Debug().Msg("polling new steps")
-				if err := r.Run(ctx); err != nil {
-					log.Error().Err(err).Msg("pipeline done with error")
-					return
-				}
+			if err := r.Run(ctx); err != nil && !errors.Is(err, &utils.ErrSignalReceived{}) {
+				log.Error().Err(err).Msg("runner done with error")
+				return
 			}
 		}()
 	}

shared/utils/context.go

@@ -22,6 +22,19 @@ import (
 	"syscall"
 )
 
+type ErrSignalReceived struct {
+	signal string
+}
+
+func (err *ErrSignalReceived) Error() string {
+	return fmt.Sprintf("received signal: %s", err.signal)
+}
+
+func (*ErrSignalReceived) Is(target error) bool {
+	_, ok := target.(*ErrSignalReceived) //nolint:errorlint
+	return ok
+}
+
 // Returns a copy of parent context that is canceled when
 // an os interrupt signal is received.
 func WithContextSigtermCallback(ctx context.Context, f func()) context.Context {
@@ -37,7 +50,7 @@ func WithContextSigtermCallback(ctx context.Context, f func()) context.Context {
 			if f != nil {
 				f()
 			}
-			cancel(fmt.Errorf("received signal: %v", receivedSignal))
+			cancel(&ErrSignalReceived{signal: fmt.Sprint(receivedSignal)})
 		}
 	}()