From f2fb8798fa1556ad80d78a9649588a6089dfedcd Mon Sep 17 00:00:00 2001 From: "Bo-Yi.Wu" Date: Thu, 1 Sep 2022 15:09:22 +0800 Subject: [PATCH] chore(gRPC): add request interface client Signed-off-by: Bo-Yi.Wu --- client/client.go | 3 +++ client/http.go | 20 +++++++++++++++++++- go.mod | 2 +- go.sum | 2 ++ poller/poller.go | 35 +++++++++++++++++++++++++++++------ runtime/runtime.go | 17 ++++------------- 6 files changed, 58 insertions(+), 21 deletions(-) diff --git a/client/client.go b/client/client.go index fb1de0c..d8f4c3c 100644 --- a/client/client.go +++ b/client/client.go @@ -21,4 +21,7 @@ type Client interface { // Register for new runner. Register(ctx context.Context, args *runnerv1.RegisterRequest) (*runnerv1.Runner, error) + + // Request requests the next available build stage for execution. + Request(ctx context.Context, args *runnerv1.RequestRequest) (*runnerv1.Stage, error) } diff --git a/client/http.go b/client/http.go index c202fbd..3ae2bf1 100644 --- a/client/http.go +++ b/client/http.go @@ -88,7 +88,7 @@ func (p *HTTPClient) Ping(ctx context.Context, machine string) error { return err } -// Ping sends a ping message to the server to test connectivity. +// Register a new runner. func (p *HTTPClient) Register(ctx context.Context, arg *runnerv1.RegisterRequest) (*runnerv1.Runner, error) { client := runnerv1connect.NewRunnerServiceClient( p.Client, @@ -105,3 +105,21 @@ func (p *HTTPClient) Register(ctx context.Context, arg *runnerv1.RegisterRequest return res.Msg.Runner, err } + +// Request requests the next available build stage for execution. +func (p *HTTPClient) Request(ctx context.Context, arg *runnerv1.RequestRequest) (*runnerv1.Stage, error) { + client := runnerv1connect.NewRunnerServiceClient( + p.Client, + p.Endpoint, + p.opts..., + ) + req := connect.NewRequest(arg) + req.Header().Set("X-Runner-Token", p.Secret) + + res, err := client.Request(ctx, req) + if err != nil { + return nil, err + } + + return res.Msg.Stage, err +} diff --git a/go.mod b/go.mod index e767c3b..ac13f19 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-20220828031749-616e40329b57 + gitea.com/gitea/proto-go v0.0.0-20220901061207-b88901a1b9bc github.com/bufbuild/connect-go v0.3.0 github.com/docker/docker v20.10.17+incompatible github.com/joho/godotenv v1.4.0 diff --git a/go.sum b/go.sum index bf949c3..6a3e2df 100644 --- a/go.sum +++ b/go.sum @@ -31,6 +31,8 @@ gitea.com/gitea/proto-go v0.0.0-20220828011358-d0a015a5b095 h1:Ng3GDJLYpsG3lYdaq gitea.com/gitea/proto-go v0.0.0-20220828011358-d0a015a5b095/go.mod h1:hD8YwSHusjwjEEgubW6XFvnZuNhMZTHz6lwjfltEt/Y= gitea.com/gitea/proto-go v0.0.0-20220828031749-616e40329b57 h1:eVM6m3h5KpmJM2+LEqroENFaMs2kAo8QNIPyMgho9jg= gitea.com/gitea/proto-go v0.0.0-20220828031749-616e40329b57/go.mod h1:hD8YwSHusjwjEEgubW6XFvnZuNhMZTHz6lwjfltEt/Y= +gitea.com/gitea/proto-go v0.0.0-20220901061207-b88901a1b9bc h1:kTVjwKxXma2yAdgXz8T1tiJihtWFK8jGLqArX2NownM= +gitea.com/gitea/proto-go v0.0.0-20220901061207-b88901a1b9bc/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 e900f8d..17f2a7a 100644 --- a/poller/poller.go +++ b/poller/poller.go @@ -10,7 +10,7 @@ import ( log "github.com/sirupsen/logrus" ) -func New(cli client.Client, dispatch func(context.Context, *runnerv1.Runner) error, filter *client.Filter) *Poller { +func New(cli client.Client, dispatch func(context.Context, *runnerv1.Stage) error, filter *client.Filter) *Poller { return &Poller{ Client: cli, Filter: filter, @@ -22,14 +22,14 @@ func New(cli client.Client, dispatch func(context.Context, *runnerv1.Runner) err type Poller struct { Client client.Client Filter *client.Filter - Dispatch func(context.Context, *runnerv1.Runner) error + Dispatch func(context.Context, *runnerv1.Stage) error routineGroup *routineGroup } func (p *Poller) Poll(ctx context.Context, n int) error { // register new runner. - runner, err := p.Client.Register(ctx, &runnerv1.RegisterRequest{ + _, err := p.Client.Register(ctx, &runnerv1.RegisterRequest{ Os: p.Filter.OS, Arch: p.Filter.Arch, Capacity: int64(p.Filter.Capacity), @@ -52,7 +52,7 @@ func (p *Poller) Poll(ctx context.Context, n int) error { log.Infof("stopping the runner: %d", i+1) return } - if err := p.poll(ctx, runner, i+1); err != nil { + if err := p.poll(ctx, i+1); err != nil { log.WithError(err).Error("poll error") } } @@ -64,11 +64,34 @@ func (p *Poller) Poll(ctx context.Context, n int) error { return nil } -func (p *Poller) poll(ctx context.Context, runner *runnerv1.Runner, thread int) error { +func (p *Poller) poll(ctx context.Context, thread int) error { log.WithField("thread", thread).Info("poller: request stage from remote server") // TODO: fetch the job from remote server time.Sleep(time.Second) - return p.Dispatch(ctx, runner) + // request a new build stage for execution from the central + // build server. + stage, err := p.Client.Request(ctx, &runnerv1.RequestRequest{ + Kind: p.Filter.Kind, + Os: p.Filter.OS, + Arch: p.Filter.Arch, + Type: p.Filter.Type, + }) + if err == context.Canceled || err == context.DeadlineExceeded { + log.WithError(err).Trace("poller: no stage returned") + return nil + } + if err != nil { + log.WithError(err).Error("poller: cannot request stage") + return err + } + + // exit if a nil or empty stage is returned from the system + // and allow the runner to retry. + if stage == nil || stage.Id == 0 { + return nil + } + + return p.Dispatch(ctx, stage) } diff --git a/runtime/runtime.go b/runtime/runtime.go index bc3e783..b8d51e8 100644 --- a/runtime/runtime.go +++ b/runtime/runtime.go @@ -23,25 +23,16 @@ type Runner struct { } // Run runs the pipeline stage. -func (s *Runner) Run(ctx context.Context, runner *runnerv1.Runner) error { +func (s *Runner) Run(ctx context.Context, stage *runnerv1.Stage) error { l := logrus. - WithField("runner.UUID", runner.Uuid). - WithField("runner.token", runner.Token) - - l.Info("request a new task") - // TODO: get new task - - return s.run(ctx, runner) -} - -func (s *Runner) run(ctx context.Context, runner *runnerv1.Runner) error { - l := logrus. - WithField("runner.Uuid", runner.Uuid) + WithField("runner.ID", stage.Id). + WithField("runner.BuildID", stage.BuildId) l.Info("start running pipeline") // TODO: docker runner with stage data // task.Run is blocking, so we need to use goroutine to run it in background // return task metadata and status to the server task := NewTask() + return task.Run(ctx) }