diff --git a/README.md b/README.md index 6ef3332..b3e9ea4 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,8 @@ runs-on: [self-hosted, vm] 目标 Go controller 组件： - `scheduler`：以常驻 Gitea RunnerService 身份直接领取 task，并把完整 assignment - 持久化到 JetStream；同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。 + 持久化到 JetStream；一个 registration 下按配置启动多个并发 `FetchTask` goroutine， + 同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。 - `pod-worker`：直接在 homelab Kubernetes 创建一次性 Pod。 - `vm-worker`：通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。 - 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker，因为 facade @@ -37,8 +38,9 @@ runs-on: [self-hosted, vm] 动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。 Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于 -VM/Kata workload；两个 backend 使用独立 durable consumer，任一执行层故障不会阻塞另一条 -部署。长期 RunnerService 协议路线见 +VM/Kata workload；两个 backend 使用独立 durable consumer 和独立容量池。assignment 根据 +`runs-on` 进入对应池，池满时留在 JetStream pending，不会创建超出容量的 workload；任一 +执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。 ## 开发 diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go index ff74ff4..f4cca96 100644 --- a/cmd/gitea-dynamic-runner/controller.go +++ b/cmd/gitea-dynamic-runner/controller.go @@ -22,6 +22,7 @@ import ( "k8s.io/client-go/tools/leaderelection/resourcelock" "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/backendpool" "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" @@ -91,6 +92,8 @@ 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 giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken) @@ -105,6 +108,7 @@ func runController(ctx context.Context) error { 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") @@ -112,6 +116,7 @@ func runController(ctx context.Context) error { if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil { return err } + vmPool.Release(assignment.ID) } return nil }, @@ -133,7 +138,7 @@ func runController(ctx context.Context) error { Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{ JetStream: producerJS, SubjectBase: config.SubjectBase, }}, - Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels}, + Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity}, OnError: func(err error) { log.Printf("scheduler: %v", err) }, } kubernetesConfig, err := rest.InClusterConfig() @@ -180,11 +185,12 @@ func runController(ctx context.Context) error { return err } for _, assignment := range assignments { + podPool.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}, registry) + component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, podPool) if err != nil { return err } @@ -209,11 +215,12 @@ func runController(ctx context.Context) error { return err } for _, assignment := range assignments { + vmPool.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}, registry) + component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, vmPool) if err != nil { return err } @@ -280,14 +287,14 @@ 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) (controller.Component, error) { +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) if err != nil { return nil, err } return assignmentqueue.ConsumerComponent{ Consumer: consumer, Capacity: capacity, - Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims}, + Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission}, OnError: func(err error) { log.Printf("%s worker: %v", backend, err) }, }, nil } diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 27da1bf..f193c8c 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -3,8 +3,9 @@ ## 目标 长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner -协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；它只在后端存在可用容量 -时领取 task，然后将该 task 交给一个一次性 Pod 或 microVM 执行。 +协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；单个 registration 内按总 +配置容量启动多个 `FetchTask` goroutine，再将 task 按 `runs-on` 交给 Pod 或 VM 的独立 +容量池，由一次性 Pod 或 microVM 执行。 ```text Gitea RunnerService @@ -47,7 +48,10 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 ## 设计约束 - 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。 -- scheduler 在没有对应 backend 容量时不领取 task，避免本地形成不可控积压。 +- scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`，不为并发槽位重复注册； + `POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。 +- task 领取并持久化后按 backend 进入独立 durable consumer；对应容量池已满时延迟 NAK， + assignment 保持 JetStream pending，且不得创建超出配置容量的 workload。 - scheduler Declare 后使用 RunnerService 长轮询；一旦 FetchTask 返回已分配 task，在 JetStream publish 成功前只重试该 assignment，不领取下一项。 - 每个 executor 只执行一个 task，完成后销毁。 @@ -72,9 +76,11 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 - executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后，facade 在后端 metadata 写入持久 terminal marker；backend reconciler 仅在执行环境也进入终态后清理， 从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。 -- pod 与 vm 使用独立 durable consumer 和并发上限。consumer 只负责将 assignment +- pod 与 vm 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment 幂等落到后端；executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得 Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。 +- admission pool 只保存可重建的并发状态：启动时从 Pod labels/annotations 或 OpenSandbox + metadata 恢复非终态 assignment，terminal update 持久化成功后释放槽位，不引入新存储。 - assignment ACK 后的运行、结果回报和清理由 backend reconciler 根据 Kubernetes、 OpenSandbox 与 Gitea 的事实状态驱动，不继续占用 JetStream delivery。 - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment；无需把完整 diff --git a/internal/assignmentqueue/jetstream.go b/internal/assignmentqueue/jetstream.go index cb2e651..93525a8 100644 --- a/internal/assignmentqueue/jetstream.go +++ b/internal/assignmentqueue/jetstream.go @@ -57,6 +57,11 @@ type Claims interface { WaitClaimed(context.Context, string) error } +type Admission interface { + Acquire(string) bool + Release(string) +} + // Message is the subset of jetstream.Msg needed by one reconciliation. type Message interface { Data() []byte @@ -70,13 +75,14 @@ type Processor struct { TrustDomain string Accepter Accepter Claims Claims + Admission Admission RetryDelay time.Duration ClaimTimeout time.Duration } func (p Processor) Process(ctx context.Context, message Message) error { - if p.Accepter == nil || p.Claims == nil { - return errors.New("assignment accepter and claim registry are required") + if p.Accepter == nil || p.Claims == nil || p.Admission == nil { + return errors.New("assignment accepter, claim registry, and backend admission pool are required") } assignment, err := taskassignment.Unmarshal(message.Data(), p.TrustDomain) if err != nil { @@ -85,8 +91,16 @@ func (p Processor) Process(ctx context.Context, message Message) error { if _, err := p.Claims.Offer(assignment); err != nil { return errors.Join(err, message.TermWithReason("conflicting assignment")) } + if !p.Admission.Acquire(assignment.ID) { + delay := p.RetryDelay + if delay <= 0 { + delay = 2 * time.Second + } + return message.NakWithDelay(delay) + } accepted, err := p.Accepter.Accept(ctx, assignment) if err != nil { + p.Admission.Release(assignment.ID) delay := p.RetryDelay if delay <= 0 { delay = 15 * time.Second diff --git a/internal/assignmentqueue/jetstream_test.go b/internal/assignmentqueue/jetstream_test.go index 5ea2cec..6ff7254 100644 --- a/internal/assignmentqueue/jetstream_test.go +++ b/internal/assignmentqueue/jetstream_test.go @@ -61,6 +61,28 @@ type fakeClaims struct { claimed bool } +type fakeAdmission struct { + allowed bool + active map[string]bool + released int +} + +func (a *fakeAdmission) Acquire(assignmentID string) bool { + if !a.allowed { + return false + } + if a.active == nil { + a.active = make(map[string]bool) + } + a.active[assignmentID] = true + return true +} + +func (a *fakeAdmission) Release(assignmentID string) { + delete(a.active, assignmentID) + a.released++ +} + func (c *fakeClaims) Offer(taskassignment.Assignment) (<-chan struct{}, error) { ready := make(chan struct{}) if c.claimed { @@ -104,7 +126,7 @@ func encodedAssignment(t *testing.T) []byte { func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) { message := &fakeMessage{data: encodedAssignment(t)} - processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}} + processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}} if err := processor.Process(context.Background(), message); err != nil { t.Fatal(err) } @@ -115,7 +137,7 @@ func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) { func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) { message := &fakeMessage{data: encodedAssignment(t)} - processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, RetryDelay: 2 * time.Second} + processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, Admission: &fakeAdmission{allowed: true}, RetryDelay: 2 * time.Second} if err := processor.Process(context.Background(), message); err != nil { t.Fatal(err) } @@ -126,10 +148,12 @@ func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) { func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) { retry := &fakeMessage{data: encodedAssignment(t)} + admission := &fakeAdmission{allowed: true} processor := Processor{ TrustDomain: "ddupan.top", Accepter: &fakeAccepter{err: errors.New("backend unavailable")}, Claims: &fakeClaims{}, + Admission: admission, RetryDelay: time.Minute, } if err := processor.Process(context.Background(), retry); err == nil { @@ -138,6 +162,9 @@ func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) if retry.nacked != time.Minute { t.Fatalf("retry delay = %s", retry.nacked) } + if admission.released != 1 { + t.Fatalf("released slots = %d", admission.released) + } poison := &fakeMessage{data: []byte("not-json")} if err := processor.Process(context.Background(), poison); err == nil { @@ -148,6 +175,24 @@ func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) } } +func TestProcessorLeavesAssignmentPendingWhenBackendPoolIsFull(t *testing.T) { + message := &fakeMessage{data: encodedAssignment(t)} + accepter := &fakeAccepter{accepted: true} + processor := Processor{ + TrustDomain: "ddupan.top", + Accepter: accepter, + Claims: &fakeClaims{}, + Admission: &fakeAdmission{}, + RetryDelay: 3 * time.Second, + } + if err := processor.Process(context.Background(), message); err != nil { + t.Fatal(err) + } + if message.acked != 0 || message.nacked != 3*time.Second { + t.Fatalf("message = %#v", message) + } +} + type fakeConsumerManager struct{ config jetstream.ConsumerConfig } func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string, config jetstream.ConsumerConfig) (jetstream.Consumer, error) { diff --git a/internal/backendpool/pool.go b/internal/backendpool/pool.go new file mode 100644 index 0000000..57eb0f9 --- /dev/null +++ b/internal/backendpool/pool.go @@ -0,0 +1,50 @@ +// Package backendpool manages runtime capacity independently for each executor backend. +package backendpool + +import "sync" + +type Pool struct { + mu sync.Mutex + capacity int + active map[string]struct{} +} + +func New(capacity int) *Pool { + return &Pool{capacity: capacity, active: make(map[string]struct{})} +} + +// Acquire reserves a backend slot without blocking. Redelivery of the same +// assignment is idempotent and succeeds even while the pool is full. +func (p *Pool) Acquire(assignmentID string) bool { + p.mu.Lock() + defer p.mu.Unlock() + if _, exists := p.active[assignmentID]; exists { + return true + } + if assignmentID == "" || len(p.active) >= p.capacity { + return false + } + p.active[assignmentID] = struct{}{} + return true +} + +func (p *Pool) Restore(assignmentID string) { + if assignmentID == "" { + return + } + p.mu.Lock() + p.active[assignmentID] = struct{}{} + p.mu.Unlock() +} + +func (p *Pool) Release(assignmentID string) { + p.mu.Lock() + delete(p.active, assignmentID) + p.mu.Unlock() +} + +func (p *Pool) Active() int { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.active) +} diff --git a/internal/backendpool/pool_test.go b/internal/backendpool/pool_test.go new file mode 100644 index 0000000..d1dba78 --- /dev/null +++ b/internal/backendpool/pool_test.go @@ -0,0 +1,26 @@ +package backendpool + +import "testing" + +func TestPoolSeparatesRuntimeCapacityFromDeliveries(t *testing.T) { + pool := New(2) + if !pool.Acquire("one") || !pool.Acquire("two") || pool.Acquire("three") { + t.Fatal("capacity was not enforced") + } + if !pool.Acquire("one") { + t.Fatal("redelivery must be idempotent") + } + pool.Release("one") + if !pool.Acquire("three") || pool.Active() != 2 { + t.Fatalf("active=%d", pool.Active()) + } +} + +func TestRestoreMayTemporarilyExceedReducedCapacity(t *testing.T) { + pool := New(1) + pool.Restore("one") + pool.Restore("two") + if pool.Active() != 2 || pool.Acquire("three") { + t.Fatalf("active=%d", pool.Active()) + } +} diff --git a/internal/taskscheduler/poller.go b/internal/taskscheduler/poller.go index b03faaf..be8866b 100644 --- a/internal/taskscheduler/poller.go +++ b/internal/taskscheduler/poller.go @@ -4,8 +4,11 @@ import ( "context" "errors" "fmt" + "sync/atomic" "time" + "golang.org/x/sync/errgroup" + runnerv1 "gitea.dev/actionslib/runner/v1" ) @@ -19,10 +22,12 @@ type PollerConfig struct { Labels []string EmptyBackoff time.Duration ErrorBackoff time.Duration + Capacity int } -// Poller is the scheduler component. Once Gitea assigns a task, it never -// fetches another one until the current assignment is durably dispatched. +// Poller is the scheduler component. Each fetcher keeps its assigned task +// until that assignment is durably dispatched; all fetchers share one runner +// declaration and a monotonic tasks version. type Poller struct { Client PollClient Scheduler *Scheduler @@ -49,9 +54,21 @@ func (p Poller) Run(ctx context.Context) error { if errorBackoff <= 0 { errorBackoff = 5 * time.Second } - var tasksVersion int64 + capacity := p.Config.Capacity + if capacity < 1 { + capacity = 1 + } + var tasksVersion atomic.Int64 + group, groupContext := errgroup.WithContext(ctx) + for range capacity { + group.Go(func() error { return p.runFetcher(groupContext, &tasksVersion, emptyBackoff, errorBackoff) }) + } + return group.Wait() +} + +func (p Poller) runFetcher(ctx context.Context, tasksVersion *atomic.Int64, emptyBackoff, errorBackoff time.Duration) error { for { - response, err := p.Client.FetchTask(ctx, tasksVersion) + response, err := p.Client.FetchTask(ctx, tasksVersion.Load()) if err != nil { if ctx.Err() != nil { return nil @@ -69,7 +86,7 @@ func (p Poller) Run(ctx context.Context) error { } continue } - tasksVersion = response.GetTasksVersion() + storeMaximum(tasksVersion, response.GetTasksVersion()) task := response.GetTask() if task == nil { if !wait(ctx, emptyBackoff) { @@ -93,6 +110,14 @@ func (p Poller) Run(ctx context.Context) error { } } +func storeMaximum(value *atomic.Int64, candidate int64) { + for current := value.Load(); candidate > current; current = value.Load() { + if value.CompareAndSwap(current, candidate) { + return + } + } +} + func (p Poller) report(err error) { if p.OnError != nil { p.OnError(err) diff --git a/internal/taskscheduler/poller_test.go b/internal/taskscheduler/poller_test.go index dbc0de1..24bddfb 100644 --- a/internal/taskscheduler/poller_test.go +++ b/internal/taskscheduler/poller_test.go @@ -115,3 +115,52 @@ func TestPollerRetriesAssignedTaskBeforeFetchingAnother(t *testing.T) { t.Fatalf("dispatches=%d fetches-before-dispatch=%d declares=%d", dispatcher.calls, dispatcher.fetchesAtSuccess, client.declared) } } + +type blockingPollClient struct { + mu sync.Mutex + declared int + started chan struct{} +} + +func (c *blockingPollClient) Declare(context.Context, string, []string) error { + c.mu.Lock() + c.declared++ + c.mu.Unlock() + return nil +} + +func (c *blockingPollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) { + c.started <- struct{}{} + <-ctx.Done() + return nil, ctx.Err() +} + +func TestPollerStartsConfiguredNumberOfFetchersAfterOneDeclare(t *testing.T) { + client := &blockingPollClient{started: make(chan struct{}, 3)} + poller := Poller{ + Client: client, + Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: &retryDispatcher{done: make(chan struct{})}}, + Config: PollerConfig{ + Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"}, Capacity: 3, + }, + } + ctx, cancel := context.WithCancel(context.Background()) + finished := make(chan error, 1) + go func() { finished <- poller.Run(ctx) }() + for range 3 { + select { + case <-client.started: + case <-time.After(time.Second): + t.Fatal("configured fetchers did not start") + } + } + cancel() + if err := <-finished; err != nil { + t.Fatal(err) + } + client.mu.Lock() + defer client.mu.Unlock() + if client.declared != 1 { + t.Fatalf("declares = %d", client.declared) + } +}