act_runner/poller/poller.go

156 lines
3.7 KiB
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)
}