act_runner/poller/poller.go

package poller

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"time"

	"gitea.com/gitea/act_runner/client"
	"gitea.com/gitea/act_runner/config"
	"gitea.com/gitea/act_runner/core"
	runnerv1 "gitea.com/gitea/proto-go/runner/v1"

	"github.com/appleboy/com/file"
	"github.com/bufbuild/connect-go"
	log "github.com/sirupsen/logrus"
)

const (
	errorRetryCounterLimit  = 3
	errorRetryTimeSleepSecs = 30
)

var (
	ErrDataLock   = errors.New("Data Lock Error")
	defaultLabels = []string{"self-hosted"}
)

func New(cli client.Client, dispatch func(context.Context, *runnerv1.Task) error, filter *client.Filter) *Poller {
	return &Poller{
		Client:       cli,
		Filter:       filter,
		Dispatch:     dispatch,
		routineGroup: newRoutineGroup(),
	}
}

type Poller struct {
	Client   client.Client
	Filter   *client.Filter
	Dispatch func(context.Context, *runnerv1.Task) error

	routineGroup      *routineGroup
	errorRetryCounter int
}

func (p *Poller) Register(ctx context.Context, cfg config.Runner) error {
	// check .runner config exist
	if file.IsFile(cfg.File) {
		return nil
	}

	// register new runner.
	resp, err := p.Client.Register(ctx, connect.NewRequest(&runnerv1.RegisterRequest{
		Name:         cfg.Name,
		Token:        cfg.Token,
		AgentLabels:  append(defaultLabels, []string{p.Filter.OS, p.Filter.Arch}...),
		CustomLabels: p.Filter.Labels,
	}))
	if err != nil {
		log.WithError(err).Error("poller: cannot register new runner")
		return err
	}

	data := &core.Runner{
		ID:    resp.Msg.Runner.Id,
		UUID:  resp.Msg.Runner.Uuid,
		Name:  resp.Msg.Runner.Name,
		Token: resp.Msg.Runner.Token,
	}

	file, err := json.MarshalIndent(data, "", "  ")
	if err != nil {
		log.WithError(err).Error("poller: cannot marshal the json input")
		return err
	}

	// store runner config in .runner file
	return os.WriteFile(cfg.File, file, 0o644)
}

func (p *Poller) Poll(ctx context.Context, n int) error {
	for i := 0; i < n; i++ {
		func(i int) {
			p.routineGroup.Run(func() {
				for {
					select {
					case <-ctx.Done():
						log.Infof("stopped the runner: %d", i+1)
						return
					default:
						if ctx.Err() != nil {
							log.Infof("stopping the runner: %d", i+1)
							return
						}
						if err := p.poll(ctx, i+1); err != nil {
							log.WithField("thread", i+1).
								WithError(err).Error("poll error")
							if p.errorRetryCounter > errorRetryCounterLimit {
								log.WithField("thread", i+1).Error("poller: too many errors, sleeping for 30 seconds")
								// FIXME: it makes ctrl+c hang up
								time.Sleep(time.Second * errorRetryTimeSleepSecs)
							}
						}
					}
				}
			})
		}(i)
	}
	p.routineGroup.Wait()
	return nil
}

func (p *Poller) poll(ctx context.Context, thread int) error {
	l := log.WithField("thread", thread)
	l.Info("poller: request stage from remote server")

	reqCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()

	// request a new build stage for execution from the central
	// build server.
	resp, err := p.Client.FetchTask(reqCtx, connect.NewRequest(&runnerv1.FetchTaskRequest{}))
	if err == context.Canceled || err == context.DeadlineExceeded {
		l.WithError(err).Trace("poller: no stage returned")
		p.errorRetryCounter++
		return nil
	}

	if err != nil && err == ErrDataLock {
		l.WithError(err).Info("task accepted by another runner")
		p.errorRetryCounter++
		return nil
	}

	if err != nil {
		l.WithError(err).Error("cannot accept task")
		p.errorRetryCounter++
		return err
	}

	// exit if a nil or empty stage is returned from the system
	// and allow the runner to retry.
	if resp.Msg.Task == nil || resp.Msg.Task.Id == 0 {
		return nil
	}

	p.errorRetryCounter = 0

	runCtx, cancel := context.WithTimeout(ctx, time.Hour)
	defer cancel()

	return p.Dispatch(runCtx, resp.Msg.Task)
}