diff --git a/cmd/daemon.go b/cmd/daemon.go index 3c7f6e5..20f59c5 100644 --- a/cmd/daemon.go +++ b/cmd/daemon.go @@ -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, diff --git a/cmd/exec.go b/cmd/exec.go index 6edba53..0c2120d 100644 --- a/cmd/exec.go +++ b/cmd/exec.go @@ -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 } diff --git a/internal/pkg/artifactcache/handler.go b/internal/pkg/artifactcache/handler.go index 86e7aa5..2e063fc 100644 --- a/internal/pkg/artifactcache/handler.go +++ b/internal/pkg/artifactcache/handler.go @@ -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 == "" { diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 1f05f68..871b346 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -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 diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 34f181c..6280f47 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -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 diff --git a/internal/pkg/report/reporter.go b/internal/pkg/report/reporter.go index f8456ab..e6b635a 100644 --- a/internal/pkg/report/reporter.go +++ b/internal/pkg/report/reporter.go @@ -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, diff --git a/runtime/runtime.go b/runtime/runtime.go index 6db611f..c9378aa 100644 --- a/runtime/runtime.go +++ b/runtime/runtime.go @@ -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 "//" + // On Windows, Workdir will be like "\\" + 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) } diff --git a/runtime/task.go b/runtime/task.go index e05e477..ed21f2c 100644 --- a/runtime/task.go +++ b/runtime/task.go @@ -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 "//" - // On Windows, Workdir will be like "\\" - 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 -}