311 lines
10 KiB
Go
311 lines
10 KiB
Go
// Package opensandboxbackend implements the VM executor backend with the official OpenSandbox SDK.
|
|
package opensandboxbackend
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
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"
|
|
const terminalMetadata = "ci.ddupan.top/terminal"
|
|
const annotationsMetadataPrefix = "ci.ddupan.top/annotations-"
|
|
const metadataValueLimit = 63
|
|
|
|
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)
|
|
PatchSandboxMetadata(context.Context, string, opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error)
|
|
DeleteSandbox(context.Context, string) error
|
|
}
|
|
|
|
// MarkTerminal persists the accepted Gitea terminal state on the sandbox. The
|
|
// lifecycle reconciler performs deletion separately so the runner receives the
|
|
// successful UpdateTask response before its VM is stopped.
|
|
func (b Backend) MarkTerminal(ctx context.Context, assignmentID string) error {
|
|
executor, err := b.Find(ctx, assignmentID)
|
|
if err != nil || executor == nil {
|
|
return err
|
|
}
|
|
value := "true"
|
|
_, err = b.Lifecycle.PatchSandboxMetadata(ctx, executor.Name, opensandbox.MetadataPatch{
|
|
terminalMetadata: &value,
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("mark sandbox %s terminal: %w", executor.Name, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// CleanupTerminated removes sandboxes whose terminal result was accepted by
|
|
// Gitea. The marker is stored by OpenSandbox, so cleanup survives restarts.
|
|
func (b Backend) CleanupTerminated(ctx context.Context) (int, error) {
|
|
if err := b.validate(); err != nil {
|
|
return 0, err
|
|
}
|
|
result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
|
|
Metadata: map[string]string{terminalMetadata: "true"},
|
|
PageSize: 100,
|
|
})
|
|
if err != nil {
|
|
return 0, fmt.Errorf("list terminal sandboxes: %w", err)
|
|
}
|
|
cleaned := 0
|
|
for _, sandbox := range result.Items {
|
|
if sandbox.Metadata[assignmentMetadata] == "" {
|
|
continue
|
|
}
|
|
if err := b.Delete(ctx, executor(sandbox)); err != nil {
|
|
return cleaned, fmt.Errorf("delete terminal sandbox %s: %w", sandbox.ID, err)
|
|
}
|
|
cleaned++
|
|
}
|
|
return cleaned, nil
|
|
}
|
|
|
|
// Lifecycle periodically reconciles durable terminal markers into deletes.
|
|
type LifecycleReconciler struct {
|
|
Backend Backend
|
|
Interval time.Duration
|
|
OnError func(error)
|
|
}
|
|
|
|
func (l LifecycleReconciler) Run(ctx context.Context) error {
|
|
interval := l.Interval
|
|
if interval <= 0 {
|
|
interval = 2 * time.Second
|
|
}
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
if _, err := l.Backend.CleanupTerminated(ctx); err != nil && ctx.Err() == nil && l.OnError != nil {
|
|
l.OnError(err)
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return nil
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
type Config struct {
|
|
Pool string
|
|
Timeout int
|
|
Entrypoint []string
|
|
Env map[string]string
|
|
}
|
|
|
|
type Backend struct {
|
|
Lifecycle Lifecycle
|
|
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
|
|
}
|
|
annotations, err := decodeAnnotations(sandbox.Metadata)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("decode sandbox %s annotations: %w", sandbox.ID, err)
|
|
}
|
|
assignment, err := taskassignment.FromMetadata(sandbox.Metadata, annotations, 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))
|
|
}
|
|
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, launch taskworker.LaunchSpec) (*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)
|
|
for key, value := range launch.Environment {
|
|
environment[key] = value
|
|
}
|
|
sandboxMetadata := clone(launch.Metadata.Labels)
|
|
annotations, err := encodeAnnotations(launch.Metadata.Annotations)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("encode sandbox annotations: %w", err)
|
|
}
|
|
for key, value := range annotations {
|
|
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)
|
|
}
|
|
annotations, err := decodeAnnotations(sandbox.Metadata)
|
|
if err != nil {
|
|
return fmt.Errorf("decode sandbox identity metadata: %w", err)
|
|
}
|
|
if annotations["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
|
|
}
|
|
|
|
// OpenSandbox metadata follows Kubernetes label-value constraints, unlike Pod
|
|
// annotations. Store the annotation map as deterministic URL-safe base64
|
|
// chunks so repository paths and SPIFFE IDs remain lossless and recoverable.
|
|
func encodeAnnotations(annotations map[string]string) (map[string]string, error) {
|
|
data, err := json.Marshal(annotations)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
encoded := base64.RawURLEncoding.EncodeToString(data)
|
|
result := make(map[string]string, (len(encoded)+metadataValueLimit-1)/metadataValueLimit)
|
|
for index := 0; len(encoded) > 0; index++ {
|
|
length := min(metadataValueLimit, len(encoded))
|
|
result[fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index)] = encoded[:length]
|
|
encoded = encoded[length:]
|
|
}
|
|
return result, nil
|
|
}
|
|
|
|
func decodeAnnotations(metadata map[string]string) (map[string]string, error) {
|
|
keys := make([]string, 0)
|
|
for key := range metadata {
|
|
if strings.HasPrefix(key, annotationsMetadataPrefix) {
|
|
keys = append(keys, key)
|
|
}
|
|
}
|
|
if len(keys) == 0 {
|
|
return nil, errors.New("sandbox annotation metadata is missing")
|
|
}
|
|
sort.Strings(keys)
|
|
var encoded strings.Builder
|
|
for index, key := range keys {
|
|
if key != fmt.Sprintf("%s%03d", annotationsMetadataPrefix, index) {
|
|
return nil, errors.New("sandbox annotation metadata chunks are incomplete")
|
|
}
|
|
encoded.WriteString(metadata[key])
|
|
}
|
|
data, err := base64.RawURLEncoding.DecodeString(encoded.String())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var annotations map[string]string
|
|
if err := json.Unmarshal(data, &annotations); err != nil {
|
|
return nil, err
|
|
}
|
|
return annotations, nil
|
|
}
|