diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go index 51f372d..8b243f0 100644 --- a/cmd/gitea-dynamic-runner/controller.go +++ b/cmd/gitea-dynamic-runner/controller.go @@ -158,6 +158,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 +187,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 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) {