chore(gRPC): register new runner

Signed-off-by: Bo-Yi.Wu <appleboy.tw@gmail.com>
pull/2/head
Bo-Yi.Wu 2022-08-28 14:05:56 +08:00 committed by Jason Song
parent a3e9bbed25
commit 449388f3ab
7 changed files with 70 additions and 46 deletions

View File

@ -6,11 +6,19 @@ import (
runnerv1 "gitea.com/gitea/proto-go/runner/v1" runnerv1 "gitea.com/gitea/proto-go/runner/v1"
) )
type Filter struct {
Kind string `json:"kind"`
Type string `json:"type"`
OS string `json:"os"`
Arch string `json:"arch"`
Capacity int `json:"capacity"`
}
// A Client manages communication with the runner. // A Client manages communication with the runner.
type Client interface { type Client interface {
// Ping sends a ping message to the server to test connectivity. // Ping sends a ping message to the server to test connectivity.
Ping(ctx context.Context, machine string) error Ping(ctx context.Context, machine string) error
// Request requests the next available build stage for execution. // Register for new runner.
Request(ctx context.Context) (*runnerv1.Stage, error) Register(ctx context.Context, args *runnerv1.RegisterRequest) (*runnerv1.Runner, error)
} }

View File

@ -59,6 +59,8 @@ func New(endpoint, secret string, skipverify bool, opts ...Option) *HTTPClient {
return client return client
} }
var _ Client = (*HTTPClient)(nil)
// An HTTPClient manages communication with the runner API. // An HTTPClient manages communication with the runner API.
type HTTPClient struct { type HTTPClient struct {
Client *http.Client Client *http.Client
@ -80,27 +82,26 @@ func (p *HTTPClient) Ping(ctx context.Context, machine string) error {
Data: machine, Data: machine,
}) })
req.Header().Set("X-Gitea-Token", p.Secret) req.Header().Set("X-Runner-Token", p.Secret)
_, err := client.Ping(ctx, req) _, err := client.Ping(ctx, req)
return err return err
} }
// Ping sends a ping message to the server to test connectivity. // Ping sends a ping message to the server to test connectivity.
func (p *HTTPClient) Request(ctx context.Context) (*runnerv1.Stage, error) { func (p *HTTPClient) Register(ctx context.Context, arg *runnerv1.RegisterRequest) (*runnerv1.Runner, error) {
client := runnerv1connect.NewRunnerServiceClient( client := runnerv1connect.NewRunnerServiceClient(
p.Client, p.Client,
p.Endpoint, p.Endpoint,
p.opts..., p.opts...,
) )
req := connect.NewRequest(&runnerv1.ConnectRequest{}) req := connect.NewRequest(arg)
req.Header().Set("X-Runner-Token", p.Secret)
req.Header().Set("X-Gitea-Token", p.Secret) res, err := client.Register(ctx, req)
res, err := client.Connect(ctx, req)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return res.Msg.Stage, err return res.Msg.Runner, err
} }

View File

@ -71,7 +71,17 @@ func runDaemon(ctx context.Context, task *runtime.Task) func(cmd *cobra.Command,
Environ: cfg.Runner.Environ, Environ: cfg.Runner.Environ,
} }
poller := poller.New(cli, runner.Run) poller := poller.New(
cli,
runner.Run,
&client.Filter{
Kind: runtime.Kind,
Type: runtime.Type,
OS: cfg.Platform.OS,
Arch: cfg.Platform.Arch,
Capacity: cfg.Runner.Capacity,
},
)
g.Go(func() error { g.Go(func() error {
log.WithField("capacity", cfg.Runner.Capacity). log.WithField("capacity", cfg.Runner.Capacity).
@ -80,8 +90,7 @@ func runDaemon(ctx context.Context, task *runtime.Task) func(cmd *cobra.Command,
WithField("arch", cfg.Platform.Arch). WithField("arch", cfg.Platform.Arch).
Infoln("polling the remote server") Infoln("polling the remote server")
poller.Poll(ctx, cfg.Runner.Capacity) return poller.Poll(ctx, cfg.Runner.Capacity)
return nil
}) })
err = g.Wait() err = g.Wait()

2
go.mod
View File

@ -3,7 +3,7 @@ module gitea.com/gitea/act_runner
go 1.18 go 1.18
require ( require (
gitea.com/gitea/proto-go v0.0.0-20220817054638-17fb0016dd41 gitea.com/gitea/proto-go v0.0.0-20220828031749-616e40329b57
github.com/bufbuild/connect-go v0.3.0 github.com/bufbuild/connect-go v0.3.0
github.com/docker/docker v20.10.17+incompatible github.com/docker/docker v20.10.17+incompatible
github.com/joho/godotenv v1.4.0 github.com/joho/godotenv v1.4.0

4
go.sum
View File

@ -27,6 +27,10 @@ gitea.com/gitea/act v0.0.0-20220812010840-c467b06265ee h1:T0wftx4RaYqbTH4t0A7bXG
gitea.com/gitea/act v0.0.0-20220812010840-c467b06265ee/go.mod h1:G37Vfz4J6kJ5NbcPI5xQUkeWPVkUCP5J+MFkaWU9jNY= gitea.com/gitea/act v0.0.0-20220812010840-c467b06265ee/go.mod h1:G37Vfz4J6kJ5NbcPI5xQUkeWPVkUCP5J+MFkaWU9jNY=
gitea.com/gitea/proto-go v0.0.0-20220817054638-17fb0016dd41 h1:FIGF6szYd3lBIwvbeedfU5Lc7uG1Xpzi7bkktS6Vdvg= gitea.com/gitea/proto-go v0.0.0-20220817054638-17fb0016dd41 h1:FIGF6szYd3lBIwvbeedfU5Lc7uG1Xpzi7bkktS6Vdvg=
gitea.com/gitea/proto-go v0.0.0-20220817054638-17fb0016dd41/go.mod h1:hD8YwSHusjwjEEgubW6XFvnZuNhMZTHz6lwjfltEt/Y= gitea.com/gitea/proto-go v0.0.0-20220817054638-17fb0016dd41/go.mod h1:hD8YwSHusjwjEEgubW6XFvnZuNhMZTHz6lwjfltEt/Y=
gitea.com/gitea/proto-go v0.0.0-20220828011358-d0a015a5b095 h1:Ng3GDJLYpsG3lYdaqDzeZFkRm5ShA2V+LWJSHRD0IQ0=
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=
github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc= 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 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8= github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=

View File

@ -2,7 +2,6 @@ package poller
import ( import (
"context" "context"
"errors"
"time" "time"
"gitea.com/gitea/act_runner/client" "gitea.com/gitea/act_runner/client"
@ -11,9 +10,10 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
func New(cli client.Client, dispatch func(context.Context, *runnerv1.Stage) error) *Poller { func New(cli client.Client, dispatch func(context.Context, *runnerv1.Runner) error, filter *client.Filter) *Poller {
return &Poller{ return &Poller{
Client: cli, Client: cli,
Filter: filter,
Dispatch: dispatch, Dispatch: dispatch,
routineGroup: newRoutineGroup(), routineGroup: newRoutineGroup(),
} }
@ -21,12 +21,24 @@ func New(cli client.Client, dispatch func(context.Context, *runnerv1.Stage) erro
type Poller struct { type Poller struct {
Client client.Client Client client.Client
Dispatch func(context.Context, *runnerv1.Stage) error Filter *client.Filter
Dispatch func(context.Context, *runnerv1.Runner) error
routineGroup *routineGroup routineGroup *routineGroup
} }
func (p *Poller) Poll(ctx context.Context, n int) { func (p *Poller) Poll(ctx context.Context, n int) error {
// register new runner.
runner, err := p.Client.Register(ctx, &runnerv1.RegisterRequest{
Os: p.Filter.OS,
Arch: p.Filter.Arch,
Capacity: int64(p.Filter.Capacity),
})
if err != nil {
log.WithError(err).Error("poller: cannot register new runner")
return err
}
for i := 0; i < n; i++ { for i := 0; i < n; i++ {
func(i int) { func(i int) {
p.routineGroup.Run(func() { p.routineGroup.Run(func() {
@ -40,7 +52,7 @@ func (p *Poller) Poll(ctx context.Context, n int) {
log.Infof("stopping the runner: %d", i+1) log.Infof("stopping the runner: %d", i+1)
return return
} }
if err := p.poll(ctx, i+1); err != nil { if err := p.poll(ctx, runner, i+1); err != nil {
log.WithError(err).Error("poll error") log.WithError(err).Error("poll error")
} }
} }
@ -49,29 +61,14 @@ func (p *Poller) Poll(ctx context.Context, n int) {
}(i) }(i)
} }
p.routineGroup.Wait() p.routineGroup.Wait()
return nil
} }
func (p *Poller) poll(ctx context.Context, thread int) error { func (p *Poller) poll(ctx context.Context, runner *runnerv1.Runner, thread int) error {
log.WithField("thread", thread).Info("poller: request stage from remote server") log.WithField("thread", thread).Info("poller: request stage from remote server")
// TODO: fetch the job from remote server // TODO: fetch the job from remote server
time.Sleep(time.Second) time.Sleep(time.Second)
// request a new build stage for execution from the central return p.Dispatch(ctx, runner)
// build server.
stage, err := p.Client.Request(ctx)
if errors.Is(err, context.Canceled) || errors.Is(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
}
if stage == nil || stage.BuildUuid == "" {
return nil
}
return p.Dispatch(ctx, stage)
} }

View File

@ -9,6 +9,12 @@ import (
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
) )
// Defines the Resource Kind and Type.
const (
Kind = "pipeline"
Type = "docker"
)
// Runner runs the pipeline. // Runner runs the pipeline.
type Runner struct { type Runner struct {
Machine string Machine string
@ -17,25 +23,24 @@ type Runner struct {
} }
// Run runs the pipeline stage. // Run runs the pipeline stage.
func (s *Runner) Run(ctx context.Context, stage *runnerv1.Stage) error { func (s *Runner) Run(ctx context.Context, runner *runnerv1.Runner) error {
l := logrus. l := logrus.
WithField("stage.build_uuid", stage.BuildUuid). WithField("runner.UUID", runner.Uuid).
WithField("stage.runner_uuid", stage.RunnerUuid) WithField("runner.token", runner.Token)
l.Info("stage received") l.Info("request a new task")
// TODO: Update stage structure // TODO: get new task
return s.run(ctx, stage) return s.run(ctx, runner)
} }
func (s *Runner) run(ctx context.Context, stage *runnerv1.Stage) error { func (s *Runner) run(ctx context.Context, runner *runnerv1.Runner) error {
l := logrus. l := logrus.
WithField("stage.build_uuid", stage.BuildUuid). WithField("runner.Uuid", runner.Uuid)
WithField("stage.runner_uuid", stage.RunnerUuid)
l.Info("start running pipeline") l.Info("start running pipeline")
// TODO: docker runner with stage data // TODO: docker runner with stage data
// task.Run is blocking, so we need to use goroutine to run it in backgroud // task.Run is blocking, so we need to use goroutine to run it in background
// return task metadata and status to the server // return task metadata and status to the server
task := NewTask() task := NewTask()
return task.Run(ctx) return task.Run(ctx)