refactor: runner

This commit is contained in:
Jason Song committed 2023-04-04 17:04:19 +08:00
1 parent 284dae0ba4
commit 4db566ef4c
8 files changed
+192 -307

No files matched your search

+1 -20
View File
@@ -13,7 +13,6 @@ import (
"github.com/spf13/cobra"
"golang.org/x/sync/errgroup"
"gitea.com/gitea/act_runner/internal/pkg/artifactcache"
"gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/envcheck"
@@ -70,25 +69,7 @@ func runDaemon(ctx context.Context, configFile *string) func(cmd *cobra.Command,
version,
)
runner := &runtime.Runner{
Client: cli,
Machine: reg.Name,
ForgeInstance: reg.Address,
Environ: cfg.Runner.Envs,
Labels: ls,
Network: cfg.Container.Network,
Version: version,
}
if *cfg.Cache.Enabled {
if handler, err := artifactcache.NewHandler(cfg.Cache.Dir, cfg.Cache.Host, cfg.Cache.Port); err != nil {
log.Errorf("cannot init cache server, it will be disabled: %v", err)
} else {
log.Infof("cache handler listens on: %v", handler.ExternalURL())
runner.CacheHandler = handler
}
}
runner := runtime.NewRunner(cfg, reg, version)
poller := poller.New(
cli,
runner.Run,
+1 -1
View File
@@ -348,7 +348,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
}
// init a cache server
handler, err := artifactcache.NewHandler("", "", 0)
handler, err := artifactcache.StartHandler("", "", 0)
if err != nil {
return err
}
+1 -1
View File
@@ -42,7 +42,7 @@ type Handler struct {
outboundIP string
}
func NewHandler(dir, outboundIP string, port uint16) (*Handler, error) {
func StartHandler(dir, outboundIP string, port uint16) (*Handler, error) {
h := &Handler{}
if dir == "" {
+2 -2
View File
@@ -38,5 +38,5 @@ cache:
port: 0
container:
# Which network to use for the job containers.
network: bridge
# Which network to use for the job containers. Could be bridge, host, none, or the name of a custom network.
network_mode: bridge
+3 -3
View File
@@ -32,7 +32,7 @@ type Config struct {
Port uint16 `yaml:"port"`
} `yaml:"cache"`
Container struct {
Network string `yaml:"network"`
NetworkMode string `yaml:"network_mode"`
}
}
@@ -87,8 +87,8 @@ func LoadDefault(file string) (*Config, error) {
cfg.Cache.Dir = filepath.Join(home, ".cache", "actcache")
}
}
if cfg.Container.Network == "" {
cfg.Container.Network = "bridge"
if cfg.Container.NetworkMode == "" {
cfg.Container.NetworkMode = "bridge"
}
return cfg, nil
+1
View File
@@ -179,6 +179,7 @@ func (r *Reporter) Close(lastWords string) error {
v.Result = runnerv1.Result_RESULT_CANCELLED
}
}
r.state.Result = runnerv1.Result_RESULT_FAILURE
r.logRows = append(r.logRows, &runnerv1.LogRow{
Time: timestamppb.Now(),
Content: lastWords,
+183 -16
View File
@@ -4,35 +4,202 @@
package runtime
import (
"bytes"
"context"
"encoding/json"
"fmt"
"path/filepath"
"sync"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"github.com/nektos/act/pkg/common"
"github.com/nektos/act/pkg/model"
"github.com/nektos/act/pkg/runner"
log "github.com/sirupsen/logrus"
"gitea.com/gitea/act_runner/internal/pkg/artifactcache"
"gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/config"
"gitea.com/gitea/act_runner/internal/pkg/labels"
"gitea.com/gitea/act_runner/internal/pkg/report"
)
// Runner runs the pipeline.
type Runner struct {
Machine string
Version string
ForgeInstance string
Environ map[string]string
Client client.Client
Labels labels.Labels
Network string
CacheHandler *artifactcache.Handler
name string
version string
cfg *config.Config
client client.Client
labels labels.Labels
envs map[string]string
runningTasks sync.Map
}
// Run runs the pipeline stage.
func (s *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
env := map[string]string{}
for k, v := range s.Environ {
env[k] = v
func NewRunner(cfg *config.Config, reg *config.Registration, version string) *Runner {
cli := client.New(reg.Address, cfg.Runner.Insecure, reg.UUID, reg.Token, version)
ls := labels.Labels{}
for _, v := range reg.Labels {
if l, err := labels.Parse(v); err == nil {
ls = append(ls, l)
}
}
if s.CacheHandler != nil {
env["ACTIONS_CACHE_URL"] = s.CacheHandler.ExternalURL() + "/"
envs := make(map[string]string, len(cfg.Runner.Envs))
for k, v := range cfg.Runner.Envs {
envs[k] = v
}
if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled {
cacheHandler, err := artifactcache.StartHandler(cfg.Cache.Dir, cfg.Cache.Host, cfg.Cache.Port)
if err != nil {
log.Errorf("cannot init cache server, it will be disabled: %v", err)
// go on
} else {
envs["ACTIONS_CACHE_URL"] = cacheHandler.ExternalURL() + "/"
}
}
return &Runner{
name: reg.Name,
version: version,
cfg: cfg,
client: cli,
labels: ls,
envs: envs,
}
return NewTask(task.Id, s.Client, env, s.Network, s.Labels.PickPlatform).Run(ctx, task, s.Machine, s.Version)
}
func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error {
if _, ok := r.runningTasks.Load(task.Id); ok {
return fmt.Errorf("task %d is already running", task.Id)
} else {
r.runningTasks.Store(task.Id, struct{}{})
defer r.runningTasks.Delete(task.Id)
}
ctx, cancel := context.WithCancel(ctx)
defer cancel()
reporter := report.NewReporter(ctx, cancel, r.client, task)
var runErr error
defer func() {
lastWords := ""
if runErr != nil {
lastWords = runErr.Error()
}
_ = reporter.Close(lastWords)
}()
reporter.RunDaemon()
runErr = r.run(ctx, task, reporter)
return nil
}
func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.Reporter) error {
reporter.Logf("%s(version:%s) received task %v of job %v, be triggered by event: %s", r.name, r.version, task.Id, task.Context.Fields["job"].GetStringValue(), task.Context.Fields["event_name"].GetStringValue())
workflow, err := model.ReadWorkflow(bytes.NewReader(task.WorkflowPayload))
if err != nil {
return err
}
jobIDs := workflow.GetJobIDs()
if len(jobIDs) != 1 {
return fmt.Errorf("multiple jobs found: %v", jobIDs)
}
jobID := jobIDs[0]
plan, err := model.CombineWorkflowPlanner(workflow).PlanJob(jobID)
if err != nil {
return err
}
job := workflow.GetJob(jobID)
reporter.ResetSteps(len(job.Steps))
taskContext := task.Context.Fields
log.Infof("task %v repo is %v %v %v", task.Id, taskContext["repository"].GetStringValue(),
taskContext["gitea_default_actions_url"].GetStringValue(),
r.client.Address())
preset := &model.GithubContext{
Event: taskContext["event"].GetStructValue().AsMap(),
RunID: taskContext["run_id"].GetStringValue(),
RunNumber: taskContext["run_number"].GetStringValue(),
Actor: taskContext["actor"].GetStringValue(),
Repository: taskContext["repository"].GetStringValue(),
EventName: taskContext["event_name"].GetStringValue(),
Sha: taskContext["sha"].GetStringValue(),
Ref: taskContext["ref"].GetStringValue(),
RefName: taskContext["ref_name"].GetStringValue(),
RefType: taskContext["ref_type"].GetStringValue(),
HeadRef: taskContext["head_ref"].GetStringValue(),
BaseRef: taskContext["base_ref"].GetStringValue(),
Token: taskContext["token"].GetStringValue(),
RepositoryOwner: taskContext["repository_owner"].GetStringValue(),
RetentionDays: taskContext["retention_days"].GetStringValue(),
}
if t := task.Secrets["GITEA_TOKEN"]; t != "" {
preset.Token = t
} else if t := task.Secrets["GITHUB_TOKEN"]; t != "" {
preset.Token = t
}
eventJSON, err := json.Marshal(preset.Event)
if err != nil {
return err
}
maxLifetime := 3 * time.Hour
if deadline, ok := ctx.Deadline(); ok {
maxLifetime = time.Until(deadline)
}
config := &runner.Config{
// On Linux, Workdir will be like "/<owner>/<repo>"
// On Windows, Workdir will be like "\<owner>\<repo>"
Workdir: filepath.FromSlash(string(filepath.Separator) + preset.Repository),
BindWorkdir: false,
ReuseContainers: false,
ForcePull: false,
ForceRebuild: false,
LogOutput: true,
JSONLogger: false,
Env: r.envs,
Secrets: task.Secrets,
InsecureSecrets: false,
Privileged: false,
UsernsMode: "",
ContainerArchitecture: "",
ContainerDaemonSocket: "",
UseGitIgnore: false,
GitHubInstance: r.client.Address(),
ContainerCapAdd: nil,
ContainerCapDrop: nil,
AutoRemove: true,
ArtifactServerPath: "",
ArtifactServerPort: "",
NoSkipCheckout: true,
PresetGitHubContext: preset,
EventJSON: string(eventJSON),
ContainerNamePrefix: fmt.Sprintf("GITEA-ACTIONS-TASK-%d", task.Id),
ContainerMaxLifetime: maxLifetime,
ContainerNetworkMode: r.cfg.Container.NetworkMode,
DefaultActionInstance: taskContext["gitea_default_actions_url"].GetStringValue(),
PlatformPicker: r.labels.PickPlatform,
}
rr, err := runner.New(config)
if err != nil {
return err
}
executor := rr.NewPlanExecutor(plan)
reporter.Logf("workflow prepared")
// add logger recorders
ctx = common.WithLoggerHook(ctx, reporter)
return executor(ctx)
}
-264
View File
@@ -2,267 +2,3 @@
// SPDX-License-Identifier: MIT
package runtime
import (
"bytes"
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"sync"
"time"
runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
"github.com/nektos/act/pkg/common"
"github.com/nektos/act/pkg/model"
"github.com/nektos/act/pkg/runner"
log "github.com/sirupsen/logrus"
"gitea.com/gitea/act_runner/internal/pkg/client"
"gitea.com/gitea/act_runner/internal/pkg/report"
)
var globalTaskMap sync.Map
type TaskInput struct {
repoDirectory string
// actor string
// workdir string
// workflowsPath string
// autodetectEvent bool
// eventPath string
// reuseContainers bool
// bindWorkdir bool
// secrets []string
envs map[string]string
// platforms []string
// dryrun bool
forcePull bool
forceRebuild bool
// noOutput bool
// envfile string
// secretfile string
insecureSecrets bool
// defaultBranch string
privileged bool
usernsMode string
containerArchitecture string
containerDaemonSocket string
// noWorkflowRecurse bool
useGitIgnore bool
containerCapAdd []string
containerCapDrop []string
// autoRemove bool
artifactServerPath string
artifactServerPort string
jsonLogger bool
// noSkipCheckout bool
// remoteName string
EnvFile string
containerNetworkMode string
}
type Task struct {
BuildID int64
Input *TaskInput
client client.Client
log *log.Entry
platformPicker func([]string) string
}
// NewTask creates a new task
func NewTask(buildID int64, client client.Client, runnerEnvs map[string]string, network string, picker func([]string) string) *Task {
task := &Task{
Input: &TaskInput{
envs: runnerEnvs,
containerNetworkMode: network,
},
BuildID: buildID,
client: client,
log: log.WithField("buildID", buildID),
platformPicker: picker,
}
task.Input.repoDirectory, _ = os.Getwd()
return task
}
// getWorkflowsPath return the workflows directory, it will try .gitea first and then fallback to .github
func getWorkflowsPath(dir string) (string, error) {
p := filepath.Join(dir, ".gitea/workflows")
_, err := os.Stat(p)
if err != nil {
if !os.IsNotExist(err) {
return "", err
}
return filepath.Join(dir, ".github/workflows"), nil
}
return p, nil
}
func getToken(task *runnerv1.Task) string {
token := task.Secrets["GITHUB_TOKEN"]
if task.Secrets["GITEA_TOKEN"] != "" {
token = task.Secrets["GITEA_TOKEN"]
}
if task.Context.Fields["token"].GetStringValue() != "" {
token = task.Context.Fields["token"].GetStringValue()
}
return token
}
func (t *Task) Run(ctx context.Context, task *runnerv1.Task, runnerName, runnerVersion string) (lastErr error) {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
_, exist := globalTaskMap.Load(task.Id)
if exist {
return fmt.Errorf("task %d already exists", task.Id)
}
// set task ve to global map
// when task is done or canceled, it will be removed from the map
globalTaskMap.Store(task.Id, t)
defer globalTaskMap.Delete(task.Id)
lastWords := ""
reporter := report.NewReporter(ctx, cancel, t.client, task)
defer func() {
// set the job to failed on an error return value
if lastErr != nil {
reporter.Fire(&log.Entry{
Data: log.Fields{
"jobResult": "failure",
},
Time: time.Now(),
})
}
_ = reporter.Close(lastWords)
}()
reporter.RunDaemon()
reporter.Logf("%s(version:%s) received task %v of job %v, be triggered by event: %s", runnerName, runnerVersion, task.Id, task.Context.Fields["job"].GetStringValue(), task.Context.Fields["event_name"].GetStringValue())
workflowsPath, err := getWorkflowsPath(t.Input.repoDirectory)
if err != nil {
lastWords = err.Error()
return err
}
t.log.Debugf("workflows path: %s", workflowsPath)
workflow, err := model.ReadWorkflow(bytes.NewReader(task.WorkflowPayload))
if err != nil {
lastWords = err.Error()
return err
}
jobIDs := workflow.GetJobIDs()
if len(jobIDs) != 1 {
err := fmt.Errorf("multiple jobs found: %v", jobIDs)
lastWords = err.Error()
return err
}
jobID := jobIDs[0]
plan, err := model.CombineWorkflowPlanner(workflow).PlanJob(jobID)
if err != nil {
lastWords = err.Error()
return err
}
job := workflow.GetJob(jobID)
reporter.ResetSteps(len(job.Steps))
log.Infof("plan: %+v", plan.Stages[0].Runs)
token := getToken(task)
dataContext := task.Context.Fields
log.Infof("task %v repo is %v %v %v", task.Id, dataContext["repository"].GetStringValue(),
dataContext["gitea_default_actions_url"].GetStringValue(),
t.client.Address())
preset := &model.GithubContext{
Event: dataContext["event"].GetStructValue().AsMap(),
RunID: dataContext["run_id"].GetStringValue(),
RunNumber: dataContext["run_number"].GetStringValue(),
Actor: dataContext["actor"].GetStringValue(),
Repository: dataContext["repository"].GetStringValue(),
EventName: dataContext["event_name"].GetStringValue(),
Sha: dataContext["sha"].GetStringValue(),
Ref: dataContext["ref"].GetStringValue(),
RefName: dataContext["ref_name"].GetStringValue(),
RefType: dataContext["ref_type"].GetStringValue(),
HeadRef: dataContext["head_ref"].GetStringValue(),
BaseRef: dataContext["base_ref"].GetStringValue(),
Token: token,
RepositoryOwner: dataContext["repository_owner"].GetStringValue(),
RetentionDays: dataContext["retention_days"].GetStringValue(),
}
eventJSON, err := json.Marshal(preset.Event)
if err != nil {
lastWords = err.Error()
return err
}
maxLifetime := 3 * time.Hour
if deadline, ok := ctx.Deadline(); ok {
maxLifetime = time.Until(deadline)
}
input := t.Input
config := &runner.Config{
// On Linux, Workdir will be like "/<owner>/<repo>"
// On Windows, Workdir will be like "\<owner>\<repo>"
Workdir: filepath.FromSlash(string(filepath.Separator) + preset.Repository),
BindWorkdir: false,
ReuseContainers: false,
ForcePull: input.forcePull,
ForceRebuild: input.forceRebuild,
LogOutput: true,
JSONLogger: input.jsonLogger,
Env: input.envs,
Secrets: task.Secrets,
InsecureSecrets: input.insecureSecrets,
Privileged: input.privileged,
UsernsMode: input.usernsMode,
ContainerArchitecture: input.containerArchitecture,
ContainerDaemonSocket: input.containerDaemonSocket,
UseGitIgnore: input.useGitIgnore,
GitHubInstance: t.client.Address(),
ContainerCapAdd: input.containerCapAdd,
ContainerCapDrop: input.containerCapDrop,
AutoRemove: true,
ArtifactServerPath: input.artifactServerPath,
ArtifactServerPort: input.artifactServerPort,
NoSkipCheckout: true,
PresetGitHubContext: preset,
EventJSON: string(eventJSON),
ContainerNamePrefix: fmt.Sprintf("GITEA-ACTIONS-TASK-%d", task.Id),
ContainerMaxLifetime: maxLifetime,
ContainerNetworkMode: input.containerNetworkMode,
DefaultActionInstance: dataContext["gitea_default_actions_url"].GetStringValue(),
PlatformPicker: t.platformPicker,
}
r, err := runner.New(config)
if err != nil {
lastWords = err.Error()
return err
}
executor := r.NewPlanExecutor(plan)
t.log.Infof("workflow prepared")
reporter.Logf("workflow prepared")
// add logger recorders
ctx = common.WithLoggerHook(ctx, reporter)
if err := executor(ctx); err != nil {
lastWords = err.Error()
return err
}
return nil
}