diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go index 51f372d..ff74ff4 100644 --- a/cmd/gitea-dynamic-runner/controller.go +++ b/cmd/gitea-dynamic-runner/controller.go @@ -15,6 +15,11 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "golang.org/x/sync/errgroup" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/leaderelection" + "k8s.io/client-go/tools/leaderelection/resourcelock" "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue" "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller" @@ -86,7 +91,6 @@ func runController(ctx context.Context) error { return err } registry := runnerfacade.NewRegistry() - gate := taskscheduler.NewSingleFlightGate() var podExecutorBackend *podbackend.Backend var vmExecutorBackend *opensandboxbackend.Backend giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken) @@ -109,7 +113,6 @@ func runController(ctx context.Context) error { return err } } - gate.Release() return nil }, } @@ -126,13 +129,25 @@ func runController(ctx context.Context) error { labels = append(labels, string(taskassignment.BackendVM)) } poller := taskscheduler.Poller{ - Client: giteaClient, Gate: gate, + Client: giteaClient, 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}, OnError: func(err error) { log.Printf("scheduler: %v", err) }, } + kubernetesConfig, err := rest.InClusterConfig() + if err != nil { + return fmt.Errorf("load leader election Kubernetes config: %w", err) + } + kubernetesClient, err := kubernetes.NewForConfig(kubernetesConfig) + if err != nil { + return fmt.Errorf("create leader election Kubernetes client: %w", err) + } + leaderIdentity := strings.TrimSpace(os.Getenv("HOSTNAME")) + if leaderIdentity == "" { + return errors.New("HOSTNAME is required for scheduler leader election") + } facadeServer := runnerfacade.Server{ Facade: facade, ListenAddress: config.FacadeListen, TrustDomain: config.TrustDomain, WorkloadAPIAddr: config.WorkloadAPIAddr, UpstreamURL: config.GiteaURL, @@ -141,7 +156,9 @@ func runController(ctx context.Context) error { 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) }) + group.Go(func() error { + return runSchedulerLeader(groupContext, kubernetesClient, config.PodNamespace, leaderIdentity, poller.Run) + }) return group.Wait() }), } @@ -158,6 +175,15 @@ func runController(ctx context.Context) error { SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID, }} podExecutorBackend = &backend + assignments, err := backend.RecoverAssignments(ctx) + if err != nil { + return err + } + for _, assignment := range assignments { + 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) if err != nil { return err @@ -178,6 +204,15 @@ func runController(ctx context.Context) error { Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr}, }} vmExecutorBackend = &backend + assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain) + if err != nil { + return err + } + for _, assignment := range assignments { + 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) if err != nil { return err @@ -193,6 +228,50 @@ func runController(ctx context.Context) error { return controller.Run(ctx, config.Components, components) } +func runSchedulerLeader(ctx context.Context, client kubernetes.Interface, namespace, identity string, run func(context.Context) error) error { + if client == nil || namespace == "" || identity == "" || run == nil { + return errors.New("leader election client, namespace, identity, and scheduler are required") + } + electionContext, cancel := context.WithCancel(ctx) + defer cancel() + result := make(chan error, 1) + lock := &resourcelock.LeaseLock{ + LeaseMeta: metav1.ObjectMeta{Name: "dynamic-runner-scheduler", Namespace: namespace}, + Client: client.CoordinationV1(), + LockConfig: resourcelock.ResourceLockConfig{ + Identity: identity, + }, + } + elector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{ + Lock: lock, LeaseDuration: 15 * time.Second, RenewDeadline: 10 * time.Second, RetryPeriod: 2 * time.Second, + ReleaseOnCancel: true, + Callbacks: leaderelection.LeaderCallbacks{ + OnStartedLeading: func(leaderContext context.Context) { + result <- run(leaderContext) + cancel() + }, + OnStoppedLeading: func() { + if ctx.Err() == nil { + select { + case result <- errors.New("scheduler leadership lost"): + default: + } + } + }, + }, + }) + if err != nil { + return fmt.Errorf("configure scheduler leader election: %w", err) + } + go elector.Run(electionContext) + select { + case err := <-result: + return err + case <-ctx.Done(): + return nil + } +} + func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn, error) { options := []nats.Option{nats.Name(clientName), nats.UserInfo(user, password)} if caFile != "" { diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 01dc6a2..27da1bf 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -80,9 +80,9 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment；无需把完整 task 写入 Pod annotation、OpenSandbox metadata 或环境变量。 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。 -- 首次生产 canary 使用 controller 进程内 single-flight gate：只有官方 runner 的终态 - `UpdateTask` 已被 Gitea 接受后才允许 Fetch 下一条任务。它把未知故障收敛为停止领取， - 而不是在 backlog 下连续创建 executor；后续容量调度必须以 backend 实际运行资源为准。 +- scheduler 在 assignment 持久化到 JetStream 后即可继续领取；Pod 与 VM 分别由 durable + consumer 的 capacity 限制并发，不共享全局执行槽位。未知后端故障由对应 consumer 的 + NAK/redelivery 收敛，不能阻塞另一种 backend。 - 两种 backend 都注入同一份 runner bootstrap 环境；Pod 仍由 homelab Kubernetes 原生 创建，只有 VM 经 OpenSandbox 创建，bootstrap 机制不改变 backend 边界。 diff --git a/go.mod b/go.mod index aa2b32b..941ad92 100644 --- a/go.mod +++ b/go.mod @@ -45,6 +45,7 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect + github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/sirupsen/logrus v1.10.2 // indirect github.com/x448/float16 v0.8.4 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect diff --git a/go.sum b/go.sum index 62546f1..e61ea34 100644 --- a/go.sum +++ b/go.sum @@ -112,6 +112,8 @@ go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2W go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= diff --git a/internal/opensandboxbackend/backend.go b/internal/opensandboxbackend/backend.go index 7976b6c..078ddec 100644 --- a/internal/opensandboxbackend/backend.go +++ b/internal/opensandboxbackend/backend.go @@ -108,6 +108,30 @@ type Backend struct { Config Config } +func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]taskassignment.Assignment, error) { + if err := b.validate(); err != nil { + return nil, err + } + result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{ + Metadata: map[string]string{"ci.ddupan.top/backend": "vm"}, PageSize: 100, + }) + if err != nil { + return nil, fmt.Errorf("list recoverable sandboxes: %w", err) + } + assignments := make([]taskassignment.Assignment, 0, len(result.Items)) + for _, sandbox := range result.Items { + if sandbox.Metadata[terminalMetadata] == "true" { + continue + } + assignment, err := taskassignment.FromMetadata(sandbox.Metadata, sandbox.Metadata, trustDomain) + if err != nil { + return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err) + } + assignments = append(assignments, assignment) + } + return assignments, nil +} + func NewLifecycleClient(baseURL, apiKey string, client *http.Client) *opensandbox.LifecycleClient { if client != nil { return opensandbox.NewLifecycleClient(baseURL, apiKey, opensandbox.WithHTTPClient(client)) diff --git a/internal/podbackend/backend.go b/internal/podbackend/backend.go index 265f9c6..8f24039 100644 --- a/internal/podbackend/backend.go +++ b/internal/podbackend/backend.go @@ -116,6 +116,28 @@ type Backend struct { Config Config } +func (b Backend) RecoverAssignments(ctx context.Context) ([]taskassignment.Assignment, error) { + if err := b.validate(); err != nil { + return nil, err + } + pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/backend=pod") + if err != nil { + return nil, fmt.Errorf("list recoverable assignment Pods: %w", err) + } + assignments := make([]taskassignment.Assignment, 0, len(pods)) + for _, pod := range pods { + if pod.Labels[terminalLabel] == "true" { + continue + } + assignment, err := taskassignment.FromMetadata(pod.Labels, pod.Annotations, b.Config.TrustDomain) + if err != nil { + return nil, fmt.Errorf("recover Pod %s: %w", pod.Name, err) + } + assignments = append(assignments, assignment) + } + return assignments, nil +} + func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) { if err := b.validate(); err != nil { return nil, err diff --git a/internal/runnerfacade/facade_test.go b/internal/runnerfacade/facade_test.go index 6300842..202bee9 100644 --- a/internal/runnerfacade/facade_test.go +++ b/internal/runnerfacade/facade_test.go @@ -225,3 +225,15 @@ func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) { t.Fatal("capability scope is invalid") } } + +func TestRegistryRecoversClaimedAssignment(t *testing.T) { + registry := NewRegistry() + assignment := facadeAssignment(t) + if err := registry.RecoverClaimed(assignment); err != nil { + t.Fatal(err) + } + resolved, err := registry.Resolve(assignment.ID, assignment.Identity.SPIFFEID) + if err != nil || resolved.Task.GetId() != assignment.Task.GetId() { + t.Fatalf("resolved=%#v err=%v", resolved, err) + } +} diff --git a/internal/runnerfacade/registry.go b/internal/runnerfacade/registry.go index ff00650..27cbd89 100644 --- a/internal/runnerfacade/registry.go +++ b/internal/runnerfacade/registry.go @@ -90,3 +90,21 @@ func (r *Registry) Remove(assignmentID string) { defer r.mu.Unlock() delete(r.claims, assignmentID) } + +// RecoverClaimed restores authorization for an executor that already claimed +// its task before the controller restarted. +func (r *Registry) RecoverClaimed(assignment taskassignment.Assignment) error { + ready, err := r.Offer(assignment) + if err != nil { + return err + } + r.mu.Lock() + entry := r.claims[assignment.ID] + if !entry.claimed { + entry.claimed = true + close(entry.ready) + } + r.mu.Unlock() + <-ready + return nil +} diff --git a/internal/taskassignment/assignment.go b/internal/taskassignment/assignment.go index 6421774..c69cbea 100644 --- a/internal/taskassignment/assignment.go +++ b/internal/taskassignment/assignment.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "slices" + "strconv" "gitea.dev/actionslib/pkg/model" runnerv1 "gitea.dev/actionslib/runner/v1" @@ -32,6 +33,32 @@ type Assignment struct { Identity taskidentity.Identity } +// FromMetadata reconstructs the minimal assignment needed to authorize an +// already-running executor after a controller restart. Backend 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") + } + id := labels["ci.ddupan.top/assignment-id"] + if id != fmt.Sprintf("gitea-task-%d", taskID) { + return Assignment{}, errors.New("backend metadata assignment ID does not match task ID") + } + identity, err := taskidentity.FromMetadata( + annotations["ci.ddupan.top/repository"], annotations["ci.ddupan.top/job-key"], + annotations["ci.ddupan.top/spiffe-id"], trustDomain, + ) + if err != nil { + return Assignment{}, err + } + return Assignment{ID: id, Backend: backend, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil +} + type envelope struct { Version int `json:"version"` ID string `json:"id"` diff --git a/internal/taskassignment/assignment_test.go b/internal/taskassignment/assignment_test.go index 0646e30..cce2ba0 100644 --- a/internal/taskassignment/assignment_test.go +++ b/internal/taskassignment/assignment_test.go @@ -73,3 +73,18 @@ func TestAssignmentWireRoundTripAndValidation(t *testing.T) { t.Fatal("expected tampered backend 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", + }, 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 { + t.Fatalf("assignment=%#v err=%v", assignment, err) + } +} diff --git a/internal/taskidentity/identity.go b/internal/taskidentity/identity.go index 3e991ac..fe9e28c 100644 --- a/internal/taskidentity/identity.go +++ b/internal/taskidentity/identity.go @@ -23,6 +23,21 @@ type Identity struct { SPIFFEID string } +// FromMetadata validates identity fields recovered from backend-owned state. +func FromMetadata(repository, task, spiffeID, trustDomain string) (Identity, error) { + parts := strings.Split(repository, "/") + if len(parts) != 2 || parts[0] == "" || parts[1] == "" || !safeTaskKey.MatchString(task) { + return Identity{}, errors.New("invalid recovered repository or task identity") + } + expected := "spiffe://" + trustDomain + "/ci/" + strings.Join([]string{ + sanitize(parts[0]), sanitize(parts[1]), task, + }, "/") + if spiffeID != expected { + return Identity{}, fmt.Errorf("recovered SPIFFE ID %q does not match %q", spiffeID, expected) + } + return Identity{Repository: repository, Task: task, SPIFFEID: spiffeID}, nil +} + // FromTask derives the repository/task SPIFFE ID from Gitea's trusted task // context. Workflow input never supplies or overrides the resulting ID. func FromTask(task *runnerv1.Task, trustDomain string) (Identity, error) { diff --git a/internal/taskscheduler/gate.go b/internal/taskscheduler/gate.go deleted file mode 100644 index b88c2d7..0000000 --- a/internal/taskscheduler/gate.go +++ /dev/null @@ -1,31 +0,0 @@ -package taskscheduler - -import "context" - -// SingleFlightGate keeps at most one fetched task in flight. Release is -// idempotent so repeated terminal updates cannot increase capacity. -type SingleFlightGate struct { - token chan struct{} -} - -func NewSingleFlightGate() *SingleFlightGate { - gate := &SingleFlightGate{token: make(chan struct{}, 1)} - gate.token <- struct{}{} - return gate -} - -func (g *SingleFlightGate) Acquire(ctx context.Context) error { - select { - case <-g.token: - return nil - case <-ctx.Done(): - return ctx.Err() - } -} - -func (g *SingleFlightGate) Release() { - select { - case g.token <- struct{}{}: - default: - } -} diff --git a/internal/taskscheduler/gate_test.go b/internal/taskscheduler/gate_test.go deleted file mode 100644 index 25815d0..0000000 --- a/internal/taskscheduler/gate_test.go +++ /dev/null @@ -1,29 +0,0 @@ -package taskscheduler - -import ( - "context" - "testing" - "time" -) - -func TestSingleFlightGateBlocksUntilTerminalRelease(t *testing.T) { - gate := NewSingleFlightGate() - if err := gate.Acquire(context.Background()); err != nil { - t.Fatal(err) - } - blocked, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) - defer cancel() - if err := gate.Acquire(blocked); err == nil { - t.Fatal("second task acquired capacity before release") - } - gate.Release() - gate.Release() - if err := gate.Acquire(context.Background()); err != nil { - t.Fatal(err) - } - blockedAgain, cancelAgain := context.WithTimeout(context.Background(), 20*time.Millisecond) - defer cancelAgain() - if err := gate.Acquire(blockedAgain); err == nil { - t.Fatal("duplicate terminal release increased capacity") - } -} diff --git a/internal/taskscheduler/poller.go b/internal/taskscheduler/poller.go index 6447e73..b03faaf 100644 --- a/internal/taskscheduler/poller.go +++ b/internal/taskscheduler/poller.go @@ -26,7 +26,6 @@ type PollerConfig struct { type Poller struct { Client PollClient Scheduler *Scheduler - Gate *SingleFlightGate Config PollerConfig OnError func(error) } @@ -51,14 +50,7 @@ func (p Poller) Run(ctx context.Context) error { errorBackoff = 5 * time.Second } var tasksVersion int64 - haveLease := false for { - if p.Gate != nil && !haveLease { - if err := p.Gate.Acquire(ctx); err != nil { - return nil - } - haveLease = true - } response, err := p.Client.FetchTask(ctx, tasksVersion) if err != nil { if ctx.Err() != nil { @@ -80,10 +72,6 @@ func (p Poller) Run(ctx context.Context) error { tasksVersion = response.GetTasksVersion() task := response.GetTask() if task == nil { - if p.Gate != nil { - p.Gate.Release() - haveLease = false - } if !wait(ctx, emptyBackoff) { return nil } @@ -91,7 +79,6 @@ func (p Poller) Run(ctx context.Context) error { } for { if err := p.Scheduler.Run(ctx, task); err == nil { - haveLease = false break } else { if ctx.Err() != nil {