gitcaddy-runner/internal/app/run/runner.go

// Copyright 2022 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT

package run

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	runnerv1 "code.gitea.io/actions-proto-go/runner/v1"
	"connectrpc.com/connect"
	"github.com/docker/docker/api/types/container"
	"github.com/nektos/act/pkg/artifactcache"
	"github.com/nektos/act/pkg/common"
	"github.com/nektos/act/pkg/model"
	"github.com/nektos/act/pkg/runner"
	log "github.com/sirupsen/logrus"

	"git.marketally.com/gitcaddy/gitcaddy-runner/internal/pkg/client"
	"git.marketally.com/gitcaddy/gitcaddy-runner/internal/pkg/config"
	"git.marketally.com/gitcaddy/gitcaddy-runner/internal/pkg/labels"
	"git.marketally.com/gitcaddy/gitcaddy-runner/internal/pkg/report"
	"git.marketally.com/gitcaddy/gitcaddy-runner/internal/pkg/ver"
)

// Runner runs the pipeline.
type Runner struct {
	name string

	cfg *config.Config

	client client.Client
	labels labels.Labels
	envs   map[string]string

	runningTasks sync.Map
}

// getJobCacheDir returns a job-isolated cache directory
func (r *Runner) getJobCacheDir(taskID int64) string {
	return filepath.Join(r.cfg.Host.WorkdirParent, "jobs", fmt.Sprintf("%d", taskID))
}

// cleanupJobCache removes the job-specific cache directory after completion
func (r *Runner) cleanupJobCache(taskID int64) {
	jobCacheDir := r.getJobCacheDir(taskID)
	if err := os.RemoveAll(jobCacheDir); err != nil {
		log.Warnf("failed to cleanup job cache %s: %v", jobCacheDir, err)
	} else {
		log.Infof("cleaned up job cache: %s", jobCacheDir)
	}
}

// CleanStaleJobCaches removes job cache directories older than maxAge
func (r *Runner) CleanStaleJobCaches(maxAge time.Duration) {
	jobsDir := filepath.Join(r.cfg.Host.WorkdirParent, "jobs")
	entries, err := os.ReadDir(jobsDir)
	if err != nil {
		return // directory may not exist yet
	}

	cutoff := time.Now().Add(-maxAge)
	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		info, err := entry.Info()
		if err != nil {
			continue
		}
		if info.ModTime().Before(cutoff) {
			jobPath := filepath.Join(jobsDir, entry.Name())
			if err := os.RemoveAll(jobPath); err != nil {
				log.Warnf("failed to remove stale job cache %s: %v", jobPath, err)
			} else {
				log.Infof("evicted stale job cache: %s", jobPath)
			}
		}
	}
}

// NewRunner creates a new Runner with the given configuration, registration, and client.
func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client) *Runner {
	ls := labels.Labels{}
	for _, v := range reg.Labels {
		if l, err := labels.Parse(v); err == nil {
			ls = append(ls, l)
		}
	}
	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 {
		if cfg.Cache.ExternalServer != "" {
			envs["ACTIONS_CACHE_URL"] = cfg.Cache.ExternalServer
		} else {
			cacheHandler, err := artifactcache.StartHandler(
				cfg.Cache.Dir,
				cfg.Cache.Host,
				cfg.Cache.Port,
				log.StandardLogger().WithField("module", "cache_request"),
			)
			if err != nil {
				log.Errorf("cannot init cache server, it will be disabled: %v", err)
				// go on
			} else {
				envs["ACTIONS_CACHE_URL"] = cacheHandler.ExternalURL() + "/"
			}
		}
	}

	// set artifact gitea api
	artifactGiteaAPI := strings.TrimSuffix(cli.Address(), "/") + "/api/actions_pipeline/"
	envs["ACTIONS_RUNTIME_URL"] = artifactGiteaAPI
	envs["ACTIONS_RESULTS_URL"] = strings.TrimSuffix(cli.Address(), "/")

	// Set specific environments to distinguish between Gitea and GitHub
	envs["GITEA_ACTIONS"] = "true"
	envs["GITEA_ACTIONS_RUNNER_VERSION"] = ver.Version()

	return &Runner{
		name:   reg.Name,
		cfg:    cfg,
		client: cli,
		labels: ls,
		envs:   envs,
	}
}

// Run executes a task from the server.
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)
	}
	r.runningTasks.Store(task.Id, struct{}{})
	defer r.runningTasks.Delete(task.Id)
	defer r.cleanupJobCache(task.Id)

	ctx, cancel := context.WithTimeout(ctx, r.cfg.Runner.Timeout)
	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
}

// getDefaultActionsURL
// when DEFAULT_ACTIONS_URL == "https://github.com" and GithubMirror is not blank,
// it should be set to GithubMirror first.
func (r *Runner) getDefaultActionsURL(_ context.Context, task *runnerv1.Task) string {
	giteaDefaultActionsURL := task.Context.Fields["gitea_default_actions_url"].GetStringValue()
	if giteaDefaultActionsURL == "https://github.com" && r.cfg.Runner.GithubMirror != "" {
		return r.cfg.Runner.GithubMirror
	}
	return giteaDefaultActionsURL
}

func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.Reporter) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic: %v", r)
		}
	}()

	reporter.Logf("%s(version:%s) received task %v of job %v, be triggered by event: %s", r.name, ver.Version(), task.Id, task.Context.Fields["job"].GetStringValue(), task.Context.Fields["event_name"].GetStringValue())

	workflow, jobID, err := generateWorkflow(task)
	if err != nil {
		return err
	}

	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(),
		r.getDefaultActionsURL(ctx, task),
		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
	}

	if actionsIDTokenRequestURL := taskContext["actions_id_token_request_url"].GetStringValue(); actionsIDTokenRequestURL != "" {
		r.envs["ACTIONS_ID_TOKEN_REQUEST_URL"] = actionsIDTokenRequestURL
		r.envs["ACTIONS_ID_TOKEN_REQUEST_TOKEN"] = taskContext["actions_id_token_request_token"].GetStringValue()
		task.Secrets["ACTIONS_ID_TOKEN_REQUEST_TOKEN"] = r.envs["ACTIONS_ID_TOKEN_REQUEST_TOKEN"]
	}

	giteaRuntimeToken := taskContext["gitea_runtime_token"].GetStringValue()
	if giteaRuntimeToken == "" {
		// use task token to action api token for previous Gitea Server Versions
		giteaRuntimeToken = preset.Token
	}
	r.envs["ACTIONS_RUNTIME_TOKEN"] = giteaRuntimeToken

	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)
	}

	// Create job-specific environment with isolated cache directories
	jobCacheDir := r.getJobCacheDir(task.Id)
	jobEnvs := make(map[string]string, len(r.envs)+2)
	for k, v := range r.envs {
		jobEnvs[k] = v
	}
	// Isolate golangci-lint cache to prevent parallel job conflicts
	jobEnvs["GOLANGCI_LINT_CACHE"] = filepath.Join(jobCacheDir, "golangci-lint")
	// Set XDG_CACHE_HOME to isolate other tools that respect it
	jobEnvs["XDG_CACHE_HOME"] = jobCacheDir

	runnerConfig := &runner.Config{
		// On Linux, Workdir will be like "/<parent_directory>/<owner>/<repo>"
		// On Windows, Workdir will be like "\<parent_directory>\<owner>\<repo>"
		Workdir:        filepath.FromSlash(fmt.Sprintf("/%s/%s", strings.TrimLeft(r.cfg.Container.WorkdirParent, "/"), preset.Repository)),
		BindWorkdir:    false,
		ActionCacheDir: filepath.FromSlash(jobCacheDir),

		ReuseContainers:       false,
		ForcePull:             r.cfg.Container.ForcePull,
		ForceRebuild:          r.cfg.Container.ForceRebuild,
		LogOutput:             true,
		JSONLogger:            false,
		Env:                   jobEnvs,
		Secrets:               task.Secrets,
		GitHubInstance:        strings.TrimSuffix(r.client.Address(), "/"),
		AutoRemove:            true,
		NoSkipCheckout:        true,
		PresetGitHubContext:   preset,
		EventJSON:             string(eventJSON),
		ContainerNamePrefix:   fmt.Sprintf("GITEA-ACTIONS-TASK-%d", task.Id),
		ContainerMaxLifetime:  maxLifetime,
		ContainerNetworkMode:  container.NetworkMode(r.cfg.Container.Network),
		ContainerOptions:      r.cfg.Container.Options,
		ContainerDaemonSocket: r.cfg.Container.DockerHost,
		Privileged:            r.cfg.Container.Privileged,
		DefaultActionInstance: r.getDefaultActionsURL(ctx, task),
		PlatformPicker:        r.labels.PickPlatform,
		Vars:                  task.Vars,
		ValidVolumes:          r.cfg.Container.ValidVolumes,
		InsecureSkipTLS:       r.cfg.Runner.Insecure,
	}

	rr, err := runner.New(runnerConfig)
	if err != nil {
		return err
	}
	executor := rr.NewPlanExecutor(plan)

	reporter.Logf("workflow prepared")

	// add logger recorders
	ctx = common.WithLoggerHook(ctx, reporter)

	if !log.IsLevelEnabled(log.DebugLevel) {
		ctx = runner.WithJobLoggerFactory(ctx, NullLogger{})
	}

	execErr := executor(ctx)
	reporter.SetOutputs(job.Outputs)
	return execErr
}

// Declare sends the runner's labels and capabilities to the server.
func (r *Runner) Declare(ctx context.Context, declareLabels []string, capabilitiesJSON string) (*connect.Response[runnerv1.DeclareResponse], error) {
	return r.client.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{
		Version:          ver.Version(),
		Labels:           declareLabels,
		CapabilitiesJson: capabilitiesJSON,
	}))
}

// MergeServerLabels merges labels returned from the server (which may include admin-added labels)
// with the runner's existing labels. This allows admins to add labels via the Gitea UI.
func (r *Runner) MergeServerLabels(serverLabels []string) {
	existing := make(map[string]bool)
	for _, l := range r.labels {
		existing[l.Name] = true
	}

	for _, labelStr := range serverLabels {
		label, err := labels.Parse(labelStr)
		if err != nil {
			log.Warnf("ignoring invalid server label %q: %v", labelStr, err)
			continue
		}
		if !existing[label.Name] {
			r.labels = append(r.labels, label)
			log.Infof("merged server label: %s", labelStr)
		}
	}
}