runner/internal/app/cmd/daemon.go

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

package cmd

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"net/http"
	"os"
	"path"
	"path/filepath"
	"runtime"
	"slices"
	"strconv"
	"strings"
	"time"

	"gitea.com/gitea/runner/act/common/logging"
	"gitea.com/gitea/runner/internal/app/poll"
	"gitea.com/gitea/runner/internal/app/run"
	"gitea.com/gitea/runner/internal/pkg/client"
	"gitea.com/gitea/runner/internal/pkg/config"
	"gitea.com/gitea/runner/internal/pkg/envcheck"
	"gitea.com/gitea/runner/internal/pkg/labels"
	"gitea.com/gitea/runner/internal/pkg/lock"
	"gitea.com/gitea/runner/internal/pkg/metrics"
	"gitea.com/gitea/runner/internal/pkg/report"
	"gitea.com/gitea/runner/internal/pkg/telemetry"
	"gitea.com/gitea/runner/internal/pkg/ver"

	"connectrpc.com/connect"
	runnerv1 "gitea.dev/actionslib/runner/v1"
	"github.com/avast/retry-go/v5"
	"github.com/mattn/go-isatty"
	"github.com/spf13/cobra"
)

func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) func(cmd *cobra.Command, args []string) error {
	return func(cmd *cobra.Command, args []string) error {
		cfg, err := config.LoadDefault(*configFile)
		if err != nil {
			return fmt.Errorf("invalid configuration: %w", err)
		}

		initLogging(cfg)
		logging.Info("Starting runner daemon")

		reg, err := config.LoadRegistration(cfg.Runner.File)
		if os.IsNotExist(err) {
			logging.Error("registration file not found, please register the runner first")
			return err
		} else if err != nil {
			return fmt.Errorf("failed to load registration file: %w", err)
		}

		// Guard against a second runner process sharing this runner file: two
		// processes with the same identity are indistinguishable to Gitea and
		// end up cancelling each other's jobs.
		releaseLock, err := lock.TryLock(cfg.Runner.File)
		switch {
		case errors.Is(err, lock.ErrLocked):
			logging.Errorf("another gitea-runner process is already using %q; each runner process needs its own runner file (runner.file)", cfg.Runner.File)
			return err
		case err != nil:
			// Best-effort guard: if the lock file can't be created (e.g. a
			// read-only runner-file mount), warn and start anyway rather than
			// refusing to run.
			logging.Warnf("could not lock runner file %q, continuing without the single-process guard: %v", cfg.Runner.File, err)
		default:
			// Held until shutdown finishes: the draining runner still owns this
			// identity on the server, so releasing early would let a restart
			// reintroduce the duplicate-identity job cancellations.
			defer func() { _ = releaseLock() }()
		}

		shutdownTelemetry, err := telemetry.Setup(ctx, reg.UUID, reg.Name, metrics.Registry)
		if err != nil {
			logging.WithError(err).Warn("OpenTelemetry export failed to start")
		}
		defer func() {
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()
			if err := shutdownTelemetry(ctx); err != nil {
				logging.WithError(err).Warn("OpenTelemetry export failed to flush")
			}
		}()

		lbls := resolveLabels(daemArgs.Labels, cfg.Runner.Labels, reg.Labels)

		ls := labels.Labels{}
		for _, l := range lbls {
			label, err := labels.Parse(l)
			if err != nil {
				logging.WithError(err).Warnf("ignored invalid label %q", l)
				continue
			}
			ls = append(ls, label)
		}
		if len(ls) == 0 {
			logging.Warn("no labels configured, runner may not be able to pick up jobs")
		}

		// Before the first Docker API call: the standard library resolves the proxy
		// environment once. Ungated because host labels still reach the daemon.
		if dockerSocketPath, err := getDockerSocketPath(cfg.Container.DockerHost); err == nil {
			run.BypassProxyForDockerHost(dockerSocketPath)
		} else {
			logging.Debugf("cannot resolve the docker socket path, so Docker API calls are not exempted from the proxy: %v", err)
		}

		if ls.RequireDocker() || cfg.Container.RequireDocker {
			// Wait for dockerd be ready
			if timeout := cfg.Container.DockerTimeout; timeout > 0 {
				tctx, cancel := context.WithTimeout(ctx, timeout)
				defer cancel()
				keepRunning := true
				for keepRunning {
					dockerSocketPath, err := getDockerSocketPath(cfg.Container.DockerHost)
					if err != nil {
						logging.Errorf("Failed to get socket path: %s", err.Error())
					} else if err = envcheck.CheckIfDockerRunning(tctx, dockerSocketPath); errors.Is(err, context.Canceled) {
						logging.Infof("Docker wait timeout of %s expired", timeout.String())
						break
					} else if err != nil {
						logging.Errorf("Docker connection failed: %s", err.Error())
					} else {
						logging.Infof("Docker is ready")
						break
					}
					select {
					case <-time.After(time.Second):
					case <-tctx.Done():
						logging.Infof("Docker wait timeout of %s expired", timeout.String())
						keepRunning = false
					}
				}
			}
			// Require dockerd be ready
			dockerSocketPath, err := getDockerSocketPath(cfg.Container.DockerHost)
			if err != nil {
				return err
			}
			if err := envcheck.CheckIfDockerRunning(ctx, dockerSocketPath); err != nil {
				return err
			}
			// if dockerSocketPath passes the check, override DOCKER_HOST with dockerSocketPath
			os.Setenv("DOCKER_HOST", dockerSocketPath)
			run.WarnIfDaemonHasNoProxy(ctx)
			// empty cfg.Container.DockerHost means runner need to find an available docker host automatically
			// and assign the path to cfg.Container.DockerHost
			if cfg.Container.DockerHost == "" {
				cfg.Container.DockerHost = dockerSocketPath
			}
			// check the scheme, if the scheme is not npipe or unix
			// set cfg.Container.DockerHost to "-" because it can't be mounted to the job container
			if protoIndex := strings.Index(cfg.Container.DockerHost, "://"); protoIndex != -1 {
				scheme := cfg.Container.DockerHost[:protoIndex]
				if !strings.EqualFold(scheme, "npipe") && !strings.EqualFold(scheme, "unix") {
					cfg.Container.DockerHost = "-"
				}
			}
		}

		if !slices.Equal(reg.Labels, ls.ToStrings()) {
			reg.Labels = ls.ToStrings()
			if err := config.SaveRegistration(cfg.Runner.File, reg); err != nil {
				return fmt.Errorf("failed to save runner config: %w", err)
			}
			logging.Infof("labels updated to: %v", reg.Labels)
		}

		cli := client.New(
			reg.Address,
			cfg.Runner.TLSConfig(),
			reg.UUID,
			reg.Token,
			config.RequestTimeout,
			cfg.Runner.ExtraHeaders,
		)

		runner := run.NewRunner(cfg, reg, cli)
		defer func() {
			if err := runner.Close(); err != nil {
				logging.Warnf("runner %s: cache server shutdown: %v", reg.Name, err)
			}
		}()

		// declare the labels of the runner before fetching tasks
		resp, err := retry.NewWithData[*connect.Response[runnerv1.DeclareResponse]](
			retry.Context(ctx),
			retry.UntilSucceeded(),
			retry.Delay(declareRetryDelay),
			retry.DelayType(retry.FixedDelay),
			retry.RetryIf(shouldRetryDeclare),
			retry.OnRetry(func(attempt uint, err error) {
				if attempt == 0 {
					logging.WithError(err).Warn("fail to invoke Declare, retrying until Gitea is reachable")
				} else {
					logging.WithError(err).Debug("fail to invoke Declare, retrying")
				}
			}),
		).Do(func() (*connect.Response[runnerv1.DeclareResponse], error) {
			return runner.Declare(ctx, ls.Names())
		})
		switch {
		case err != nil && ctx.Err() != nil && !client.IsRegistrationRejected(err):
			return nil
		case err != nil && connect.CodeOf(err) == connect.CodeUnimplemented:
			logging.Errorf("Your Gitea version is too old to support runner declare, please upgrade to v1.21 or later")
			return err
		case err != nil:
			logging.WithError(err).Error("fail to invoke Declare")
			if client.IsRegistrationRejected(err) {
				removeEphemeralRegistration(reg, cfg.Runner.File)
			}
			return err
		default:
			logging.Infof("runner: %s, with version: %s, with labels: %v, declare successfully",
				resp.Msg.Runner.Name, resp.Msg.Runner.Version, resp.Msg.Runner.Labels)
		}
		runner.SetCapabilitiesFromDeclare(resp)

		poller := poll.New(cfg, cli, runner)

		metrics.Init() // OTLP exports the registry even without the server
		metrics.RunnerInfo.WithLabelValues(ver.Version(), resp.Msg.Runner.Name).Set(1)
		metrics.RunnerCapacity.Set(float64(cfg.Runner.Capacity))
		metrics.RegisterUptimeFunc(time.Now())
		metrics.RegisterRunningJobsFunc(runner.RunningCount, cfg.Runner.Capacity)
		metrics.RegisterStateFunc(func() string {
			if runner.RunningCount() > 0 {
				return metrics.LabelStateBusy
			}
			if ready, _ := poller.Ready(cfg.Metrics.ReadinessGrace); !ready {
				return metrics.LabelStateUnavailable
			}
			return metrics.LabelStateIdle
		})
		if cfg.Metrics.Enabled {
			metrics.StartServer(ctx, cfg.Metrics.Addr, func() (bool, string) {
				return poller.Ready(cfg.Metrics.ReadinessGrace)
			})
		}

		if daemArgs.Once || reg.Ephemeral {
			done := make(chan struct{})
			go func() {
				defer close(done)
				poller.PollOnce(func() { removeEphemeralRegistration(reg, cfg.Runner.File) })
			}()

			// shutdown when we complete a job or cancel is requested
			select {
			case <-ctx.Done():
			case <-done:
			}
		} else {
			go poller.Poll()

			// Stop either on an external cancellation or when the poller shuts
			// itself down (e.g. after the runner has been unregistered).
			select {
			case <-ctx.Done():
			case <-poller.Done():
			}
		}

		logging.Infof("runner: %s shutdown initiated, waiting %s for running jobs to complete before shutting down", resp.Msg.Runner.Name, cfg.Runner.ShutdownTimeout)

		ctx, cancel := context.WithTimeout(context.Background(), cfg.Runner.ShutdownTimeout)
		defer cancel()

		err = poller.Shutdown(ctx)
		if err != nil {
			logging.Warnf("runner: %s cancelled in progress jobs during shutdown", resp.Msg.Runner.Name)
		}

		if poller.RegistrationRejected() {
			removeEphemeralRegistration(reg, cfg.Runner.File)
		}

		if poller.Unregistered() {
			return errors.New("runner is no longer registered with the server; please register it again")
		}

		return nil
	}
}

func removeEphemeralRegistration(reg *config.Registration, file string) {
	if !reg.Ephemeral {
		return
	}
	if err := os.Remove(file); err == nil {
		logging.Infof("removed the spent ephemeral registration %q", file)
	} else if !os.IsNotExist(err) {
		logging.WithError(err).Warn("failed to remove the spent ephemeral registration")
	}
}

type daemonArgs struct {
	Once   bool
	Labels string
}

// resolveLabels picks the labels to run with: --labels/GITEA_RUNNER_LABELS > config > .runner.
// The flag lets a registered runner change its labels without deleting the .runner file.
func resolveLabels(argLabels string, cfgLabels, regLabels []string) []string {
	if lbls := splitLabels(argLabels); len(lbls) > 0 {
		return lbls
	}
	if len(cfgLabels) > 0 {
		return cfgLabels
	}
	return regLabels
}

func splitLabels(s string) []string {
	var lbls []string
	for l := range strings.SplitSeq(s, ",") {
		if l = strings.TrimSpace(l); l != "" {
			lbls = append(lbls, l)
		}
	}
	return lbls
}

// initLogging sets up the process-wide logger.
func initLogging(cfg *config.Config) {
	callPrettyfier := func(f *runtime.Frame) (string, string) {
		// get function name
		s := strings.Split(f.Function, ".")
		funcname := "[" + s[len(s)-1] + "]"
		// get file name and line number
		_, filename := path.Split(f.File)
		filename = "[" + filename + ":" + strconv.Itoa(f.Line) + "]"
		return funcname, filename
	}

	isTerm := isatty.IsTerminal(os.Stdout.Fd())
	logging.SetTextFormat(logging.TextOptions{
		DisableColors:    !isTerm,
		FullTimestamp:    true,
		CallerPrettyfier: callPrettyfier,
		Filter:           report.MaskGlobal,
	})

	l := cfg.Log.Level
	if l == "" {
		logging.Infof("Log level not set, sticking to info")
		return
	}

	level, err := logging.ParseLevel(l)
	if err != nil {
		logging.WithError(err).
			Errorf("invalid log level: %q", l)
	}

	// debug level
	switch level {
	case logging.LevelDebug, logging.LevelTrace:
		logging.SetReportCaller(true) // Only in debug or trace because it takes a performance toll
		logging.Infof("Log level %s requested, setting up report caller for further debugging", level)
	}

	if logging.GetLevel() != level {
		logging.Infof("log level set to %v", level)
		logging.SetLevel(level)
	}
}

var declareRetryDelay = 30 * time.Second

func shouldRetryDeclare(err error) bool {
	if errors.As(err, new(*tls.CertificateVerificationError)) || errors.As(err, new(tls.AlertError)) || errors.Is(err, http.ErrSchemeMismatch) { // misconfigurations never heal
		return false
	}
	switch connect.CodeOf(err) {
	case connect.CodeUnauthenticated, connect.CodeUnimplemented, connect.CodePermissionDenied, connect.CodeInvalidArgument:
		return false
	default:
		return !client.IsRegistrationRejected(err)
	}
}

var commonSocketPaths = []string{
	"/var/run/docker.sock",
	"/run/podman/podman.sock",
	"$HOME/.colima/docker.sock",
	"$XDG_RUNTIME_DIR/docker.sock",
	"$XDG_RUNTIME_DIR/podman/podman.sock",
	`\\.\pipe\docker_engine`,
	"$HOME/.docker/run/docker.sock",
}

func getDockerSocketPath(configDockerHost string) (string, error) {
	// a `-` means don't mount the docker socket to job containers
	if configDockerHost != "" && configDockerHost != "-" {
		return configDockerHost, nil
	}

	socket, found := os.LookupEnv("DOCKER_HOST")
	if found {
		return socket, nil
	}

	for _, p := range commonSocketPaths {
		if _, err := os.Lstat(os.ExpandEnv(p)); err == nil {
			if strings.HasPrefix(p, `\\.\`) {
				return "npipe://" + filepath.ToSlash(os.ExpandEnv(p)), nil
			}
			return "unix://" + filepath.ToSlash(os.ExpandEnv(p)), nil
		}
	}

	return "", errors.New("daemon Docker Engine socket not found and docker_host config was invalid")
}