From e2715377601daa90b1f9467d530fc4351f114d06 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Sun, 20 Sep 2026 20:24:45 +0000 Subject: [PATCH] feat: assemble Go scheduler and backend workers --- README.md | 32 ++- cmd/gitea-dynamic-runner/controller.go | 240 ++++++++++++++++++++ cmd/gitea-dynamic-runner/controller_test.go | 59 +++++ cmd/gitea-dynamic-runner/main.go | 14 +- container/controller.Dockerfile | 20 +- docs/runner-protocol-roadmap.md | 5 + internal/taskworker/worker.go | 16 +- internal/taskworker/worker_test.go | 4 +- 8 files changed, 355 insertions(+), 35 deletions(-) create mode 100644 cmd/gitea-dynamic-runner/controller.go create mode 100644 cmd/gitea-dynamic-runner/controller_test.go diff --git a/README.md b/README.md index 2fe01cc..41eaff3 100644 --- a/README.md +++ b/README.md @@ -15,11 +15,14 @@ runs-on: [self-hosted, vm] 只执行一个 job，并在 job 结束后连同本地状态一起销毁。完整的设计约束见 [`docs/design-principles.md`](docs/design-principles.md)。 -组件： +目标 Go controller 组件： -- `controller`：接收 Gitea `workflow_job` webhook，将任务持久化到 NATS JetStream；仅在 - 显式启用 VM consumer 时调用 OpenSandbox Lifecycle - API，从 `ci-pod` 或 `ci-vm` Pool 创建一次性环境。 +- `scheduler`：以常驻 Gitea RunnerService 身份直接领取 task，并把完整 assignment + 持久化到 JetStream；同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。 +- `pod-worker`：直接在 homelab Kubernetes 创建一次性 Pod。 +- `vm-worker`：通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。 +- 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker，因为 facade + 的 assignment claim registry 仍是进程内状态；支持安全拆分前进程会明确拒绝该配置。 - `microvm-runner-launch`：为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP，运行 Cloud Hypervisor，退出后完整清理。 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral @@ -27,7 +30,7 @@ runs-on: [self-hosted, vm] - `opensandbox-identity`：在 sandbox 集群按实际 Pod UID 创建并清理临时 SPIFFE entry；不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见 [`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。 -- `pod-worker`：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用 +- Pod executor：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用 host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在 支持原生 job hooks 的 3.x 版本，在 workflow 第一步前等待实际任务对应的 SVID。 - `jwt-broker`：早期共享 Kubernetes runner 的过渡实验；目标架构不部署它，每个 @@ -45,8 +48,27 @@ python -m venv .venv . .venv/bin/activate pip install -e '.[test]' pytest + +go test ./... +go vet ./... ``` +## Go controller 首次集成配置 + +controller 默认执行 `controller` 子命令，runner 镜像执行 `executor` 子命令。所有长期 +credential 都从挂载文件读取，不接受明文环境变量： + +- `GITEA_RUNNER_UUID_FILE`、`GITEA_RUNNER_TOKEN_FILE`：scheduler 的常驻 RunnerService + registration；该 credential 不下发给 executor。 +- `NATS_PASSWORD_FILE`：assignment stream 的连接密码。 +- `RUNNER_FACADE_CAPABILITY_KEY_FILE`：至少 32 字节的 controller HMAC key。 +- `OPENSANDBOX_API_KEY_FILE`：仅启用 `vm-worker` 时读取。 + +必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`（应使用 digest）、 +`RUNNER_FACADE_URL`、`RUNNER_FACADE_SPIFFE_ID` 和 `SPIFFE_ENDPOINT_SOCKET`。默认 +`COMPONENTS=all`、Pod 并发 4、VM 并发 1；首次 smoke test 应显式设为 +`COMPONENTS=scheduler,pod-worker`，先验证 Pod 链路，避免同时消耗 VM 容量。 + ## 安全边界 - OpenSandbox API key、webhook secret 和 Gitea registration token 只从文件读取。 diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go new file mode 100644 index 0000000..0607b69 --- /dev/null +++ b/cmd/gitea-dynamic-runner/controller.go @@ -0,0 +1,240 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "slices" + "strconv" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "golang.org/x/sync/errgroup" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/giteaactions" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/opensandboxbackend" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/podbackend" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerbootstrap" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskscheduler" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker" +) + +type runComponent func(context.Context) error + +func (function runComponent) Run(ctx context.Context) error { return function(ctx) } + +type controllerConfig struct { + Components controller.Selection + TrustDomain, WorkloadAPIAddr string + GiteaURL, GiteaUUID, GiteaToken string + NATSURL, NATSUser, NATSPassword, NATSCA, Stream, SubjectBase string + FacadeListen, FacadeURL, FacadeSPIFFEID string + CapabilityKey []byte + PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string + PodExecutorUID, PodCapacity int + OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool string + VMTimeout, VMCapacity int +} + +func runController(ctx context.Context) error { + config, err := loadControllerConfig() + if err != nil { + return err + } + if !slices.Contains(config.Components, controller.Scheduler) { + return errors.New("split worker deployment is not yet safe: scheduler/facade must be enabled with workers") + } + if !slices.Contains(config.Components, controller.PodWorker) && !slices.Contains(config.Components, controller.VMWorker) { + return errors.New("scheduler requires at least one local backend worker") + } + + natsOptions := []nats.Option{nats.Name("gitea-dynamic-runner")} + if config.NATSUser != "" || config.NATSPassword != "" { + natsOptions = append(natsOptions, nats.UserInfo(config.NATSUser, config.NATSPassword)) + } + if config.NATSCA != "" { + natsOptions = append(natsOptions, nats.RootCAs(config.NATSCA)) + } + natsConnection, err := nats.Connect(config.NATSURL, natsOptions...) + if err != nil { + return fmt.Errorf("connect NATS: %w", err) + } + defer natsConnection.Close() + js, err := jetstream.New(natsConnection) + if err != nil { + return fmt.Errorf("open JetStream: %w", err) + } + if _, err := js.Stream(ctx, config.Stream); err != nil { + return fmt.Errorf("open assignment stream %s: %w", config.Stream, err) + } + + capabilities, err := runnerfacade.NewCapabilities(config.CapabilityKey) + if err != nil { + return err + } + registry := runnerfacade.NewRegistry() + giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken) + facade := &runnerfacade.Facade{Registry: registry, Capabilities: capabilities, Upstream: giteaClient} + bootstrap := runnerbootstrap.Bootstrap{ + Capabilities: capabilities, FacadeURL: config.FacadeURL, FacadeSPIFFEID: config.FacadeSPIFFEID, + } + + labels := []string{"self-hosted"} + if slices.Contains(config.Components, controller.PodWorker) { + labels = append(labels, string(taskassignment.BackendPod)) + } + if slices.Contains(config.Components, controller.VMWorker) { + labels = append(labels, string(taskassignment.BackendVM)) + } + poller := taskscheduler.Poller{ + Client: giteaClient, + Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{ + JetStream: js, SubjectBase: config.SubjectBase, + }}, + Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels}, + OnError: func(err error) { log.Printf("scheduler: %v", err) }, + } + facadeServer := runnerfacade.Server{ + Facade: facade, ListenAddress: config.FacadeListen, TrustDomain: config.TrustDomain, + WorkloadAPIAddr: config.WorkloadAPIAddr, + } + components := controller.Registry{ + controller.Scheduler: runComponent(func(ctx context.Context) error { + group, groupContext := errgroup.WithContext(ctx) + group.Go(func() error { return facadeServer.Run(groupContext) }) + group.Go(func() error { return poller.Run(groupContext) }) + return group.Wait() + }), + } + + if slices.Contains(config.Components, controller.PodWorker) { + client, err := podbackend.NewInClusterClient() + if err != nil { + return err + } + backend := podbackend.Backend{API: client, Config: podbackend.Config{ + Namespace: config.PodNamespace, Image: config.PodImage, ServiceAccount: config.PodServiceAccount, + ExecutorArgs: []string{"executor"}, TrustDomain: config.TrustDomain, + SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass, ExecutorUID: config.PodExecutorUID, + }} + component, err := workerComponent(ctx, js, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry) + if err != nil { + return err + } + components[controller.PodWorker] = component + } + if slices.Contains(config.Components, controller.VMWorker) { + lifecycle := opensandboxbackend.NewLifecycleClient(config.OpenSandboxURL, config.OpenSandboxAPIKey, &http.Client{Timeout: 60 * time.Second}) + backend := opensandboxbackend.Backend{Lifecycle: lifecycle, Config: opensandboxbackend.Config{ + Pool: config.OpenSandboxPool, Timeout: config.VMTimeout, + Entrypoint: []string{"/usr/local/bin/gitea-dynamic-runner", "executor"}, + Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr}, + }} + component, err := workerComponent(ctx, js, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry) + if err != nil { + return err + } + components[controller.VMWorker] = component + } + return controller.Run(ctx, config.Components, components) +} + +func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims) (controller.Component, error) { + consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity) + if err != nil { + return nil, err + } + return assignmentqueue.ConsumerComponent{ + Consumer: consumer, Capacity: capacity, + Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims}, + OnError: func(err error) { log.Printf("%s worker: %v", backend, err) }, + }, nil +} + +func loadControllerConfig() (controllerConfig, error) { + selection, err := controller.ParseSelection(os.Getenv("COMPONENTS")) + if err != nil { + return controllerConfig{}, err + } + read := func(name string) (string, error) { + path := os.Getenv(name) + if path == "" { + return "", fmt.Errorf("%s is required", name) + } + value, err := os.ReadFile(path) + if err != nil { + return "", fmt.Errorf("read %s: %w", name, err) + } + return strings.TrimSpace(string(value)), nil + } + uuid, err := read("GITEA_RUNNER_UUID_FILE") + if err != nil { + return controllerConfig{}, err + } + token, err := read("GITEA_RUNNER_TOKEN_FILE") + if err != nil { + return controllerConfig{}, err + } + natsPassword, err := read("NATS_PASSWORD_FILE") + if err != nil { + return controllerConfig{}, err + } + capabilityKey, err := read("RUNNER_FACADE_CAPABILITY_KEY_FILE") + if err != nil { + return controllerConfig{}, err + } + config := controllerConfig{ + Components: selection, TrustDomain: env("TRUST_DOMAIN", "ddupan.top"), WorkloadAPIAddr: os.Getenv("SPIFFE_ENDPOINT_SOCKET"), + GiteaURL: env("GITEA_INSTANCE_URL", "https://git.ddupan.top"), GiteaUUID: uuid, GiteaToken: token, + NATSURL: env("NATS_URL", "tls://nats.ad.ddupan.top:4222"), NATSUser: env("NATS_USER", "ci-worker"), NATSPassword: natsPassword, + NATSCA: os.Getenv("NATS_CA_FILE"), Stream: env("NATS_STREAM", "CI_RUNNER"), SubjectBase: env("NATS_SUBJECT_BASE", "ci.runner"), + FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey), + PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"), + SPIRECluster: env("SPIRE_CLUSTER", "homelab"), SPIREClass: env("SPIRE_CLASS", "spire-mgmt-spire"), PodExecutorUID: envInt("POD_EXECUTOR_UID", 2000), PodCapacity: envInt("POD_CAPACITY", 4), + OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1), + } + if config.WorkloadAPIAddr == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" { + return controllerConfig{}, errors.New("SPIFFE_ENDPOINT_SOCKET, RUNNER_FACADE_URL, and RUNNER_FACADE_SPIFFE_ID are required") + } + if slices.Contains(selection, controller.PodWorker) && config.PodImage == "" { + return controllerConfig{}, errors.New("POD_EXECUTOR_IMAGE is required for pod-worker") + } + if slices.Contains(selection, controller.VMWorker) { + if config.OpenSandboxURL == "" { + return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker") + } + config.OpenSandboxAPIKey, err = read("OPENSANDBOX_API_KEY_FILE") + if err != nil { + return controllerConfig{}, err + } + } + return config, nil +} + +func env(name, fallback string) string { + if value := strings.TrimSpace(os.Getenv(name)); value != "" { + return value + } + return fallback +} + +func envInt(name string, fallback int) int { + value := strings.TrimSpace(os.Getenv(name)) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil || parsed < 1 { + return fallback + } + return parsed +} diff --git a/cmd/gitea-dynamic-runner/controller_test.go b/cmd/gitea-dynamic-runner/controller_test.go new file mode 100644 index 0000000..0e14844 --- /dev/null +++ b/cmd/gitea-dynamic-runner/controller_test.go @@ -0,0 +1,59 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller" +) + +func secretFile(t *testing.T, name, value string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, []byte(value+"\n"), 0o600); err != nil { + t.Fatal(err) + } + return path +} + +func TestLoadControllerConfigUsesFileSecrets(t *testing.T) { + t.Setenv("COMPONENTS", "scheduler,pod-worker") + t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "scheduler-uuid")) + t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "scheduler-token")) + t.Setenv("NATS_PASSWORD_FILE", secretFile(t, "nats", "nats-password")) + t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef")) + t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock") + t.Setenv("RUNNER_FACADE_URL", "https://gitea-runner-facade.gitea-actions.svc:8443") + t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/ns/gitea-actions/sa/gitea-dynamic-runner") + t.Setenv("POD_EXECUTOR_IMAGE", "zot.ddupan.top/ci/gitea-runner@sha256:abc") + + config, err := loadControllerConfig() + if err != nil { + t.Fatal(err) + } + if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.PodWorker { + t.Fatalf("components = %#v", config.Components) + } + if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSPassword != "nats-password" { + t.Fatal("file secrets were not loaded") + } + if string(config.CapabilityKey) != "0123456789abcdef0123456789abcdef" || config.PodExecutorUID != 2000 { + t.Fatalf("config = %#v", config) + } +} + +func TestLoadControllerConfigRequiresOpenSandboxSecretOnlyForVM(t *testing.T) { + t.Setenv("COMPONENTS", "scheduler,vm-worker") + t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "uuid")) + t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "token")) + t.Setenv("NATS_PASSWORD_FILE", secretFile(t, "nats", "password")) + t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef")) + t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock") + t.Setenv("RUNNER_FACADE_URL", "https://facade:8443") + t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/controller") + t.Setenv("OPENSANDBOX_API", "http://opensandbox.internal") + if _, err := loadControllerConfig(); err == nil { + t.Fatal("expected missing OpenSandbox API key file error") + } +} diff --git a/cmd/gitea-dynamic-runner/main.go b/cmd/gitea-dynamic-runner/main.go index 71b99a6..5870bee 100644 --- a/cmd/gitea-dynamic-runner/main.go +++ b/cmd/gitea-dynamic-runner/main.go @@ -19,12 +19,18 @@ func main() { } func run() error { - if len(os.Args) != 2 { - return errors.New("usage: gitea-dynamic-runner executor") + if len(os.Args) > 2 { + return errors.New("usage: gitea-dynamic-runner [controller|executor]") } ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) defer stop() - switch os.Args[1] { + command := "controller" + if len(os.Args) == 2 { + command = os.Args[1] + } + switch command { + case "controller": + return runController(ctx) case "executor": config, err := runnerbootstrap.ExecutorConfigFromEnvironment() if err != nil { @@ -32,6 +38,6 @@ func run() error { } return runnerbootstrap.RunExecutor(ctx, config) default: - return fmt.Errorf("unknown command %q", os.Args[1]) + return fmt.Errorf("unknown command %q", command) } } diff --git a/container/controller.Dockerfile b/container/controller.Dockerfile index f2e746d..05815c0 100644 --- a/container/controller.Dockerfile +++ b/container/controller.Dockerfile @@ -1,17 +1,17 @@ -FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire - -FROM python:3.12.11-alpine3.22 AS build +FROM docker.io/library/golang:1.27-alpine@sha256:4cb7ac979db5fcc41cae44b2227ba5ab8a51e8807f40d9ba4dee20a0ad960b5b AS build WORKDIR /src -COPY pyproject.toml README.md ./ -COPY src ./src -RUN python -m venv /venv && /venv/bin/pip install --no-cache-dir . +COPY go.mod go.sum ./ +RUN go mod download +COPY cmd ./cmd +COPY internal ./internal +RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/gitea-dynamic-runner ./cmd/gitea-dynamic-runner FROM python:3.12.11-alpine3.22 RUN addgroup -S -g 65532 runner \ && adduser -S -D -H -u 65532 -G runner runner \ && install -d -o 65532 -g 65532 /var/run/secrets/kubernetes.io/serviceaccount -COPY --from=build /venv /venv -COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent +COPY --from=build /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner USER 65532:65532 -EXPOSE 8787 -ENTRYPOINT ["/venv/bin/gitea-dynamic-runner-controller"] +EXPOSE 8443 +ENTRYPOINT ["/usr/local/bin/gitea-dynamic-runner"] +CMD ["controller"] diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index f748ddf..8dea173 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -22,6 +22,11 @@ controller 使用单一 Go 二进制；默认在同一进程启用 `scheduler` `vm-worker`，也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界， 共享进程不意味着共享后端状态或把 assignment 降级为内存 channel。 +首轮集成的 facade pending/claimed registry 与三个组件同进程。虽然二进制保留组件选择 +接口，但当前会拒绝“worker 不带 scheduler/facade”的拆分部署：普通 Kubernetes Service +无法保证 executor 回到持有其 assignment 的 replica。后续拆分必须增加按 assignment +路由或可重建的 claim 分发，不能新增一套生命周期数据库来掩盖该问题。 + executor 直接运行固定版本的官方 Gitea Runner 二进制，不 fork workflow 执行引擎。 controller 暴露兼容 RunnerService 的 facade：`FetchTask` 只返回已分配 assignment， `UpdateTask` 与 `UpdateLog` 转发真实 Gitea。facade 同时验证逻辑 SPIFFE ID、assignment diff --git a/internal/taskworker/worker.go b/internal/taskworker/worker.go index 69de29b..e52a9be 100644 --- a/internal/taskworker/worker.go +++ b/internal/taskworker/worker.go @@ -77,28 +77,16 @@ type Worker struct { // returns true, all recovery information exists in Kubernetes/OpenSandbox and // the assignment message can be acknowledged immediately. func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment) (bool, error) { - if w.Backend == nil || w.Tasks == nil || w.Bootstrap == nil { - return false, errors.New("backend, Gitea task state, and runner bootstrap are required") + if w.Backend == nil || w.Bootstrap == nil { + return false, errors.New("backend and runner bootstrap are required") } if assignment.ID == "" || assignment.Task == nil { return false, errors.New("valid assignment is required") } - terminal, err := w.Tasks.Terminal(ctx, assignment.Task.GetId()) - if err != nil { - return false, err - } executor, err := w.Backend.Find(ctx, assignment.ID) if err != nil { return false, err } - if terminal { - if executor != nil { - if err := w.Backend.Delete(ctx, executor); err != nil { - return false, err - } - } - return true, nil - } if executor == nil { launch, launchErr := w.launchSpec(assignment) if launchErr != nil { diff --git a/internal/taskworker/worker_test.go b/internal/taskworker/worker_test.go index c03be14..3729bb6 100644 --- a/internal/taskworker/worker_test.go +++ b/internal/taskworker/worker_test.go @@ -79,7 +79,7 @@ func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) { func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) { backend := &fakeBackend{} - worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}} + worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}} accepted, err := worker.Accept(context.Background(), assignment()) if err != nil || !accepted { @@ -92,7 +92,7 @@ func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) { func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) { backend := &fakeBackend{executor: &Executor{Name: "pending", Phase: PhasePending}} - worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}} + worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}} accepted, err := worker.Accept(context.Background(), assignment()) if err != nil || accepted {