diff --git a/cmd/daemon.go b/cmd/daemon.go index 0dd8cf8..bd64041 100644 --- a/cmd/daemon.go +++ b/cmd/daemon.go @@ -79,8 +79,6 @@ func runDaemon(ctx context.Context, task *runtime.Task) func(cmd *cobra.Command, cli, runner.Run, &client.Filter{ - Kind: runtime.Kind, - Type: runtime.Type, OS: cfg.Platform.OS, Arch: cfg.Platform.Arch, Capacity: cfg.Runner.Capacity, diff --git a/go.mod b/go.mod index 03564de..ffdf51e 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module gitea.com/gitea/act_runner go 1.18 require ( - gitea.com/gitea/proto-go v0.0.0-20220925101213-1ac8a05257e1 + gitea.com/gitea/proto-go v0.0.0-20220929140437-812ae50fdce4 github.com/avast/retry-go/v4 v4.1.0 github.com/bufbuild/connect-go v0.5.0 github.com/docker/docker v20.10.17+incompatible diff --git a/go.sum b/go.sum index ef85cc7..55309d5 100644 --- a/go.sum +++ b/go.sum @@ -27,6 +27,8 @@ gitea.com/gitea/act v0.0.0-20220922135643-52a5bba9e7fa h1:HHqlvfIvqFlny3sgJgAM1B gitea.com/gitea/act v0.0.0-20220922135643-52a5bba9e7fa/go.mod h1:9W/Nz16tjfnWp7O5DUo3EjZBnZFBI/5rlWstX4o7+hU= gitea.com/gitea/proto-go v0.0.0-20220925101213-1ac8a05257e1 h1:JGApntYc07jawNxrxv1WhU6IHX0i73nqhloZlaUR5Nc= gitea.com/gitea/proto-go v0.0.0-20220925101213-1ac8a05257e1/go.mod h1:hD8YwSHusjwjEEgubW6XFvnZuNhMZTHz6lwjfltEt/Y= +gitea.com/gitea/proto-go v0.0.0-20220929140437-812ae50fdce4 h1:HW38qGi3yd/7eUk8ihkz+opF6YGb1uLn8d1ZCUaxNg8= +gitea.com/gitea/proto-go v0.0.0-20220929140437-812ae50fdce4/go.mod h1:hD8YwSHusjwjEEgubW6XFvnZuNhMZTHz6lwjfltEt/Y= github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc= github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8= github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8= diff --git a/poller/poller.go b/poller/poller.go index 77132ed..1a4770d 100644 --- a/poller/poller.go +++ b/poller/poller.go @@ -2,6 +2,7 @@ package poller import ( "context" + "errors" "time" "gitea.com/gitea/act_runner/client" @@ -11,6 +12,8 @@ import ( log "github.com/sirupsen/logrus" ) +var ErrDataLock = errors.New("Data Lock Error") + func New(cli client.Client, dispatch func(context.Context, *runnerv1.Task) error, filter *client.Filter) *Poller { return &Poller{ Client: cli, @@ -67,25 +70,32 @@ func (p *Poller) Poll(ctx context.Context, n int) error { } func (p *Poller) poll(ctx context.Context, thread int) error { - logger := log.WithField("thread", thread) - logger.Info("poller: request stage from remote server") + l := log.WithField("thread", thread) + l.Info("poller: request stage from remote server") ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() // request a new build stage for execution from the central // build server. - resp, err := p.Client.Request(ctx, connect.NewRequest(&runnerv1.RequestRequest{ + resp, err := p.Client.FetchTask(ctx, connect.NewRequest(&runnerv1.FetchTaskRequest{ Kind: p.Filter.Kind, Os: p.Filter.OS, Arch: p.Filter.Arch, Type: p.Filter.Type, })) if err == context.Canceled || err == context.DeadlineExceeded { - logger.WithError(err).Trace("poller: no stage returned") + l.WithError(err).Trace("poller: no stage returned") return nil } + + if err != nil && err == ErrDataLock { + l.WithError(err).Info("task accepted by another runner") + return nil + } + if err != nil { + l.WithError(err).Error("cannot accept task") return err } diff --git a/runtime/runtime.go b/runtime/runtime.go index 86484f2..37b16c5 100644 --- a/runtime/runtime.go +++ b/runtime/runtime.go @@ -2,23 +2,13 @@ package runtime import ( "context" - "errors" "gitea.com/gitea/act_runner/client" runnerv1 "gitea.com/gitea/proto-go/runner/v1" - "github.com/bufbuild/connect-go" log "github.com/sirupsen/logrus" ) -var ErrDataLock = errors.New("Data Lock Error") - -// Defines the Resource Kind and Type. -const ( - Kind = "pipeline" - Type = "docker" -) - // Runner runs the pipeline. type Runner struct { Machine string @@ -30,28 +20,7 @@ type Runner struct { func (s *Runner) Run(ctx context.Context, task *runnerv1.Task) error { l := log. WithField("task.id", task.Id) - l.Info("start running pipeline") - // update machine in stage - task.Machine = s.Machine - data, err := s.Client.Detail(ctx, connect.NewRequest(&runnerv1.DetailRequest{ - Task: task, - })) - if err != nil && err == ErrDataLock { - l.Info("task accepted by another runner") - return nil - } - if err != nil { - l.WithError(err).Error("cannot accept task") - return err - } - - l.Info("task details fetched") - - return s.run(ctx, data.Msg.Task) -} - -func (s *Runner) run(ctx context.Context, task *runnerv1.Task) error { return NewTask(task.Id, s.Client).Run(ctx, task) }