diff --git a/docs/design-principles.md b/docs/design-principles.md index 3c39cae..32f9040 100644 --- a/docs/design-principles.md +++ b/docs/design-principles.md @@ -93,5 +93,5 @@ OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意 基础设施协议优先使用上游维护的成熟客户端，不在 controller 内重复实现认证、连接、 资源编码或错误语义。Kubernetes 使用 `client-go`，NATS JetStream 使用 `nats.go`， -Gitea RunnerService 使用 `actionslib`；自定义代码只保留领域模型、reconcile 规则及 -上游客户端未覆盖的最小适配层。 +Gitea RunnerService 使用 `actionslib`，OpenSandbox Lifecycle API 使用官方 Go SDK； +自定义代码只保留领域模型、reconcile 规则及上游客户端未覆盖的最小适配层。 diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 0a90051..23285d9 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -43,6 +43,8 @@ dynamic-runner scheduler 不承担 executor 生命周期记录。 - worker 按稳定 assignment ID reconcile 后端资源，进程内只保留并发控制等可丢弃状态； 不新增数据库，也不依赖内存中的 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 的崩溃窗口。 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。 diff --git a/go.mod b/go.mod index 737922b..f236b3b 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.27 require ( connectrpc.com/connect v1.20.0 gitea.dev/actionslib v1.0.0 + github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5 github.com/nats-io/nats.go v1.54.0 google.golang.org/protobuf v1.36.12 k8s.io/api v0.37.0 @@ -48,6 +49,7 @@ require ( golang.org/x/crypto v0.57.0 // indirect golang.org/x/net v0.58.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect + golang.org/x/sync v0.23.0 // indirect golang.org/x/sys v0.48.0 // indirect golang.org/x/term v0.46.0 // indirect golang.org/x/text v0.42.0 // indirect diff --git a/go.sum b/go.sum index 9e3dbde..0f54ddc 100644 --- a/go.sum +++ b/go.sum @@ -2,6 +2,8 @@ connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ= connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4= gitea.dev/actionslib v1.0.0 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY= gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g= +github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5 h1:7mZNkBh4VaI+i4VOTJIknLCBkbGizTkdHYqgcMqeDBM= +github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5/go.mod h1:w0nIMCTL1L3oSS67ABFOdYxrezYQZhk01UtmKSi2UhA= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= @@ -98,6 +100,8 @@ golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= +golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= +golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= diff --git a/internal/opensandboxbackend/backend.go b/internal/opensandboxbackend/backend.go new file mode 100644 index 0000000..3b098c5 --- /dev/null +++ b/internal/opensandboxbackend/backend.go @@ -0,0 +1,147 @@ +// Package opensandboxbackend implements the VM executor backend with the official OpenSandbox SDK. +package opensandboxbackend + +import ( + "context" + "errors" + "fmt" + "net/http" + + opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker" +) + +const assignmentMetadata = "ci.ddupan.top/assignment-id" + +type Lifecycle interface { + ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error) + CreateSandbox(context.Context, opensandbox.CreateSandboxRequest) (*opensandbox.SandboxInfo, error) + GetSandbox(context.Context, string) (*opensandbox.SandboxInfo, error) + DeleteSandbox(context.Context, string) error +} + +type Config struct { + Pool string + Timeout int + Entrypoint []string + Env map[string]string +} + +type Backend struct { + Lifecycle Lifecycle + Config Config +} + +func NewLifecycleClient(baseURL, apiKey string, client *http.Client) *opensandbox.LifecycleClient { + if client != nil { + return opensandbox.NewLifecycleClient(baseURL, apiKey, opensandbox.WithHTTPClient(client)) + } + return opensandbox.NewLifecycleClient(baseURL, apiKey) +} + +func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) { + if err := b.validate(); err != nil { + return nil, err + } + result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{ + Metadata: map[string]string{assignmentMetadata: assignmentID}, PageSize: 2, + }) + if err != nil { + return nil, fmt.Errorf("list assignment sandboxes: %w", err) + } + if len(result.Items) > 1 { + return nil, fmt.Errorf("assignment %s owns %d sandboxes", assignmentID, len(result.Items)) + } + if len(result.Items) == 0 { + return nil, nil + } + return executor(result.Items[0]), nil +} + +func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, metadata taskworker.Metadata) (*taskworker.Executor, error) { + 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) + } + environment := clone(b.Config.Env) + environment["CI_ASSIGNMENT_ID"] = assignment.ID + environment["CI_SPIFFE_ID"] = assignment.Identity.SPIFFEID + sandboxMetadata := clone(metadata.Annotations) + for key, value := range metadata.Labels { + sandboxMetadata[key] = value + } + request := opensandbox.CreateSandboxRequest{ + Timeout: &b.Config.Timeout, Entrypoint: append([]string{}, b.Config.Entrypoint...), + Env: environment, Metadata: sandboxMetadata, + Extensions: map[string]string{"poolRef": b.Config.Pool}, + ResourceLimits: opensandbox.ResourceLimits{}, + } + sandbox, err := b.Lifecycle.CreateSandbox(ctx, request) + if err != nil { + return nil, fmt.Errorf("create assignment sandbox: %w", err) + } + return executor(*sandbox), nil +} + +func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor, identity taskidentity.Identity) error { + if executor == nil || executor.Name == "" { + return errors.New("sandbox ID is required for identity binding") + } + sandbox, err := b.Lifecycle.GetSandbox(ctx, executor.Name) + if err != nil { + return fmt.Errorf("verify sandbox identity metadata: %w", err) + } + if sandbox.Metadata["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID { + return fmt.Errorf("sandbox %s has inconsistent SPIFFE identity metadata", executor.Name) + } + return nil +} + +func (b Backend) Delete(ctx context.Context, executor *taskworker.Executor) error { + if executor == nil || executor.Name == "" { + return nil + } + err := b.Lifecycle.DeleteSandbox(ctx, executor.Name) + var apiError *opensandbox.APIError + if errors.As(err, &apiError) && apiError.StatusCode == http.StatusNotFound { + return nil + } + return err +} + +func (b Backend) validate() error { + if b.Lifecycle == nil || b.Config.Pool == "" || b.Config.Timeout < 1 || len(b.Config.Entrypoint) == 0 { + return errors.New("OpenSandbox Lifecycle client, pool, timeout, and entrypoint are required") + } + return nil +} + +func executor(sandbox opensandbox.SandboxInfo) *taskworker.Executor { + return &taskworker.Executor{Name: sandbox.ID, IdentityTarget: sandbox.ID, Phase: phase(sandbox.Status.State)} +} + +func phase(state opensandbox.SandboxState) taskworker.Phase { + switch state { + case opensandbox.StateRunning: + return taskworker.PhaseRunning + case opensandbox.StateTerminated, opensandbox.StateFailed: + // A successful executor reports Gitea before it exits. If Gitea is not + // terminal when the sandbox stops, termination is an execution failure. + return taskworker.PhaseFailed + default: + return taskworker.PhasePending + } +} + +func clone(source map[string]string) map[string]string { + result := make(map[string]string, len(source)) + for key, value := range source { + result[key] = value + } + return result +} diff --git a/internal/opensandboxbackend/backend_test.go b/internal/opensandboxbackend/backend_test.go new file mode 100644 index 0000000..cc879f5 --- /dev/null +++ b/internal/opensandboxbackend/backend_test.go @@ -0,0 +1,79 @@ +package opensandboxbackend + +import ( + "context" + "testing" + + runnerv1 "gitea.dev/actionslib/runner/v1" + opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskworker" +) + +type fakeLifecycle struct { + items []opensandbox.SandboxInfo + created opensandbox.CreateSandboxRequest + deleted string +} + +func (f *fakeLifecycle) ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error) { + return &opensandbox.ListSandboxesResponse{Items: f.items}, nil +} +func (f *fakeLifecycle) CreateSandbox(_ context.Context, request opensandbox.CreateSandboxRequest) (*opensandbox.SandboxInfo, error) { + f.created = request + return &opensandbox.SandboxInfo{ID: "sandbox-42", Status: opensandbox.SandboxStatus{State: opensandbox.StatePending}, Metadata: request.Metadata}, nil +} +func (f *fakeLifecycle) GetSandbox(_ context.Context, id string) (*opensandbox.SandboxInfo, error) { + for _, item := range f.items { + if item.ID == id { + return &item, nil + } + } + return &opensandbox.SandboxInfo{ID: id, Metadata: map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID}}, nil +} +func (f *fakeLifecycle) DeleteSandbox(_ context.Context, id string) error { f.deleted = id; return nil } + +func backend(lifecycle Lifecycle) Backend { + return Backend{Lifecycle: lifecycle, Config: Config{ + Pool: "ci-vm", Timeout: 14400, + Entrypoint: []string{"/usr/local/bin/gitea-task-executor"}, + Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": "unix:///run/spire/agent-sockets/spire-agent.sock"}, + }} +} + +func assignment() taskassignment.Assignment { + return taskassignment.Assignment{ + ID: "gitea-task-42", Backend: taskassignment.BackendVM, + Task: &runnerv1.Task{Id: 42}, + Identity: taskidentity.Identity{Repository: "owner/repo", Task: "publish", SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"}, + } +} + +func TestFindRecoversSandboxByMetadata(t *testing.T) { + lifecycle := &fakeLifecycle{items: []opensandbox.SandboxInfo{{ID: "sandbox-42", Status: opensandbox.SandboxStatus{State: opensandbox.StateRunning}}}} + executor, err := backend(lifecycle).Find(context.Background(), assignment().ID) + if err != nil || executor.Name != "sandbox-42" || executor.Phase != taskworker.PhaseRunning { + t.Fatalf("executor=%#v err=%v", executor, err) + } +} + +func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) { + lifecycle := &fakeLifecycle{} + metadata := taskworker.BackendMetadata(assignment()) + executor, err := backend(lifecycle).Create(context.Background(), assignment(), metadata) + if err != nil { + t.Fatal(err) + } + if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" { + t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor) + } +} + +func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) { + lifecycle := &fakeLifecycle{} + if err := backend(lifecycle).BindIdentity(context.Background(), &taskworker.Executor{Name: "sandbox-42"}, assignment().Identity); err != nil { + t.Fatal(err) + } +}