chore(gRPC): add request interface client

Signed-off-by: Bo-Yi.Wu <appleboy.tw@gmail.com>

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)
 }

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
+}

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

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=

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)
 }

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.ID", stage.Id).
-		WithField("runner.token", runner.Token)
+		WithField("runner.BuildID", stage.BuildId)
-
-	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)
 
 	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)
 }