112 lines
2.4 KiB
Go
112 lines
2.4 KiB
Go
package taskscheduler
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
runnerv1 "gitea.dev/actionslib/runner/v1"
|
|
)
|
|
|
|
type PollClient interface {
|
|
Declare(context.Context, string, []string) error
|
|
FetchTask(context.Context, int64) (*runnerv1.FetchTaskResponse, error)
|
|
}
|
|
|
|
type PollerConfig struct {
|
|
Version string
|
|
Labels []string
|
|
EmptyBackoff time.Duration
|
|
ErrorBackoff time.Duration
|
|
}
|
|
|
|
// Poller is the scheduler component. Once Gitea assigns a task, it never
|
|
// fetches another one until the current assignment is durably dispatched.
|
|
type Poller struct {
|
|
Client PollClient
|
|
Scheduler *Scheduler
|
|
Config PollerConfig
|
|
OnError func(error)
|
|
}
|
|
|
|
func (p Poller) Run(ctx context.Context) error {
|
|
if p.Client == nil || p.Scheduler == nil {
|
|
return errors.New("Gitea poll client and task scheduler are required")
|
|
}
|
|
if p.Config.Version == "" || len(p.Config.Labels) == 0 {
|
|
return errors.New("runner version and labels are required")
|
|
}
|
|
if err := p.Client.Declare(ctx, p.Config.Version, p.Config.Labels); err != nil {
|
|
return fmt.Errorf("declare scheduler labels: %w", err)
|
|
}
|
|
|
|
emptyBackoff := p.Config.EmptyBackoff
|
|
if emptyBackoff <= 0 {
|
|
emptyBackoff = time.Second
|
|
}
|
|
errorBackoff := p.Config.ErrorBackoff
|
|
if errorBackoff <= 0 {
|
|
errorBackoff = 5 * time.Second
|
|
}
|
|
var tasksVersion int64
|
|
for {
|
|
response, err := p.Client.FetchTask(ctx, tasksVersion)
|
|
if err != nil {
|
|
if ctx.Err() != nil {
|
|
return nil
|
|
}
|
|
p.report(fmt.Errorf("fetch Gitea task: %w", err))
|
|
if !wait(ctx, errorBackoff) {
|
|
return nil
|
|
}
|
|
continue
|
|
}
|
|
if response == nil {
|
|
p.report(errors.New("fetch Gitea task returned an empty response"))
|
|
if !wait(ctx, errorBackoff) {
|
|
return nil
|
|
}
|
|
continue
|
|
}
|
|
tasksVersion = response.GetTasksVersion()
|
|
task := response.GetTask()
|
|
if task == nil {
|
|
if !wait(ctx, emptyBackoff) {
|
|
return nil
|
|
}
|
|
continue
|
|
}
|
|
for {
|
|
if err := p.Scheduler.Run(ctx, task); err == nil {
|
|
break
|
|
} else {
|
|
if ctx.Err() != nil {
|
|
return nil
|
|
}
|
|
p.report(fmt.Errorf("dispatch Gitea task %d: %w", task.GetId(), err))
|
|
}
|
|
if !wait(ctx, errorBackoff) {
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (p Poller) report(err error) {
|
|
if p.OnError != nil {
|
|
p.OnError(err)
|
|
}
|
|
}
|
|
|
|
func wait(ctx context.Context, duration time.Duration) bool {
|
|
timer := time.NewTimer(duration)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-ctx.Done():
|
|
return false
|
|
case <-timer.C:
|
|
return true
|
|
}
|
|
}
|