From be697a61cf58e7972dcbd41cc5b38beca8604a93 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Mon, 21 Sep 2026 05:24:38 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=87=AA=E5=8A=A8=E6=B8=85=E7=90=86?= =?UTF-8?q?=E7=BB=88=E6=80=81=20Pod=20=E6=89=A7=E8=A1=8C=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 7 +++- cmd/gitea-dynamic-runner/controller.go | 23 ++++++++++++-- docs/runner-protocol-roadmap.md | 5 +-- internal/podbackend/backend.go | 44 +++++++++++++++++++++++++- internal/podbackend/backend_test.go | 26 +++++++++++++++ internal/podbackend/client.go | 11 +++++++ internal/podbackend/client_test.go | 7 ++++ internal/podbackend/lifecycle.go | 32 +++++++++++++++++++ internal/runnerfacade/facade.go | 6 ++-- internal/runnerfacade/facade_test.go | 3 +- 10 files changed, 155 insertions(+), 9 deletions(-) create mode 100644 internal/podbackend/lifecycle.go diff --git a/README.md b/README.md index afa813e..6ef3332 100644 --- a/README.md +++ b/README.md @@ -65,11 +65,16 @@ credential 都从挂载文件读取，不接受明文环境变量： - `RUNNER_FACADE_CAPABILITY_KEY_FILE`：至少 32 字节的 controller HMAC key。 - `OPENSANDBOX_API_KEY_FILE`：仅启用 `vm-worker` 时读取。 -必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`（应使用 digest）、 +必要的非 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 容量。 +Pod task 的 terminal update 被 Gitea 接受后，controller 会在 Pod 上持久写入 +`ci.ddupan.top/terminal=true` label。生命周期 reconciler 只清理同时带该 label 且已经 +进入 `Succeeded` 或 `Failed` phase 的 Pod 及其同名 `ClusterStaticEntry`；controller +重启不影响清理恢复，上报终态前失败的 Pod 也不会被误删。 + ## 安全边界 - OpenSandbox API key、webhook secret 和 Gitea registration token 只从文件读取。 diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go index ed85907..6801004 100644 --- a/cmd/gitea-dynamic-runner/controller.go +++ b/cmd/gitea-dynamic-runner/controller.go @@ -87,10 +87,22 @@ func runController(ctx context.Context) error { } registry := runnerfacade.NewRegistry() gate := taskscheduler.NewSingleFlightGate() + var podExecutorBackend *podbackend.Backend giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken) facade := &runnerfacade.Facade{ Registry: registry, Capabilities: capabilities, Upstream: giteaClient, - OnTerminal: func(taskassignment.Assignment) { gate.Release() }, + OnTerminal: func(ctx context.Context, assignment taskassignment.Assignment) error { + if assignment.Backend == 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 + } + } + gate.Release() + return nil + }, } bootstrap := runnerbootstrap.Bootstrap{ Capabilities: capabilities, FacadeURL: config.FacadeURL, FacadeSPIFFEID: config.FacadeSPIFFEID, @@ -136,11 +148,18 @@ func runController(ctx context.Context) error { SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass, SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID, }} + podExecutorBackend = &backend component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry) if err != nil { return err } - components[controller.PodWorker] = component + lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }} + components[controller.PodWorker] = 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) { lifecycle := opensandboxbackend.NewLifecycleClient(config.OpenSandboxURL, config.OpenSandboxAPIKey, &http.Client{Timeout: 60 * time.Second}) diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index f8c0ed4..01dc6a2 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -69,8 +69,9 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把 不新增数据库，也不依赖内存中的 runner-to-executor 映射。 - VM worker 使用 OpenSandbox 官方 Go SDK，并把 assignment ID、repository、job key 和 SPIFFE ID写入 sandbox metadata；通过 `extensions.poolRef=ci-vm` 使用既有 Kata Pool。 -- 结果处理顺序固定为回报 Gitea、清理后端、ACK assignment；重投时先查询 Gitea 终态， - 从而关闭后端已删除但消息尚未 ACK 的崩溃窗口。 +- executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后，facade 在后端 + metadata 写入持久 terminal marker；backend reconciler 仅在执行环境也进入终态后清理， + 从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。 - pod 与 vm 使用独立 durable consumer 和并发上限。consumer 只负责将 assignment 幂等落到后端；executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得 Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。 diff --git a/internal/podbackend/backend.go b/internal/podbackend/backend.go index fb827ef..265f9c6 100644 --- a/internal/podbackend/backend.go +++ b/internal/podbackend/backend.go @@ -13,7 +13,10 @@ import ( "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker" ) -const assignmentLabel = "ci.ddupan.top/assignment-id" +const ( + assignmentLabel = "ci.ddupan.top/assignment-id" + terminalLabel = "ci.ddupan.top/terminal" +) // Pod is the backend state required by the reconciler, not an in-memory lifecycle record. type Pod struct { @@ -53,10 +56,49 @@ type API interface { ListPods(context.Context, string, string) ([]Pod, error) CreatePod(context.Context, PodManifest) (Pod, error) DeletePod(context.Context, string, string) error + LabelPod(context.Context, string, string, map[string]string) error EnsureIdentityEntry(context.Context, IdentityEntry) error DeleteIdentityEntry(context.Context, string) error } +// MarkTerminal persists Gitea's accepted terminal state on the backend +// resource. Cleanup can therefore resume after a controller restart. +func (b Backend) MarkTerminal(ctx context.Context, assignmentID string) error { + if err := b.validate(); err != nil { + return err + } + if assignmentID == "" { + return errors.New("assignment ID is required") + } + if err := b.API.LabelPod(ctx, b.Config.Namespace, assignmentID, map[string]string{terminalLabel: "true"}); err != nil { + return fmt.Errorf("mark assignment Pod terminal: %w", err) + } + return nil +} + +// CleanupTerminated removes only executors whose terminal update was accepted +// by Gitea and whose process has exited. +func (b Backend) CleanupTerminated(ctx context.Context) (int, error) { + if err := b.validate(); err != nil { + return 0, err + } + pods, err := b.API.ListPods(ctx, b.Config.Namespace, terminalLabel+"=true") + if err != nil { + return 0, fmt.Errorf("list terminal assignment Pods: %w", err) + } + cleaned := 0 + for _, pod := range pods { + if pod.Phase != "Succeeded" && pod.Phase != "Failed" { + continue + } + if err := b.Delete(ctx, executor(pod)); err != nil { + return cleaned, err + } + cleaned++ + } + return cleaned, nil +} + type Config struct { Namespace string Image string diff --git a/internal/podbackend/backend_test.go b/internal/podbackend/backend_test.go index e8ad748..e1d96ba 100644 --- a/internal/podbackend/backend_test.go +++ b/internal/podbackend/backend_test.go @@ -18,6 +18,7 @@ type fakeAPI struct { entry IdentityEntry entryGone string deleted string + marked map[string]string } func (a *fakeAPI) ListPods(_ context.Context, _ string, selector string) ([]Pod, error) { @@ -32,6 +33,10 @@ func (a *fakeAPI) DeletePod(_ context.Context, _, name string) error { a.deleted = name return nil } +func (a *fakeAPI) LabelPod(_ context.Context, _, _ string, labels map[string]string) error { + a.marked = labels + return nil +} func (a *fakeAPI) EnsureIdentityEntry(_ context.Context, entry IdentityEntry) error { a.entry = entry return nil @@ -119,6 +124,27 @@ func TestDeleteRemovesIdentityBeforePod(t *testing.T) { } } +func TestTerminalMarkerAndCleanupUseBackendState(t *testing.T) { + api := &fakeAPI{pods: []Pod{ + {Name: "running", UID: "running-uid", Phase: "Running"}, + {Name: "finished", UID: "finished-uid", Phase: "Succeeded"}, + }} + backend := backend(api) + if err := backend.MarkTerminal(context.Background(), "gitea-task-42"); err != nil { + t.Fatal(err) + } + if api.marked[terminalLabel] != "true" { + t.Fatalf("labels = %#v", api.marked) + } + cleaned, err := backend.CleanupTerminated(context.Background()) + if err != nil { + t.Fatal(err) + } + if api.selector != terminalLabel+"=true" || cleaned != 1 || api.deleted != "finished" || api.entryGone != "finished" { + t.Fatalf("selector=%q cleaned=%d deleted=%q entry=%q", api.selector, cleaned, api.deleted, api.entryGone) + } +} + func TestFindRejectsDuplicatePods(t *testing.T) { api := &fakeAPI{pods: []Pod{{Name: "one"}, {Name: "two"}}} if _, err := backend(api).Find(context.Background(), "gitea-task-42"); err == nil { diff --git a/internal/podbackend/client.go b/internal/podbackend/client.go index fe1d46b..f3926a8 100644 --- a/internal/podbackend/client.go +++ b/internal/podbackend/client.go @@ -2,6 +2,7 @@ package podbackend import ( "context" + "encoding/json" "fmt" "reflect" @@ -10,6 +11,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/dynamic" "k8s.io/client-go/kubernetes" "k8s.io/client-go/rest" @@ -102,6 +104,15 @@ func (c *Client) DeletePod(ctx context.Context, namespace, name string) error { return err } +func (c *Client) LabelPod(ctx context.Context, namespace, name string, labels map[string]string) error { + patch, err := json.Marshal(map[string]any{"metadata": map[string]any{"labels": labels}}) + if err != nil { + return err + } + _, err = c.Kubernetes.CoreV1().Pods(namespace).Patch(ctx, name, types.MergePatchType, patch, metav1.PatchOptions{}) + return err +} + func (c *Client) EnsureIdentityEntry(ctx context.Context, entry IdentityEntry) error { resource := c.Dynamic.Resource(identityEntryResource) existing, err := resource.Get(ctx, entry.Name, metav1.GetOptions{}) diff --git a/internal/podbackend/client_test.go b/internal/podbackend/client_test.go index 36ea21d..2cdfb17 100644 --- a/internal/podbackend/client_test.go +++ b/internal/podbackend/client_test.go @@ -37,6 +37,13 @@ func TestClientPodLifecycleUsesTypedClient(t *testing.T) { if got := pod.Spec.Containers[0].Env; len(got) != 1 || got[0].Name != "CI_RUNNER_CAPABILITY" || got[0].Value != "capability" { t.Fatalf("environment = %#v", got) } + if err := client.LabelPod(context.Background(), "gitea-actions", created.Name, map[string]string{terminalLabel: "true"}); err != nil { + t.Fatal(err) + } + pod, err = client.Kubernetes.CoreV1().Pods("gitea-actions").Get(context.Background(), created.Name, metav1.GetOptions{}) + if err != nil || pod.Labels[terminalLabel] != "true" { + t.Fatalf("terminal label pod=%#v err=%v", pod, err) + } pod.UID = types.UID("pod-uid") pod.Status.Phase = corev1.PodRunning if _, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Update(context.Background(), pod, metav1.UpdateOptions{}); err != nil { diff --git a/internal/podbackend/lifecycle.go b/internal/podbackend/lifecycle.go new file mode 100644 index 0000000..b06e34a --- /dev/null +++ b/internal/podbackend/lifecycle.go @@ -0,0 +1,32 @@ +package podbackend + +import ( + "context" + "time" +) + +// Lifecycle reconciles durable terminal markers into backend cleanup. +type Lifecycle struct { + Backend Backend + Interval time.Duration + OnError func(error) +} + +func (l Lifecycle) Run(ctx context.Context) error { + interval := l.Interval + if interval <= 0 { + interval = 2 * time.Second + } + for { + if _, err := l.Backend.CleanupTerminated(ctx); err != nil && ctx.Err() == nil && l.OnError != nil { + l.OnError(err) + } + timer := time.NewTimer(interval) + select { + case <-ctx.Done(): + timer.Stop() + return nil + case <-timer.C: + } + } +} diff --git a/internal/runnerfacade/facade.go b/internal/runnerfacade/facade.go index b713487..bbff68b 100644 --- a/internal/runnerfacade/facade.go +++ b/internal/runnerfacade/facade.go @@ -57,7 +57,7 @@ type Facade struct { Registry *Registry Capabilities Capabilities Upstream Upstream - OnTerminal func(taskassignment.Assignment) + OnTerminal func(context.Context, taskassignment.Assignment) error } func (f *Facade) Handler() (string, http.Handler) { @@ -101,7 +101,9 @@ func (f *Facade) UpdateTask(ctx context.Context, request *connect.Request[runner } response, err := f.Upstream.UpdateTask(ctx, connect.NewRequest(request.Msg)) if err == nil && request.Msg.GetState().GetResult() != runnerv1.Result_RESULT_UNSPECIFIED && f.OnTerminal != nil { - f.OnTerminal(assignment) + if terminalErr := f.OnTerminal(ctx, assignment); terminalErr != nil { + return nil, connect.NewError(connect.CodeUnavailable, terminalErr) + } } return response, err } diff --git a/internal/runnerfacade/facade_test.go b/internal/runnerfacade/facade_test.go index dca6821..73fe3c4 100644 --- a/internal/runnerfacade/facade_test.go +++ b/internal/runnerfacade/facade_test.go @@ -170,11 +170,12 @@ func TestFacadeSignalsTerminalTaskAfterUpstreamAcceptsIt(t *testing.T) { t.Fatal(err) } completed := 0 - facade.OnTerminal = func(got taskassignment.Assignment) { + facade.OnTerminal = func(_ context.Context, got taskassignment.Assignment) error { if got.ID != assignment.ID { t.Fatalf("terminal assignment = %s", got.ID) } completed++ + return nil } if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{ State: &runnerv1.TaskState{Id: 42, Result: runnerv1.Result_RESULT_SUCCESS},