ddmt/gitdts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(runner): update to latest proto format.

Browse Source 
Signed-off-by: Bo-Yi.Wu <appleboy.tw@gmail.com>
This commit is contained in:
Bo-Yi.Wu
2022-10-10 13:58:47 +08:00
committed by Jason Song
parent f355188e12
commit 4dd135580f
6 changed files with 56 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
routers/api/bots/runner/runner.go
View File

@@ -10,7 +10,6 @@ import (
"code.gitea.io/gitea/core"
bots_model "code.gitea.io/gitea/models/bots"
repo_model "code.gitea.io/gitea/models/repo"
"code.gitea.io/gitea/modules/log"
runnerv1 "gitea.com/gitea/proto-go/runner/v1"
"gitea.com/gitea/proto-go/runner/v1/runnerv1connect"
@@ -18,6 +17,8 @@ import (
"github.com/bufbuild/connect-go"
)
var _ runnerv1connect.RunnerServiceClient = (*Service)(nil)
type Service struct {
Scheduler core.Scheduler
@@ -38,25 +39,24 @@ func (s *Service) Register(
return nil, errors.New("missing runner token")
}
// TODO: Get token data from runner_token table
runner, err := bots_model.GetRunnerByToken(token)
if err != nil {
return nil, errors.New("runner not found")
}
// update runner information
runner.Arch = req.Msg.Arch
runner.OS = req.Msg.Os
runner.Capacity = req.Msg.Capacity
if err := bots_model.UpdateRunner(ctx, runner, []string{"arch", "os", "capacity"}...); err != nil {
runner.AgentLabels = req.Msg.AgentLabels
runner.CustomLabels = req.Msg.CustomLabels
runner.Name = req.Msg.Name
if err := bots_model.UpdateRunner(ctx, runner, []string{"name", "agent_labels", "custom_labels"}...); err != nil {
return nil, errors.New("can't update runner")
}
res := connect.NewResponse(&runnerv1.RegisterResponse{
Runner: &runnerv1.Runner{
Uuid: runner.UUID,
Os: req.Msg.Os,
Arch: req.Msg.Arch,
Capacity: req.Msg.Capacity,
Uuid: runner.UUID,
Token: runner.Token,
},
})
@@ -64,15 +64,13 @@ func (s *Service) Register(
}
// Request requests the next available build stage for execution.
func (s *Service) Request(
func (s *Service) FetchTask(
ctx context.Context,
req *connect.Request[runnerv1.RequestRequest],
) (*connect.Response[runnerv1.RequestResponse], error) {
req *connect.Request[runnerv1.FetchTaskRequest],
) (*connect.Response[runnerv1.FetchTaskResponse], error) {
log.Debug("manager: request queue item")
stage, err := s.Scheduler.Request(ctx, core.Filter{
Kind: req.Msg.Kind,
Type: req.Msg.Type,
task, err := s.Scheduler.Request(ctx, core.Filter{
OS: req.Msg.Os,
Arch: req.Msg.Arch,
})
@@ -85,86 +83,29 @@ func (s *Service) Request(
return nil, err
}
res := connect.NewResponse(&runnerv1.RequestResponse{
Stage: stage,
// TODO: update task and check data lock
task.Machine = req.Msg.Os
res := connect.NewResponse(&runnerv1.FetchTaskResponse{
Task: task,
})
return res, nil
}
// Details fetches build details
func (s *Service) Detail(
// UpdateTask updates the task status.
func (s *Service) UpdateTask(
ctx context.Context,
req *connect.Request[runnerv1.DetailRequest],
) (*connect.Response[runnerv1.DetailResponse], error) {
log.Info("stag id %d", req.Msg.Stage.Id)
// fetch stage data
stage, err := bots_model.GetStageByID(req.Msg.Stage.Id)
if err != nil {
return nil, err
}
stage.Machine = req.Msg.Stage.Machine
stage.Status = core.StatusPending
count, err := bots_model.UpdateBuildStage(stage, "machine", "status")
if err != nil {
return nil, err
}
if count != 1 {
return nil, core.ErrDataLock
}
// fetch build data
build, err := bots_model.GetBuildByID(stage.BuildID)
if err != nil {
return nil, err
}
// fetch repo data
repo, err := repo_model.GetRepositoryByID(build.RepoID)
if err != nil {
return nil, err
}
res := connect.NewResponse(&runnerv1.DetailResponse{
Stage: &runnerv1.Stage{
Id: stage.ID,
BuildId: stage.BuildID,
Name: stage.Name,
Kind: stage.Kind,
Type: stage.Type,
Status: string(stage.Status),
Started: int64(stage.Started),
Stopped: int64(stage.Stopped),
Machine: stage.Machine,
},
Build: &runnerv1.Build{
Id: build.ID,
Name: build.Name,
},
Repo: &runnerv1.Repo{
Id: repo.ID,
Name: repo.Name,
},
})
req *connect.Request[runnerv1.UpdateTaskRequest],
) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
res := connect.NewResponse(&runnerv1.UpdateTaskResponse{})
return res, nil
}
// Update updates the build stage.
func (s *Service) Update(
// UpdateLog uploads log of the task.
func (s *Service) UpdateLog(
ctx context.Context,
req *connect.Request[runnerv1.UpdateRequest],
) (*connect.Response[runnerv1.UpdateResponse], error) {
res := connect.NewResponse(&runnerv1.UpdateResponse{})
return res, nil
}
// UpdateStep updates the build step.
func (s *Service) UpdateStep(
ctx context.Context,
req *connect.Request[runnerv1.UpdateStepRequest],
) (*connect.Response[runnerv1.UpdateStepResponse], error) {
res := connect.NewResponse(&runnerv1.UpdateStepResponse{})
req *connect.Request[runnerv1.UpdateLogRequest],
) (*connect.Response[runnerv1.UpdateLogResponse], error) {
res := connect.NewResponse(&runnerv1.UpdateLogResponse{})
return res, nil
}

26
routers/api/bots/scheduler/queue/queue.go
View File

@@ -19,7 +19,7 @@ type worker struct {
typ string
os string
arch string
channel chan *runnerv1.Stage
channel chan *runnerv1.Task
}
type queue struct {
@@ -32,7 +32,7 @@ type queue struct {
ctx context.Context
}
func (q *queue) Schedule(ctx context.Context, stage *runnerv1.Stage) error {
func (q *queue) Schedule(ctx context.Context, stage *runnerv1.Task) error {
select {
case q.ready <- struct{}{}:
default:
@@ -40,13 +40,13 @@ func (q *queue) Schedule(ctx context.Context, stage *runnerv1.Stage) error {
return nil
}
func (q *queue) Request(ctx context.Context, params core.Filter) (*runnerv1.Stage, error) {
func (q *queue) Request(ctx context.Context, params core.Filter) (*runnerv1.Task, error) {
w := &worker{
kind: params.Kind,
typ: params.Type,
os: params.OS,
arch: params.Arch,
channel: make(chan *runnerv1.Stage),
channel: make(chan *runnerv1.Task),
}
q.Lock()
q.workers[w] = struct{}{}
@@ -123,15 +123,15 @@ func (q *queue) signal(ctx context.Context) error {
}
}
stage := &runnerv1.Stage{
Id: item.ID,
BuildId: item.BuildID,
Name: item.Name,
Kind: item.Name,
Type: item.Type,
Status: string(item.Status),
Started: int64(item.Started),
Stopped: int64(item.Stopped),
stage := &runnerv1.Task{
Id: item.ID,
// BuildId: item.BuildID,
// Name: item.Name,
// Kind: item.Name,
// Type: item.Type,
// Status: string(item.Status),
// Started: int64(item.Started),
// Stopped: int64(item.Stopped),
}
w.channel <- stage