gitea-dynamic-runner/internal/runnerbootstrap/executor.go

package runnerbootstrap

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"time"

	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)

type ExecutorConfig struct {
	AssignmentID    string
	Capability      string
	Backend         taskassignment.Backend
	FacadeURL       string
	FacadeSPIFFEID  string
	WorkloadAPIAddr string
	RunnerBinary    string
	ListenAddress   string
	WorkDir         string
	Stdout          *os.File
	Stderr          *os.File
}

// RunExecutor runs the SPIFFE proxy and one unmodified official Runner process.
// The generated registration file exists only in the executor's temporary
// work directory and the runner exits after its preassigned task.
func RunExecutor(ctx context.Context, config ExecutorConfig) error {
	if config.RunnerBinary == "" {
		config.RunnerBinary = "gitea-runner"
	}
	if config.ListenAddress == "" {
		config.ListenAddress = "127.0.0.1:0"
	}
	listener, err := net.Listen("tcp", config.ListenAddress)
	if err != nil {
		return fmt.Errorf("listen for local runner proxy: %w", err)
	}
	defer listener.Close()
	address, ok := listener.Addr().(*net.TCPAddr)
	if !ok || !address.IP.IsLoopback() {
		return errors.New("runner proxy must listen on a loopback address")
	}

	proxy, err := NewProxy(ctx, config.FacadeURL, config.FacadeSPIFFEID, config.WorkloadAPIAddr)
	if err != nil {
		return err
	}
	defer proxy.Close()

	workDir := config.WorkDir
	removeWorkDir := false
	if workDir == "" {
		workDir, err = os.MkdirTemp("", "gitea-task-runner-")
		if err != nil {
			return fmt.Errorf("create runner work directory: %w", err)
		}
		removeWorkDir = true
	}
	if removeWorkDir {
		defer os.RemoveAll(workDir)
	}
	registration, err := RegistrationJSON(
		config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Backend,
	)
	if err != nil {
		return err
	}
	if err := os.WriteFile(filepath.Join(workDir, ".runner"), registration, 0o600); err != nil {
		return fmt.Errorf("write one-shot runner registration: %w", err)
	}

	server := &http.Server{Handler: proxy.Handler, ReadHeaderTimeout: 10 * time.Second}
	serverErrors := make(chan error, 1)
	go func() { serverErrors <- server.Serve(listener) }()
	readyContext, cancelReady := context.WithTimeout(ctx, 2*time.Minute)
	readyErr := waitForFacade(readyContext, "http://"+listener.Addr().String())
	cancelReady()
	if readyErr != nil {
		shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		shutdownErr := server.Shutdown(shutdownContext)
		cancel()
		serverErr := <-serverErrors
		if errors.Is(serverErr, http.ErrServerClosed) {
			serverErr = nil
		}
		return errors.Join(readyErr, shutdownErr, serverErr)
	}

	command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once")
	command.Dir = workDir
	command.Stdout = config.Stdout
	command.Stderr = config.Stderr
	runnerErr := command.Run()
	shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	shutdownErr := server.Shutdown(shutdownContext)
	cancel()
	serverErr := <-serverErrors
	if errors.Is(serverErr, http.ErrServerClosed) {
		serverErr = nil
	}
	return errors.Join(runnerErr, shutdownErr, serverErr)
}

func waitForFacade(ctx context.Context, endpoint string) error {
	client := &http.Client{Timeout: 2 * time.Second}
	ticker := time.NewTicker(250 * time.Millisecond)
	defer ticker.Stop()
	for {
		request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
		if err != nil {
			return fmt.Errorf("create facade readiness request: %w", err)
		}
		response, requestErr := client.Do(request)
		if requestErr == nil {
			_ = response.Body.Close()
			if response.StatusCode < http.StatusInternalServerError {
				return nil
			}
		}
		select {
		case <-ctx.Done():
			return fmt.Errorf("wait for runner facade: %w", ctx.Err())
		case <-ticker.C:
		}
	}
}

// ExecutorConfigFromEnvironment reads the non-secret image configuration and
// the assignment-scoped values injected by the backend. The Workload API
// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
func ExecutorConfigFromEnvironment() (ExecutorConfig, error) {
	backend := taskassignment.Backend(os.Getenv(EnvBackend))
	if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
		return ExecutorConfig{}, fmt.Errorf("invalid %s %q", EnvBackend, backend)
	}
	config := ExecutorConfig{
		AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability),
		Backend: backend, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID),
		RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), ListenAddress: "127.0.0.1:0",
		Stdout: os.Stdout, Stderr: os.Stderr,
	}
	if config.AssignmentID == "" || config.Capability == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
		return ExecutorConfig{}, errors.New("complete runner assignment and facade environment is required")
	}
	return config, nil
}