接入 OpenSandbox 官方 Go SDK
This commit is contained in:
@@ -93,5 +93,5 @@ OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意
|
|||||||
|
|
||||||
基础设施协议优先使用上游维护的成熟客户端,不在 controller 内重复实现认证、连接、
|
基础设施协议优先使用上游维护的成熟客户端,不在 controller 内重复实现认证、连接、
|
||||||
资源编码或错误语义。Kubernetes 使用 `client-go`,NATS JetStream 使用 `nats.go`,
|
资源编码或错误语义。Kubernetes 使用 `client-go`,NATS JetStream 使用 `nats.go`,
|
||||||
Gitea RunnerService 使用 `actionslib`;自定义代码只保留领域模型、reconcile 规则及
|
Gitea RunnerService 使用 `actionslib`,OpenSandbox Lifecycle API 使用官方 Go SDK;
|
||||||
上游客户端未覆盖的最小适配层。
|
自定义代码只保留领域模型、reconcile 规则及上游客户端未覆盖的最小适配层。
|
||||||
|
|||||||
@@ -43,6 +43,8 @@ dynamic-runner scheduler
|
|||||||
不承担 executor 生命周期记录。
|
不承担 executor 生命周期记录。
|
||||||
- worker 按稳定 assignment ID reconcile 后端资源,进程内只保留并发控制等可丢弃状态;
|
- worker 按稳定 assignment ID reconcile 后端资源,进程内只保留并发控制等可丢弃状态;
|
||||||
不新增数据库,也不依赖内存中的 runner-to-executor 映射。
|
不新增数据库,也不依赖内存中的 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 终态,
|
- 结果处理顺序固定为回报 Gitea、清理后端、ACK assignment;重投时先查询 Gitea 终态,
|
||||||
从而关闭后端已删除但消息尚未 ACK 的崩溃窗口。
|
从而关闭后端已删除但消息尚未 ACK 的崩溃窗口。
|
||||||
- Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。
|
- Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ go 1.27
|
|||||||
require (
|
require (
|
||||||
connectrpc.com/connect v1.20.0
|
connectrpc.com/connect v1.20.0
|
||||||
gitea.dev/actionslib v1.0.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
|
github.com/nats-io/nats.go v1.54.0
|
||||||
google.golang.org/protobuf v1.36.12
|
google.golang.org/protobuf v1.36.12
|
||||||
k8s.io/api v0.37.0
|
k8s.io/api v0.37.0
|
||||||
@@ -48,6 +49,7 @@ require (
|
|||||||
golang.org/x/crypto v0.57.0 // indirect
|
golang.org/x/crypto v0.57.0 // indirect
|
||||||
golang.org/x/net v0.58.0 // indirect
|
golang.org/x/net v0.58.0 // indirect
|
||||||
golang.org/x/oauth2 v0.36.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/sys v0.48.0 // indirect
|
||||||
golang.org/x/term v0.46.0 // indirect
|
golang.org/x/term v0.46.0 // indirect
|
||||||
golang.org/x/text v0.42.0 // indirect
|
golang.org/x/text v0.42.0 // indirect
|
||||||
|
|||||||
@@ -2,6 +2,8 @@ connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ=
|
|||||||
connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4=
|
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 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY=
|
||||||
gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g=
|
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.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.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
|
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/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 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
|
||||||
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
|
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 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||||
golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
|
golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user