diff --git a/README.md b/README.md index 2b86479..6e60d5d 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,19 @@ # Gitea dynamic runner -为 Gitea Actions 按需创建一次性执行环境。对 workflow 提供两种稳定的 runner -接口: +为 Gitea Actions 按需创建一次性执行环境。workflow 分别声明 workload class 与 +placement driver: ```yaml -runs-on: [self-hosted, pod] +runs-on: [self-hosted, container, kubernetes] ``` ```yaml -runs-on: [self-hosted, vm] +runs-on: [self-hosted, vm, opensandbox] ``` -`pod` 使用动态 Kubernetes Pod,`vm` 使用动态 Cloud Hypervisor microVM。每个环境 +兼容标签 `[self-hosted, pod]` 严格映射为 `container+kubernetes`, +`[self-hosted, vm]` 和 `vm-dev` 严格映射为 `vm+opensandbox`;显式 driver 容量耗尽时 +不会回退到其他 driver。每个环境 只执行一个 job,并在 job 结束后连同本地状态一起销毁。完整的设计约束见 [`docs/design-principles.md`](docs/design-principles.md)。 @@ -24,8 +26,8 @@ runs-on: [self-hosted, vm] - `scheduler`:以常驻 Gitea RunnerService 身份直接领取 task,并把完整 assignment 持久化到 JetStream;一个 registration 下按配置启动多个并发 `FetchTask` goroutine, 同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。 -- `pod-worker`:直接在 homelab Kubernetes 创建一次性 Pod。 -- `vm-worker`:通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。 +- `kubernetes-worker`:直接在 homelab Kubernetes 创建一次性 container workload。 +- `opensandbox-worker`:通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 VM workload。 - 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker,因为 facade 的 assignment claim registry 仍是进程内状态;支持安全拆分前进程会明确拒绝该配置。 - `microvm-runner-launch`:为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP,运行 @@ -42,8 +44,8 @@ runs-on: [self-hosted, vm] - `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个 动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。 -Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于 -VM/Kata workload;两个 backend 使用独立 durable consumer 和独立容量池。assignment 根据 +Container/Kubernetes 路径由 homelab 集群中的 `kubernetes-worker` 直接创建 Pod。 +OpenSandbox 只用于 VM/Kata workload;每个 placement 使用独立 durable consumer 和容量池。assignment 根据 `runs-on` 进入对应池,池满时留在 JetStream pending,不会创建超出容量的 workload;任一 执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。 @@ -70,12 +72,13 @@ credential 都从挂载文件读取,不接受明文环境变量: - `NATS_PRODUCER_PASSWORD_FILE`、`NATS_WORKER_PASSWORD_FILE`:分别使用现有最小权限的 `ci-producer` publish 连接和 `ci-worker` pull/ACK 连接,controller 不合并权限。 - `RUNNER_FACADE_CAPABILITY_KEY_FILE`:至少 32 字节的 controller HMAC key。 -- `OPENSANDBOX_API_KEY_FILE`:仅启用 `vm-worker` 时读取。 +- `OPENSANDBOX_API_KEY_FILE`:仅启用 `opensandbox-worker` 时读取。 必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`(应使用 digest)、`SPIRE_AGENT_ID`、 `RUNNER_FACADE_URL`、`RUNNER_FACADE_SPIFFE_ID` 和 `SPIFFE_ENDPOINT_SOCKET`。默认 `COMPONENTS=all`、Pod 并发 4、VM 并发 1;首次 smoke test 应显式设为 -`COMPONENTS=scheduler,pod-worker`,先验证 Pod 链路,避免同时消耗 VM 容量。 +`COMPONENTS=scheduler,kubernetes-worker`,先验证 Kubernetes 链路,避免同时消耗 VM +容量。旧组件名 `pod-worker`、`vm-worker` 只作为配置兼容别名保留。 Pod task 的 terminal update 被 Gitea 接受后,controller 会在 Pod 上持久写入 `ci.ddupan.top/terminal=true` label。生命周期 reconciler 只清理同时带该 label 且已经 diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go index 0d753f4..334b998 100644 --- a/cmd/gitea-dynamic-runner/controller.go +++ b/cmd/gitea-dynamic-runner/controller.go @@ -38,6 +38,15 @@ type runComponent func(context.Context) error func (function runComponent) Run(ctx context.Context) error { return function(ctx) } +type terminalBackend interface { + MarkTerminal(context.Context, string) error +} + +type placementRuntime struct { + backend terminalBackend + pool *backendpool.Pool +} + type controllerConfig struct { Components controller.Selection TrustDomain, WorkloadAPIAddr string @@ -61,7 +70,7 @@ func runController(ctx context.Context) error { 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) { + if !slices.Contains(config.Components, controller.KubernetesWorker) && !slices.Contains(config.Components, controller.OpenSandboxWorker) { return errors.New("scheduler requires at least one local backend worker") } @@ -92,32 +101,19 @@ func runController(ctx context.Context) error { return err } registry := runnerfacade.NewRegistry() - podPool := backendpool.New(config.PodCapacity) - vmPool := backendpool.New(config.VMCapacity) - var podExecutorBackend *podbackend.Backend - var vmExecutorBackend *opensandboxbackend.Backend + runtimes := make(map[taskassignment.Placement]placementRuntime) giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken) facade := &runnerfacade.Facade{ Registry: registry, Capabilities: capabilities, Upstream: giteaClient, OnTerminal: func(ctx context.Context, assignment taskassignment.Assignment) error { - switch assignment.Backend { - case taskassignment.BackendPod: - if podExecutorBackend == nil { - return errors.New("Pod lifecycle backend is not configured") - } - if err := podExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil { - return err - } - podPool.Release(assignment.ID) - case taskassignment.BackendVM: - if vmExecutorBackend == nil { - return errors.New("VM lifecycle backend is not configured") - } - if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil { - return err - } - vmPool.Release(assignment.ID) + runtime, ok := runtimes[assignment.Placement] + if !ok { + return fmt.Errorf("placement runtime %s is not configured", assignment.Placement.Key()) } + if err := runtime.backend.MarkTerminal(ctx, assignment.ID); err != nil { + return err + } + runtime.pool.Release(assignment.ID) return nil }, } @@ -127,11 +123,11 @@ func runController(ctx context.Context) error { } labels := []string{"self-hosted"} - if slices.Contains(config.Components, controller.PodWorker) { - labels = append(labels, string(taskassignment.BackendPod)) + if slices.Contains(config.Components, controller.KubernetesWorker) { + labels = append(labels, "pod", string(taskassignment.WorkloadContainer), string(taskassignment.DriverKubernetes)) } - if slices.Contains(config.Components, controller.VMWorker) { - labels = append(labels, config.VMRunnerLabel) + if slices.Contains(config.Components, controller.OpenSandboxWorker) { + labels = append(labels, config.VMRunnerLabel, string(taskassignment.DriverOpenSandbox)) } poller := taskscheduler.Poller{ Client: giteaClient, @@ -168,7 +164,9 @@ func runController(ctx context.Context) error { }), } - if slices.Contains(config.Components, controller.PodWorker) { + if slices.Contains(config.Components, controller.KubernetesWorker) { + placement := taskassignment.KubernetesContainer + pool := backendpool.New(config.PodCapacity) client, err := podbackend.NewInClusterClient() if err != nil { return err @@ -179,57 +177,59 @@ func runController(ctx context.Context) error { SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass, SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID, }} - podExecutorBackend = &backend + runtimes[placement] = placementRuntime{backend: backend, pool: pool} assignments, err := backend.RecoverAssignments(ctx) if err != nil { return err } for _, assignment := range assignments { - podPool.Restore(assignment.ID) + pool.Restore(assignment.ID) if err := registry.RecoverClaimed(assignment); err != nil { return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err) } } - component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendPod)}, registry, podPool) + component, err := workerComponent(ctx, workerJS, config, placement, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(placement)}, registry, pool) if err != nil { return err } lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { - slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendPod, "error", err) + slog.Error("backend lifecycle error", "component", "lifecycle", "placement", placement.Key(), "error", err) }} - components[controller.PodWorker] = runComponent(func(ctx context.Context) error { + components[controller.KubernetesWorker] = runComponent(func(ctx context.Context) error { group, groupContext := errgroup.WithContext(ctx) group.Go(func() error { return component.Run(groupContext) }) group.Go(func() error { return lifecycle.Run(groupContext) }) return group.Wait() }) } - if slices.Contains(config.Components, controller.VMWorker) { + if slices.Contains(config.Components, controller.OpenSandboxWorker) { + placement := taskassignment.OpenSandboxVM + pool := backendpool.New(config.VMCapacity) 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}, }} - vmExecutorBackend = &backend + runtimes[placement] = placementRuntime{backend: backend, pool: pool} assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain) if err != nil { return err } for _, assignment := range assignments { - vmPool.Restore(assignment.ID) + pool.Restore(assignment.ID) if err := registry.RecoverClaimed(assignment); err != nil { return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err) } } - component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendVM)}, registry, vmPool) + component, err := workerComponent(ctx, workerJS, config, placement, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(placement)}, registry, pool) if err != nil { return err } lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { - slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendVM, "error", err) + slog.Error("backend lifecycle error", "component", "lifecycle", "placement", placement.Key(), "error", err) }} - components[controller.VMWorker] = runComponent(func(ctx context.Context) error { + components[controller.OpenSandboxWorker] = runComponent(func(ctx context.Context) error { group, groupContext := errgroup.WithContext(ctx) group.Go(func() error { return component.Run(groupContext) }) group.Go(func() error { return lifecycleReconciler.Run(groupContext) }) @@ -291,25 +291,25 @@ func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn, return nats.Connect(server, options...) } -func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) { - consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity) +func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, placement taskassignment.Placement, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) { + consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, placement, capacity) if err != nil { return nil, err } return assignmentqueue.ConsumerComponent{ Consumer: consumer, Capacity: capacity, Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission, OnEvent: func(event assignmentqueue.Event) { - slog.Info("assignment transition", "component", "worker", "event", event.Name, "backend", event.Backend, "assignment", event.AssignmentID, "retry_delay", event.RetryDelay) + slog.Info("assignment transition", "component", "worker", "event", event.Name, "placement", event.Placement.Key(), "assignment", event.AssignmentID, "retry_delay", event.RetryDelay) }}, OnError: func(err error) { - slog.Error("assignment processing error", "component", "worker", "backend", backend, "error", err) + slog.Error("assignment processing error", "component", "worker", "placement", placement.Key(), "error", err) }, }, nil } -func workerEventLogger(backend taskassignment.Backend) func(taskworker.Event) { +func workerEventLogger(placement taskassignment.Placement) func(taskworker.Event) { return func(event taskworker.Event) { - slog.Info("executor transition", "component", "worker", "event", event.Name, "backend", backend, "assignment", event.AssignmentID, "executor", event.Executor, "phase", event.Phase) + slog.Info("executor transition", "component", "worker", "event", event.Name, "placement", placement.Key(), "assignment", event.AssignmentID, "executor", event.Executor, "phase", event.Phase) } } @@ -364,18 +364,18 @@ func loadControllerConfig() (controllerConfig, error) { 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.KubernetesWorker) && config.PodImage == "" { + return controllerConfig{}, errors.New("POD_EXECUTOR_IMAGE is required for kubernetes-worker") } - if slices.Contains(selection, controller.PodWorker) && config.SPIREAgentID == "" { - return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker") + if slices.Contains(selection, controller.KubernetesWorker) && config.SPIREAgentID == "" { + return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for kubernetes-worker") } - if slices.Contains(selection, controller.VMWorker) { + if slices.Contains(selection, controller.OpenSandboxWorker) { if config.VMRunnerLabel != "vm" && config.VMRunnerLabel != "vm-dev" { return controllerConfig{}, errors.New("VM_RUNNER_LABEL must be vm or vm-dev") } if config.OpenSandboxURL == "" { - return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker") + return controllerConfig{}, errors.New("OPENSANDBOX_API is required for opensandbox-worker") } config.OpenSandboxAPIKey, err = read("OPENSANDBOX_API_KEY_FILE") if err != nil { diff --git a/cmd/gitea-dynamic-runner/controller_test.go b/cmd/gitea-dynamic-runner/controller_test.go index 0566001..a3ba373 100644 --- a/cmd/gitea-dynamic-runner/controller_test.go +++ b/cmd/gitea-dynamic-runner/controller_test.go @@ -34,7 +34,7 @@ func TestLoadControllerConfigUsesFileSecrets(t *testing.T) { if err != nil { t.Fatal(err) } - if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.PodWorker { + if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.KubernetesWorker { t.Fatalf("components = %#v", config.Components) } if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSProducerPassword != "producer-password" || config.NATSWorkerPassword != "worker-password" { diff --git a/docs/design-principles.md b/docs/design-principles.md index 3f87fe8..d0122d1 100644 --- a/docs/design-principles.md +++ b/docs/design-principles.md @@ -2,19 +2,22 @@ ## 对 workflow 的接口 -Runner 只向 workflow 暴露两个执行环境: +Runner 将执行环境的能力类型与实现 driver 分开声明: ```yaml -runs-on: [self-hosted, pod] +runs-on: [self-hosted, container, kubernetes] ``` ```yaml -runs-on: [self-hosted, vm] +runs-on: [self-hosted, vm, opensandbox] ``` - `self-hosted` 是固定前缀。 -- `pod` 表示一次性 Kubernetes Pod,承担常规 CI、镜像构建和 kind 等任务。 -- `vm` 表示一次性 microVM,承担需要独立内核、KVM、systemd 或更强隔离的任务。 +- `container`、`vm` 是 workload class;`kubernetes`、`opensandbox` 是 placement driver。 +- 当前支持 `container+kubernetes` 和 `vm+opensandbox`。旧 `pod` 与 `vm` 标签分别是 + 两个组合的严格兼容别名,显式指定 driver 后不得因容量或故障回退到另一 driver。 +- container 承担常规 CI、镜像构建和 kind 等任务;VM 承担需要独立内核、KVM、 + systemd 或更强隔离的任务。 执行后端是基础设施选择,不是权限角色。workflow 不需要额外声明由 controller 维护的 role 或权限 label。 diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index f193c8c..9e66532 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -3,9 +3,9 @@ ## 目标 长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner -协议客户端注册,并声明 `self-hosted`、`pod` 和 `vm` labels;单个 registration 内按总 -配置容量启动多个 `FetchTask` goroutine,再将 task 按 `runs-on` 交给 Pod 或 VM 的独立 -容量池,由一次性 Pod 或 microVM 执行。 +协议客户端注册,并声明 workload class 与 driver labels;单个 registration 内按总 +配置容量启动多个 `FetchTask` goroutine,再将 task 按 `runs-on` 交给 placement 对应的 +独立容量池,由一次性 Pod 或 microVM 执行。 ```text Gitea RunnerService @@ -13,14 +13,14 @@ Gitea RunnerService ▼ dynamic-runner scheduler │ 已领取的 task + lease - ├── Pod executor - └── microVM executor + ├── container.kubernetes executor + └── vm.opensandbox executor │ logs / state / result └──────────────────────► Gitea ``` -controller 使用单一 Go 二进制;默认在同一进程启用 `scheduler`、`pod-worker` 和 -`vm-worker`,也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界, +controller 使用单一 Go 二进制;默认在同一进程启用 `scheduler`、`kubernetes-worker` +和 `opensandbox-worker`,也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界, 共享进程不意味着共享后端状态或把 assignment 降级为内存 channel。 首轮集成的 facade pending/claimed registry 与三个组件同进程。虽然二进制保留组件选择 @@ -47,10 +47,13 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 ## 设计约束 -- 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。 +- 规范接口为 `[self-hosted, container, kubernetes]` 和 + `[self-hosted, vm, opensandbox]`。旧 `pod`、`vm`、`vm-dev` 标签保留严格映射,不能与 + 冲突 class/driver 混用;显式 driver 不允许自动回退。 - scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`,不为并发槽位重复注册; `POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。 -- task 领取并持久化后按 backend 进入独立 durable consumer;对应容量池已满时延迟 NAK, +- task 领取并持久化后按 placement 进入独立 durable consumer;v2 subject 为 + `..`。对应容量池已满时延迟 NAK, assignment 保持 JetStream pending,且不得创建超出配置容量的 workload。 - scheduler Declare 后使用 RunnerService 长轮询;一旦 FetchTask 返回已分配 task,在 JetStream publish 成功前只重试该 assignment,不领取下一项。 @@ -66,7 +69,7 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 - JetStream 只持久化和投递 assignment,不保存 executor 生命周期状态。Pod labels/annotations 与 OpenSandbox metadata 是后端运行状态的权威来源,Gitea 是 task 终态的权威来源。 - assignment 使用版本化 envelope 保存完整 Gitea protobuf task,并从 workflow `runs-on` - 严格选择 pod 或 vm subject;消费者解码后重新派生 backend 与身份,拒绝被篡改的冗余字段。 + 严格选择 workload class 与 driver;消费者解码后重新派生 placement 与身份,拒绝被篡改的冗余字段。 - JetStream 的 message ID 等于稳定 assignment ID `gitea-task-`,仅用于发布去重, 不承担 executor 生命周期记录。 - worker 按稳定 assignment ID reconcile 后端资源,进程内只保留并发控制等可丢弃状态; @@ -76,7 +79,7 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 - executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后,facade 在后端 metadata 写入持久 terminal marker;backend reconciler 仅在执行环境也进入终态后清理, 从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。 -- pod 与 vm 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment +- 每个 placement 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment 幂等落到后端;executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得 Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。 - admission pool 只保存可重建的并发状态:启动时从 Pod labels/annotations 或 OpenSandbox @@ -86,10 +89,10 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment;无需把完整 task 写入 Pod annotation、OpenSandbox metadata 或环境变量。 - Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。 -- scheduler 在 assignment 持久化到 JetStream 后即可继续领取;Pod 与 VM 分别由 durable +- scheduler 在 assignment 持久化到 JetStream 后即可继续领取;各 placement 分别由 durable consumer 的 capacity 限制并发,不共享全局执行槽位。未知后端故障由对应 consumer 的 NAK/redelivery 收敛,不能阻塞另一种 backend。 -- 两种 backend 都注入同一份 runner bootstrap 环境;Pod 仍由 homelab Kubernetes 原生 +- 两种现有 driver 都注入同一份 runner bootstrap 环境;container 仍由 homelab Kubernetes 原生 创建,只有 VM 经 OpenSandbox 创建,bootstrap 机制不改变 backend 边界。 ## 实现顺序 diff --git a/internal/assignmentqueue/jetstream.go b/internal/assignmentqueue/jetstream.go index 74e8ce6..80de9d5 100644 --- a/internal/assignmentqueue/jetstream.go +++ b/internal/assignmentqueue/jetstream.go @@ -19,7 +19,7 @@ type publishAPI interface { PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error) } -// Publisher implements the scheduler dispatcher with one subject per backend. +// Publisher implements the scheduler dispatcher with one subject per placement. type Publisher struct { JetStream publishAPI SubjectBase string @@ -38,7 +38,7 @@ func (p Publisher) Dispatch(ctx context.Context, assignment taskassignment.Assig return errors.New("assignment subject base is required") } message := &nats.Msg{ - Subject: base + "." + string(assignment.Backend), + Subject: base + "." + assignment.Placement.Key(), Header: nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}}, Data: body, } @@ -86,13 +86,13 @@ type Processor struct { type Event struct { Name string AssignmentID string - Backend taskassignment.Backend + Placement taskassignment.Placement RetryDelay time.Duration } func (p Processor) event(name string, assignment taskassignment.Assignment, retryDelay time.Duration) { if p.OnEvent != nil { - p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Backend: assignment.Backend, RetryDelay: retryDelay}) + p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Placement: assignment.Placement, RetryDelay: retryDelay}) } } @@ -178,24 +178,25 @@ type consumerManager interface { // OpenConsumer creates the durable backend cursor. Capacity is enforced both // server-side and by ConsumerComponent's local semaphore. -func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, backend taskassignment.Backend, capacity int) (jetstream.Consumer, error) { +func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, placement taskassignment.Placement, capacity int) (jetstream.Consumer, error) { if manager == nil || stream == "" || strings.TrimSuffix(subjectBase, ".") == "" || capacity < 1 { return nil, errors.New("JetStream manager, stream, subject base, and positive capacity are required") } - if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM { - return nil, fmt.Errorf("unsupported assignment backend %q", backend) + if err := placement.Validate(); err != nil { + return nil, err } + key := placement.Key() consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{ - Name: string(backend), - Durable: string(backend), - FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + string(backend), + Name: key, + Durable: key, + FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + key, AckPolicy: jetstream.AckExplicitPolicy, AckWait: 5 * time.Minute, MaxAckPending: capacity, MaxDeliver: 1000, }) if err != nil { - return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err) + return nil, fmt.Errorf("open %s assignment consumer: %w", key, err) } return consumer, nil } diff --git a/internal/assignmentqueue/jetstream_test.go b/internal/assignmentqueue/jetstream_test.go index b053208..9c1490f 100644 --- a/internal/assignmentqueue/jetstream_test.go +++ b/internal/assignmentqueue/jetstream_test.go @@ -38,13 +38,13 @@ func (p *fakePublisher) PublishMsg(_ context.Context, message *nats.Msg, _ ...je return &jetstream.PubAck{}, nil } -func TestPublisherUsesBackendSubjectAndAssignmentDeduplication(t *testing.T) { +func TestPublisherUsesPlacementSubjectAndAssignmentDeduplication(t *testing.T) { api := &fakePublisher{} publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"} if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil { t.Fatal(err) } - if api.message.Subject != "ci.assignment.pod" { + if api.message.Subject != "ci.assignment.container.kubernetes" { t.Fatalf("subject = %q", api.message.Subject) } if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" { @@ -139,7 +139,7 @@ func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) { t.Fatalf("events = %#v", events) } for index := range want { - if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Backend != taskassignment.BackendPod { + if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Placement != taskassignment.KubernetesContainer { t.Fatalf("event[%d] = %#v", index, events[index]) } } @@ -210,12 +210,12 @@ func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string return nil, nil } -func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) { +func TestOpenConsumerUsesIndependentDurablePerPlacement(t *testing.T) { manager := &fakeConsumerManager{} - if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil { + if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.KubernetesContainer, 4); err != nil { t.Fatal(err) } - if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 { + if manager.config.Durable != "container.kubernetes" || manager.config.FilterSubject != "ci.assignment.container.kubernetes" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 { t.Fatalf("config = %#v", manager.config) } } diff --git a/internal/controller/components.go b/internal/controller/components.go index 1fcdf48..e3db264 100644 --- a/internal/controller/components.go +++ b/internal/controller/components.go @@ -14,12 +14,12 @@ import ( type ComponentName string const ( - Scheduler ComponentName = "scheduler" - PodWorker ComponentName = "pod-worker" - VMWorker ComponentName = "vm-worker" + Scheduler ComponentName = "scheduler" + KubernetesWorker ComponentName = "kubernetes-worker" + OpenSandboxWorker ComponentName = "opensandbox-worker" ) -var defaultComponents = []ComponentName{Scheduler, PodWorker, VMWorker} +var defaultComponents = []ComponentName{Scheduler, KubernetesWorker, OpenSandboxWorker} // Selection parses --components. An empty value enables all components. type Selection []ComponentName @@ -31,6 +31,12 @@ func ParseSelection(value string) (Selection, error) { var selected Selection for _, raw := range strings.Split(value, ",") { name := ComponentName(strings.TrimSpace(raw)) + switch name { + case "pod-worker": + name = KubernetesWorker + case "vm-worker": + name = OpenSandboxWorker + } if !slices.Contains(defaultComponents, name) { return nil, fmt.Errorf("unknown controller component %q", name) } diff --git a/internal/controller/components_test.go b/internal/controller/components_test.go index a5cce8c..780dbc2 100644 --- a/internal/controller/components_test.go +++ b/internal/controller/components_test.go @@ -13,18 +13,18 @@ func TestParseSelectionDefaultsToAll(t *testing.T) { if err != nil { t.Fatal(err) } - if len(selection) != 3 || selection[0] != Scheduler || selection[1] != PodWorker || selection[2] != VMWorker { + if len(selection) != 3 || selection[0] != Scheduler || selection[1] != KubernetesWorker || selection[2] != OpenSandboxWorker { t.Fatalf("selection = %v", selection) } } } func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) { - selection, err := ParseSelection("vm-worker,scheduler,vm-worker") + selection, err := ParseSelection("opensandbox-worker,scheduler,opensandbox-worker") if err != nil { t.Fatal(err) } - if len(selection) != 2 || selection[0] != VMWorker || selection[1] != Scheduler { + if len(selection) != 2 || selection[0] != OpenSandboxWorker || selection[1] != Scheduler { t.Fatalf("selection = %v", selection) } if _, err := ParseSelection("webhook"); err == nil { @@ -32,6 +32,16 @@ func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) { } } +func TestParseSelectionNormalizesLegacyWorkerNames(t *testing.T) { + selection, err := ParseSelection("pod-worker,vm-worker") + if err != nil { + t.Fatal(err) + } + if len(selection) != 2 || selection[0] != KubernetesWorker || selection[1] != OpenSandboxWorker { + t.Fatalf("selection = %v", selection) + } +} + type componentFunc func(context.Context) error func (f componentFunc) Run(ctx context.Context) error { return f(ctx) } @@ -45,14 +55,14 @@ func TestRunStartsSelectedComponentsAndCancelsPeers(t *testing.T) { started <- Scheduler return errors.New("poll failed") }), - PodWorker: componentFunc(func(ctx context.Context) error { - started <- PodWorker + KubernetesWorker: componentFunc(func(ctx context.Context) error { + started <- KubernetesWorker <-ctx.Done() once.Do(func() { close(peerStopped) }) return ctx.Err() }), } - err := Run(context.Background(), Selection{Scheduler, PodWorker}, registry) + err := Run(context.Background(), Selection{Scheduler, KubernetesWorker}, registry) if err == nil || !errors.Is(err, context.Canceled) && err.Error() != "component scheduler: poll failed" { t.Fatalf("Run() error = %v", err) } diff --git a/internal/opensandboxbackend/backend.go b/internal/opensandboxbackend/backend.go index d81d780..aaf5770 100644 --- a/internal/opensandboxbackend/backend.go +++ b/internal/opensandboxbackend/backend.go @@ -119,7 +119,10 @@ func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([] return nil, err } result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{ - Metadata: map[string]string{"ci.ddupan.top/backend": "vm"}, PageSize: 100, + Metadata: map[string]string{ + "ci.ddupan.top/workload-class": "vm", + "ci.ddupan.top/driver": "opensandbox", + }, PageSize: 100, }) if err != nil { return nil, fmt.Errorf("list recoverable sandboxes: %w", err) @@ -172,8 +175,8 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen if err := b.validate(); err != nil { return nil, err } - if assignment.Backend != taskassignment.BackendVM { - return nil, fmt.Errorf("OpenSandbox backend cannot create %q assignment", assignment.Backend) + if assignment.Placement != taskassignment.OpenSandboxVM { + return nil, fmt.Errorf("OpenSandbox backend cannot create %q", assignment.Placement.Key()) } environment := clone(b.Config.Env) for key, value := range launch.Environment { diff --git a/internal/opensandboxbackend/backend_test.go b/internal/opensandboxbackend/backend_test.go index 4f8dfc0..885d30c 100644 --- a/internal/opensandboxbackend/backend_test.go +++ b/internal/opensandboxbackend/backend_test.go @@ -63,7 +63,7 @@ func backend(lifecycle Lifecycle) Backend { func assignment() taskassignment.Assignment { return taskassignment.Assignment{ - ID: "gitea-task-42", Backend: taskassignment.BackendVM, + ID: "gitea-task-42", Placement: taskassignment.OpenSandboxVM, Task: &runnerv1.Task{Id: 42}, Identity: taskidentity.Identity{Repository: "owner/repo", Task: "publish", SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"}, } diff --git a/internal/podbackend/backend.go b/internal/podbackend/backend.go index 8f24039..2cc85da 100644 --- a/internal/podbackend/backend.go +++ b/internal/podbackend/backend.go @@ -120,7 +120,7 @@ func (b Backend) RecoverAssignments(ctx context.Context) ([]taskassignment.Assig if err := b.validate(); err != nil { return nil, err } - pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/backend=pod") + pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/workload-class=container,ci.ddupan.top/driver=kubernetes") if err != nil { return nil, fmt.Errorf("list recoverable assignment Pods: %w", err) } @@ -159,8 +159,8 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen if err := b.validate(); err != nil { return nil, err } - if assignment.Backend != taskassignment.BackendPod { - return nil, fmt.Errorf("Pod backend cannot create %q assignment", assignment.Backend) + if assignment.Placement != taskassignment.KubernetesContainer { + return nil, fmt.Errorf("Kubernetes backend cannot create %q", assignment.Placement.Key()) } labels := clone(launch.Metadata.Labels) labels["app.kubernetes.io/name"] = "gitea-dynamic-runner" diff --git a/internal/podbackend/backend_test.go b/internal/podbackend/backend_test.go index e1d96ba..e928ae8 100644 --- a/internal/podbackend/backend_test.go +++ b/internal/podbackend/backend_test.go @@ -59,7 +59,7 @@ func backend(api API) Backend { func assignment() taskassignment.Assignment { return taskassignment.Assignment{ - ID: "gitea-task-42", Backend: taskassignment.BackendPod, + ID: "gitea-task-42", Placement: taskassignment.KubernetesContainer, Task: &runnerv1.Task{Id: 42}, Identity: taskidentity.Identity{ Repository: "owner/repo", Task: "publish", diff --git a/internal/runnerbootstrap/bootstrap.go b/internal/runnerbootstrap/bootstrap.go index c4151ee..e04499a 100644 --- a/internal/runnerbootstrap/bootstrap.go +++ b/internal/runnerbootstrap/bootstrap.go @@ -13,12 +13,13 @@ import ( ) const ( - EnvAssignmentID = "CI_ASSIGNMENT_ID" - EnvCapability = "CI_RUNNER_CAPABILITY" - EnvFacadeURL = "CI_RUNNER_FACADE_URL" - EnvFacadeID = "CI_RUNNER_FACADE_SPIFFE_ID" - EnvSPIFFEID = "CI_SPIFFE_ID" - EnvBackend = "CI_RUNNER_BACKEND" + EnvAssignmentID = "CI_ASSIGNMENT_ID" + EnvCapability = "CI_RUNNER_CAPABILITY" + EnvFacadeURL = "CI_RUNNER_FACADE_URL" + EnvFacadeID = "CI_RUNNER_FACADE_SPIFFE_ID" + EnvSPIFFEID = "CI_SPIFFE_ID" + EnvWorkloadClass = "CI_WORKLOAD_CLASS" + EnvDriver = "CI_WORKLOAD_DRIVER" ) // Bootstrap emits assignment-scoped launch configuration. FacadeURL is the @@ -48,7 +49,8 @@ func (b Bootstrap) Environment(assignment taskassignment.Assignment) (map[string EnvFacadeURL: b.FacadeURL, EnvFacadeID: b.FacadeSPIFFEID, EnvSPIFFEID: assignment.Identity.SPIFFEID, - EnvBackend: string(assignment.Backend), + EnvWorkloadClass: string(assignment.Placement.Class), + EnvDriver: string(assignment.Placement.Driver), "SPIFFE_ENDPOINT_SOCKET": b.WorkloadAPIAddr, }, nil } @@ -67,7 +69,7 @@ type Registration struct { Ephemeral bool `json:"ephemeral"` } -func RegistrationJSON(assignmentID, capability, localProxyURL string, backend taskassignment.Backend) ([]byte, error) { +func RegistrationJSON(assignmentID, capability, localProxyURL string, placement taskassignment.Placement) ([]byte, error) { if assignmentID == "" || capability == "" { return nil, errors.New("assignment ID and runner capability are required") } @@ -75,13 +77,13 @@ func RegistrationJSON(assignmentID, capability, localProxyURL string, backend ta if err != nil || parsed.Scheme != "http" || parsed.Host == "" { return nil, errors.New("local runner proxy URL must be an absolute http URL") } - if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM { - return nil, fmt.Errorf("unsupported runner backend %q", backend) + if err := placement.Validate(); err != nil { + return nil, err } registration := Registration{ Warning: "Generated for one preassigned task by gitea-dynamic-runner.", UUID: assignmentID, Name: assignmentID, Token: capability, - Address: localProxyURL, Labels: []string{"self-hosted", string(backend)}, Ephemeral: true, + Address: localProxyURL, Labels: []string{"self-hosted", string(placement.Class), string(placement.Driver)}, Ephemeral: true, } data, err := json.MarshalIndent(registration, "", " ") if err != nil { diff --git a/internal/runnerbootstrap/bootstrap_test.go b/internal/runnerbootstrap/bootstrap_test.go index 0ab3fbf..5f2aaa9 100644 --- a/internal/runnerbootstrap/bootstrap_test.go +++ b/internal/runnerbootstrap/bootstrap_test.go @@ -25,7 +25,7 @@ func testBootstrap(t *testing.T) Bootstrap { func testAssignment() taskassignment.Assignment { return taskassignment.Assignment{ - ID: "gitea-task-42", Backend: taskassignment.BackendPod, + ID: "gitea-task-42", Placement: taskassignment.KubernetesContainer, Identity: taskidentity.Identity{SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"}, } } @@ -46,7 +46,7 @@ func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) { if first[EnvAssignmentID] != "gitea-task-42" || first[EnvSPIFFEID] != testAssignment().Identity.SPIFFEID { t.Fatalf("environment = %#v", first) } - if first[EnvBackend] != "pod" || first[EnvFacadeID] == "" { + if first[EnvWorkloadClass] != "container" || first[EnvDriver] != "kubernetes" || first[EnvFacadeID] == "" { t.Fatalf("environment = %#v", first) } if first["SPIFFE_ENDPOINT_SOCKET"] != "unix:///run/spire/agent-sockets/spire-agent.sock" { @@ -55,7 +55,7 @@ func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) { } func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) { - data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.BackendVM) + data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.OpenSandboxVM) if err != nil { t.Fatal(err) } @@ -66,13 +66,13 @@ func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) { if registration.UUID != "gitea-task-42" || registration.Token != "capability" || registration.Address != "http://127.0.0.1:8080" || !registration.Ephemeral { t.Fatalf("registration = %#v", registration) } - if len(registration.Labels) != 2 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" { + if len(registration.Labels) != 3 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" || registration.Labels[2] != "opensandbox" { t.Fatalf("labels = %#v", registration.Labels) } } func TestRegistrationRejectsNonLocalTLSAddress(t *testing.T) { - if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.BackendPod); err == nil { + if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.KubernetesContainer); err == nil { t.Fatal("expected local proxy URL validation error") } } diff --git a/internal/runnerbootstrap/executor.go b/internal/runnerbootstrap/executor.go index 4158b81..c3415e0 100644 --- a/internal/runnerbootstrap/executor.go +++ b/internal/runnerbootstrap/executor.go @@ -17,7 +17,7 @@ import ( type ExecutorConfig struct { AssignmentID string Capability string - Backend taskassignment.Backend + Placement taskassignment.Placement FacadeURL string FacadeSPIFFEID string WorkloadAPIAddr string @@ -71,7 +71,7 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error { defer os.RemoveAll(workDir) } registration, err := RegistrationJSON( - config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Backend, + config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Placement, ) if err != nil { return err @@ -144,13 +144,13 @@ func waitForFacade(ctx context.Context, endpoint string) error { // 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) + placement := taskassignment.Placement{Class: taskassignment.WorkloadClass(os.Getenv(EnvWorkloadClass)), Driver: taskassignment.Driver(os.Getenv(EnvDriver))} + if err := placement.Validate(); err != nil { + return ExecutorConfig{}, err } config := ExecutorConfig{ AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability), - Backend: backend, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID), + Placement: placement, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID), RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), RunnerConfig: os.Getenv("GITEA_RUNNER_CONFIG_FILE"), ListenAddress: "127.0.0.1:0", Stdout: os.Stdout, Stderr: os.Stderr, diff --git a/internal/taskassignment/assignment.go b/internal/taskassignment/assignment.go index 497df49..2c5f0aa 100644 --- a/internal/taskassignment/assignment.go +++ b/internal/taskassignment/assignment.go @@ -6,7 +6,6 @@ import ( "encoding/json" "errors" "fmt" - "slices" "strconv" "gitea.dev/actionslib/pkg/model" @@ -16,34 +15,27 @@ import ( "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" ) -const wireVersion = 1 - -type Backend string - -const ( - BackendPod Backend = "pod" - BackendVM Backend = "vm" -) +const wireVersion = 2 // Assignment is the only document persisted in the handoff queue. type Assignment struct { - ID string - Backend Backend - Task *runnerv1.Task - Identity taskidentity.Identity + ID string + Placement Placement + Task *runnerv1.Task + Identity taskidentity.Identity } // FromMetadata reconstructs the minimal assignment needed to authorize an -// already-running executor after a controller restart. Backend metadata was +// already-running executor after a controller restart. Placement metadata was // originally derived from the trusted Gitea task and is validated again here. func FromMetadata(labels, annotations map[string]string, trustDomain string) (Assignment, error) { taskID, err := strconv.ParseInt(labels["ci.ddupan.top/task-id"], 10, 64) if err != nil || taskID < 1 { return Assignment{}, errors.New("backend metadata has invalid task ID") } - backend := Backend(labels["ci.ddupan.top/backend"]) - if backend != BackendPod && backend != BackendVM { - return Assignment{}, errors.New("backend metadata has invalid backend") + placement := Placement{Class: WorkloadClass(labels["ci.ddupan.top/workload-class"]), Driver: Driver(labels["ci.ddupan.top/driver"])} + if err := placement.Validate(); err != nil { + return Assignment{}, err } id := labels["ci.ddupan.top/assignment-id"] if id != fmt.Sprintf("gitea-task-%d", taskID) { @@ -56,15 +48,15 @@ func FromMetadata(labels, annotations map[string]string, trustDomain string) (As if err != nil { return Assignment{}, err } - return Assignment{ID: id, Backend: backend, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil + return Assignment{ID: id, Placement: placement, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil } type envelope struct { - Version int `json:"version"` - ID string `json:"id"` - Backend Backend `json:"backend"` - Task []byte `json:"task"` - Identity taskidentity.Identity `json:"identity"` + Version int `json:"version"` + ID string `json:"id"` + Placement Placement `json:"placement"` + Task []byte `json:"task"` + Identity taskidentity.Identity `json:"identity"` } // New derives all trusted assignment fields from the task fetched from Gitea. @@ -76,40 +68,28 @@ func New(task *runnerv1.Task, trustDomain string) (Assignment, error) { if err != nil { return Assignment{}, err } - backend, err := backendFromTask(task) + placement, err := placementFromTask(task) if err != nil { return Assignment{}, err } return Assignment{ - ID: fmt.Sprintf("gitea-task-%d", task.GetId()), - Backend: backend, - Task: task, - Identity: identity, + ID: fmt.Sprintf("gitea-task-%d", task.GetId()), + Placement: placement, + Task: task, + Identity: identity, }, nil } -func backendFromTask(task *runnerv1.Task) (Backend, error) { +func placementFromTask(task *runnerv1.Task) (Placement, error) { workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload())) if err != nil { - return "", fmt.Errorf("parse task workflow for backend: %w", err) + return Placement{}, fmt.Errorf("parse task workflow for placement: %w", err) } jobIDs := workflow.GetJobIDs() if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil { - return "", fmt.Errorf("task workflow must contain exactly one non-empty job") + return Placement{}, fmt.Errorf("task workflow must contain exactly one non-empty job") } - labels := workflow.GetJob(jobIDs[0]).RunsOnLabels() - if !slices.Contains(labels, "self-hosted") { - return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels) - } - hasPod := slices.Contains(labels, string(BackendPod)) - hasVM := slices.Contains(labels, string(BackendVM)) || slices.Contains(labels, "vm-dev") - if hasPod == hasVM { - return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels) - } - if hasPod { - return BackendPod, nil - } - return BackendVM, nil + return PlacementFromLabels(workflow.GetJob(jobIDs[0]).RunsOnLabels()) } // Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task. @@ -117,13 +97,16 @@ func Marshal(assignment Assignment) ([]byte, error) { if assignment.Task == nil { return nil, errors.New("assignment task is required") } + if err := assignment.Placement.Validate(); err != nil { + return nil, err + } task, err := proto.Marshal(assignment.Task) if err != nil { return nil, fmt.Errorf("marshal Gitea task: %w", err) } return json.Marshal(envelope{ Version: wireVersion, - ID: assignment.ID, Backend: assignment.Backend, + ID: assignment.ID, Placement: assignment.Placement, Task: task, Identity: assignment.Identity, }) } @@ -145,7 +128,7 @@ func Unmarshal(data []byte, trustDomain string) (Assignment, error) { if err != nil { return Assignment{}, err } - if wire.ID != canonical.ID || wire.Backend != canonical.Backend || wire.Identity != canonical.Identity { + if wire.ID != canonical.ID || wire.Placement != canonical.Placement || wire.Identity != canonical.Identity { return Assignment{}, errors.New("assignment metadata does not match its Gitea task") } return canonical, nil diff --git a/internal/taskassignment/assignment_test.go b/internal/taskassignment/assignment_test.go index d2413db..0733338 100644 --- a/internal/taskassignment/assignment_test.go +++ b/internal/taskassignment/assignment_test.go @@ -21,29 +21,37 @@ func task(t *testing.T, labels string) *runnerv1.Task { } } -func TestNewSelectsBackendFromRunsOn(t *testing.T) { +func TestNewSelectsPlacementFromRunsOn(t *testing.T) { for _, test := range []struct { - labels string - backend Backend + labels string + placement Placement }{ - {"[self-hosted, pod]", BackendPod}, - {"[self-hosted, vm]", BackendVM}, - {"[self-hosted, vm-dev]", BackendVM}, + {"[self-hosted, pod]", KubernetesContainer}, + {"[self-hosted, container]", KubernetesContainer}, + {"[self-hosted, container, kubernetes]", KubernetesContainer}, + {"[self-hosted, vm]", OpenSandboxVM}, + {"[self-hosted, vm-dev]", OpenSandboxVM}, + {"[self-hosted, vm, opensandbox]", OpenSandboxVM}, } { assignment, err := New(task(t, test.labels), "ddupan.top") if err != nil { t.Fatal(err) } - if assignment.Backend != test.backend || assignment.ID != "gitea-task-42" { + if assignment.Placement != test.placement || assignment.ID != "gitea-task-42" { t.Fatalf("assignment = %#v", assignment) } } } -func TestNewRejectsAmbiguousBackend(t *testing.T) { +func TestNewRejectsInvalidPlacement(t *testing.T) { for _, labels := range []string{ "[self-hosted]", "[self-hosted, pod, vm]", + "[self-hosted, pod, container]", + "[self-hosted, pod, kubernetes]", + "[self-hosted, container, opensandbox]", + "[self-hosted, vm, kubernetes]", + "[self-hosted, container, kubernetes, opensandbox]", "[pod]", } { if _, err := New(task(t, labels), "ddupan.top"); err == nil { @@ -65,27 +73,28 @@ func TestAssignmentWireRoundTripAndValidation(t *testing.T) { if err != nil { t.Fatal(err) } - if got.ID != want.ID || got.Backend != want.Backend || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) { + if got.ID != want.ID || got.Placement != want.Placement || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) { t.Fatalf("round trip = %#v, want %#v", got, want) } - tampered := bytes.Replace(data, []byte(`"backend":"pod"`), []byte(`"backend":"vm"`), 1) + tampered := bytes.Replace(data, []byte(`"driver":"kubernetes"`), []byte(`"driver":"opensandbox"`), 1) if _, err := Unmarshal(tampered, "ddupan.top"); err == nil { - t.Fatal("expected tampered backend to fail") + t.Fatal("expected tampered placement to fail") } } func TestFromMetadataRecoversMinimalAssignment(t *testing.T) { assignment, err := FromMetadata(map[string]string{ - "ci.ddupan.top/assignment-id": "gitea-task-42", - "ci.ddupan.top/task-id": "42", - "ci.ddupan.top/backend": "vm", + "ci.ddupan.top/assignment-id": "gitea-task-42", + "ci.ddupan.top/task-id": "42", + "ci.ddupan.top/workload-class": "vm", + "ci.ddupan.top/driver": "opensandbox", }, map[string]string{ "ci.ddupan.top/repository": "owner/repo", "ci.ddupan.top/job-key": "publish", "ci.ddupan.top/spiffe-id": "spiffe://ddupan.top/ci/owner/repo/publish", }, "ddupan.top") - if err != nil || assignment.Task.GetId() != 42 || assignment.Backend != BackendVM { + if err != nil || assignment.Task.GetId() != 42 || assignment.Placement != OpenSandboxVM { t.Fatalf("assignment=%#v err=%v", assignment, err) } } diff --git a/internal/taskassignment/placement.go b/internal/taskassignment/placement.go new file mode 100644 index 0000000..8b218a5 --- /dev/null +++ b/internal/taskassignment/placement.go @@ -0,0 +1,76 @@ +package taskassignment + +import ( + "errors" + "fmt" + "slices" +) + +type WorkloadClass string + +const ( + WorkloadContainer WorkloadClass = "container" + WorkloadVM WorkloadClass = "vm" +) + +type Driver string + +const ( + DriverKubernetes Driver = "kubernetes" + DriverOpenSandbox Driver = "opensandbox" +) + +type Placement struct { + Class WorkloadClass `json:"workload_class"` + Driver Driver `json:"driver"` +} + +var ( + KubernetesContainer = Placement{Class: WorkloadContainer, Driver: DriverKubernetes} + OpenSandboxVM = Placement{Class: WorkloadVM, Driver: DriverOpenSandbox} +) + +func (p Placement) Validate() error { + switch p { + case KubernetesContainer, OpenSandboxVM: + return nil + default: + return fmt.Errorf("unsupported workload placement %s/%s", p.Class, p.Driver) + } +} + +func (p Placement) Key() string { return string(p.Class) + "." + string(p.Driver) } + +func PlacementFromLabels(labels []string) (Placement, error) { + if !slices.Contains(labels, "self-hosted") { + return Placement{}, fmt.Errorf("task runs-on labels must include self-hosted: %v", labels) + } + legacyPod := slices.Contains(labels, "pod") + container := slices.Contains(labels, string(WorkloadContainer)) + vm := slices.Contains(labels, string(WorkloadVM)) || slices.Contains(labels, "vm-dev") + kubernetes := slices.Contains(labels, string(DriverKubernetes)) + opensandbox := slices.Contains(labels, string(DriverOpenSandbox)) + + if legacyPod { + if container || vm || kubernetes || opensandbox { + return Placement{}, errors.New("legacy pod label cannot be combined with workload or VM driver labels") + } + return KubernetesContainer, nil + } + if container == vm { + return Placement{}, fmt.Errorf("task runs-on labels must select exactly one workload class: %v", labels) + } + if kubernetes && opensandbox { + return Placement{}, fmt.Errorf("task runs-on labels select multiple drivers: %v", labels) + } + if container { + if opensandbox { + return Placement{}, fmt.Errorf("opensandbox does not support container workloads") + } + return KubernetesContainer, nil + } + if kubernetes { + return Placement{}, fmt.Errorf("kubernetes does not support VM workloads") + } + return OpenSandboxVM, nil +} diff --git a/internal/taskworker/worker.go b/internal/taskworker/worker.go index ea590ad..f307df9 100644 --- a/internal/taskworker/worker.go +++ b/internal/taskworker/worker.go @@ -206,10 +206,11 @@ func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, er func BackendMetadata(assignment taskassignment.Assignment) Metadata { return Metadata{ Labels: map[string]string{ - "ci.ddupan.top/runner": "true", - "ci.ddupan.top/assignment-id": assignment.ID, - "ci.ddupan.top/task-id": strconv.FormatInt(assignment.Task.GetId(), 10), - "ci.ddupan.top/backend": string(assignment.Backend), + "ci.ddupan.top/runner": "true", + "ci.ddupan.top/assignment-id": assignment.ID, + "ci.ddupan.top/task-id": strconv.FormatInt(assignment.Task.GetId(), 10), + "ci.ddupan.top/workload-class": string(assignment.Placement.Class), + "ci.ddupan.top/driver": string(assignment.Placement.Driver), }, Annotations: map[string]string{ "ci.ddupan.top/repository": assignment.Identity.Repository, diff --git a/internal/taskworker/worker_test.go b/internal/taskworker/worker_test.go index efdf111..c76ae1d 100644 --- a/internal/taskworker/worker_test.go +++ b/internal/taskworker/worker_test.go @@ -53,9 +53,9 @@ func (t *fakeTasks) Report(_ context.Context, _ int64, phase Phase) error { func assignment() taskassignment.Assignment { return taskassignment.Assignment{ - ID: "gitea-task-42", - Backend: taskassignment.BackendPod, - Task: &runnerv1.Task{Id: 42}, + ID: "gitea-task-42", + Placement: taskassignment.KubernetesContainer, + Task: &runnerv1.Task{Id: 42}, Identity: taskidentity.Identity{ Repository: "owner/repo", Task: "publish",