实现 client-go Pod backend

This commit is contained in:
2026-09-20 18:29:26 +00:00
parent 0d8b36380d
commit fccaf319ac
7 changed files with 728 additions and 12 deletions
+202
View File
@@ -0,0 +1,202 @@
// Package podbackend implements the native homelab Kubernetes executor backend.
package podbackend
import (
"context"
"errors"
"fmt"
"net/url"
"strings"
"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 assignmentLabel = "ci.ddupan.top/assignment-id"
// Pod is the backend state required by the reconciler, not an in-memory lifecycle record.
type Pod struct {
Name string
Namespace string
UID string
Phase string
Labels map[string]string
Annotations map[string]string
}
// PodManifest leaves task execution wiring to the executor image while fixing
// the metadata contract needed for recovery.
type PodManifest struct {
Name string
Namespace string
Labels map[string]string
Annotations map[string]string
Image string
ServiceAccount string
Args []string
}
// IdentityEntry is a ClusterStaticEntry pinned to one concrete Pod UID.
type IdentityEntry struct {
Name string
Labels map[string]string
ClassName string
ParentID string
SPIFFEID string
Selectors []string
}
// API is the narrow Kubernetes boundary used by the backend adapter.
type API interface {
ListPods(context.Context, string, string) ([]Pod, error)
CreatePod(context.Context, PodManifest) (Pod, error)
DeletePod(context.Context, string, string) error
EnsureIdentityEntry(context.Context, IdentityEntry) error
DeleteIdentityEntry(context.Context, string) error
}
type Config struct {
Namespace string
Image string
ServiceAccount string
ExecutorArgs []string
TrustDomain string
SPIRECluster string
SPIREClass string
ExecutorUID int
}
type Backend struct {
API API
Config Config
}
func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) {
if err := b.validate(); err != nil {
return nil, err
}
pods, err := b.API.ListPods(ctx, b.Config.Namespace, assignmentLabel+"="+assignmentID)
if err != nil {
return nil, fmt.Errorf("list assignment Pods: %w", err)
}
if len(pods) > 1 {
return nil, fmt.Errorf("assignment %s owns %d Pods", assignmentID, len(pods))
}
if len(pods) == 0 {
return nil, nil
}
return executor(pods[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.BackendPod {
return nil, fmt.Errorf("Pod backend cannot create %q assignment", assignment.Backend)
}
labels := clone(metadata.Labels)
labels["app.kubernetes.io/name"] = "gitea-dynamic-runner"
labels["app.kubernetes.io/component"] = "executor"
pod, err := b.API.CreatePod(ctx, PodManifest{
Name: assignment.ID,
Namespace: b.Config.Namespace,
Labels: labels,
Annotations: clone(metadata.Annotations),
Image: b.Config.Image,
ServiceAccount: b.Config.ServiceAccount,
Args: append(append([]string{}, b.Config.ExecutorArgs...), assignment.ID),
})
if err != nil {
return nil, fmt.Errorf("create assignment Pod: %w", err)
}
return executor(pod), nil
}
func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor, identity taskidentity.Identity) error {
if err := b.validate(); err != nil {
return err
}
if executor == nil || executor.Name == "" || executor.IdentityTarget == "" {
return errors.New("Pod name and UID are required for identity binding")
}
if _, err := identityPath(identity.SPIFFEID, b.Config.TrustDomain); err != nil {
return err
}
return b.API.EnsureIdentityEntry(ctx, IdentityEntry{
Name: executor.Name,
Labels: map[string]string{
"app.kubernetes.io/name": "gitea-dynamic-runner",
"app.kubernetes.io/component": "pod-identity",
assignmentLabel: executor.Name,
},
ClassName: b.Config.SPIREClass,
ParentID: fmt.Sprintf(
"spiffe://%s/spire/agent/k8s_psat/%s/pod/%s",
b.Config.TrustDomain, b.Config.SPIRECluster, executor.IdentityTarget,
),
SPIFFEID: identity.SPIFFEID,
Selectors: []string{fmt.Sprintf("unix:uid:%d", b.Config.ExecutorUID)},
})
}
func identityPath(spiffeID, trustDomain string) (string, error) {
parsed, err := url.Parse(spiffeID)
if err != nil || parsed.Scheme != "spiffe" || parsed.Host != trustDomain || !strings.HasPrefix(parsed.Path, "/ci/") {
return "", fmt.Errorf("invalid CI SPIFFE ID %q", spiffeID)
}
return strings.TrimPrefix(parsed.Path, "/ci/"), nil
}
func (b Backend) Delete(ctx context.Context, executor *taskworker.Executor) error {
if err := b.validate(); err != nil {
return err
}
if executor == nil || executor.Name == "" {
return nil
}
if err := b.API.DeleteIdentityEntry(ctx, executor.Name); err != nil {
return fmt.Errorf("delete Pod identity entry: %w", err)
}
if err := b.API.DeletePod(ctx, b.Config.Namespace, executor.Name); err != nil {
return fmt.Errorf("delete assignment Pod: %w", err)
}
return nil
}
func (b Backend) validate() error {
if b.API == nil || b.Config.Namespace == "" || b.Config.Image == "" || b.Config.ServiceAccount == "" || b.Config.TrustDomain == "" || b.Config.SPIRECluster == "" || b.Config.SPIREClass == "" || b.Config.ExecutorUID < 1 {
return errors.New("Pod API and complete executor/SPIRE configuration are required")
}
return nil
}
func executor(pod Pod) *taskworker.Executor {
return &taskworker.Executor{
Name: pod.Name,
IdentityTarget: pod.UID,
Phase: phase(pod.Phase),
}
}
func phase(value string) taskworker.Phase {
switch value {
case "Succeeded":
return taskworker.PhaseSucceeded
case "Failed":
return taskworker.PhaseFailed
case "Running":
return taskworker.PhaseRunning
default:
return taskworker.PhasePending
}
}
func clone(source map[string]string) map[string]string {
target := make(map[string]string, len(source))
for key, value := range source {
target[key] = value
}
return target
}
+120
View File
@@ -0,0 +1,120 @@
package podbackend
import (
"context"
"testing"
runnerv1 "gitea.dev/actionslib/runner/v1"
"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 fakeAPI struct {
pods []Pod
selector string
created PodManifest
entry IdentityEntry
entryGone string
deleted string
}
func (a *fakeAPI) ListPods(_ context.Context, _ string, selector string) ([]Pod, error) {
a.selector = selector
return a.pods, nil
}
func (a *fakeAPI) CreatePod(_ context.Context, manifest PodManifest) (Pod, error) {
a.created = manifest
return Pod{Name: manifest.Name, Namespace: manifest.Namespace, UID: "pod-uid", Phase: "Pending"}, nil
}
func (a *fakeAPI) DeletePod(_ context.Context, _, name string) error {
a.deleted = name
return nil
}
func (a *fakeAPI) EnsureIdentityEntry(_ context.Context, entry IdentityEntry) error {
a.entry = entry
return nil
}
func (a *fakeAPI) DeleteIdentityEntry(_ context.Context, name string) error {
a.entryGone = name
return nil
}
func backend(api API) Backend {
return Backend{API: api, Config: Config{
Namespace: "gitea-actions", Image: "zot/ci-executor:main",
ServiceAccount: "gitea-task-executor", ExecutorArgs: []string{"execute"},
TrustDomain: "ddupan.top", SPIRECluster: "homelab",
SPIREClass: "spire-mgmt-spire", ExecutorUID: 2000,
}}
}
func assignment() taskassignment.Assignment {
return taskassignment.Assignment{
ID: "gitea-task-42", Backend: taskassignment.BackendPod,
Task: &runnerv1.Task{Id: 42},
Identity: taskidentity.Identity{
Repository: "owner/repo", Task: "publish",
SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish",
},
}
}
func TestFindRecoversPodByAssignmentLabel(t *testing.T) {
api := &fakeAPI{pods: []Pod{{Name: "gitea-task-42", UID: "uid", Phase: "Running"}}}
executor, err := backend(api).Find(context.Background(), "gitea-task-42")
if err != nil {
t.Fatal(err)
}
if api.selector != "ci.ddupan.top/assignment-id=gitea-task-42" || executor.Name != "gitea-task-42" || executor.IdentityTarget != "uid" || executor.Phase != taskworker.PhaseRunning {
t.Fatalf("selector=%q executor=%#v", api.selector, executor)
}
}
func TestCreateUsesDeterministicNameAndRecoveryMetadata(t *testing.T) {
api := &fakeAPI{}
metadata := taskworker.BackendMetadata(assignment())
executor, err := backend(api).Create(context.Background(), assignment(), metadata)
if err != nil {
t.Fatal(err)
}
if api.created.Name != "gitea-task-42" || api.created.Labels[assignmentLabel] != "gitea-task-42" {
t.Fatalf("manifest = %#v", api.created)
}
if api.created.Annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
t.Fatalf("annotations = %#v", api.created.Annotations)
}
if len(api.created.Args) != 2 || api.created.Args[1] != "gitea-task-42" || executor.IdentityTarget != "pod-uid" {
t.Fatalf("args=%v executor=%#v", api.created.Args, executor)
}
}
func TestBindIdentityCreatesEntryPinnedToPodUID(t *testing.T) {
api := &fakeAPI{}
executor := &taskworker.Executor{Name: "gitea-task-42", IdentityTarget: "pod-uid"}
if err := backend(api).BindIdentity(context.Background(), executor, assignment().Identity); err != nil {
t.Fatal(err)
}
if api.entry.Name != executor.Name || api.entry.SPIFFEID != assignment().Identity.SPIFFEID || api.entry.ParentID != "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/pod/pod-uid" || len(api.entry.Selectors) != 1 || api.entry.Selectors[0] != "unix:uid:2000" {
t.Fatalf("entry=%#v", api.entry)
}
}
func TestDeleteRemovesIdentityBeforePod(t *testing.T) {
api := &fakeAPI{}
executor := &taskworker.Executor{Name: "gitea-task-42", IdentityTarget: "pod-uid"}
if err := backend(api).Delete(context.Background(), executor); err != nil {
t.Fatal(err)
}
if api.entryGone != executor.Name || api.deleted != executor.Name {
t.Fatalf("entry=%q pod=%q", api.entryGone, api.deleted)
}
}
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 {
t.Fatal("expected duplicate executor error")
}
}
+163
View File
@@ -0,0 +1,163 @@
package podbackend
import (
"context"
"fmt"
"reflect"
corev1 "k8s.io/api/core/v1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/client-go/dynamic"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
)
var identityEntryResource = schema.GroupVersionResource{
Group: "spire.spiffe.io", Version: "v1alpha1", Resource: "clusterstaticentries",
}
// Client uses client-go's typed client for Pods and its dynamic client for the
// SPIRE Operator CRD.
type Client struct {
Kubernetes kubernetes.Interface
Dynamic dynamic.Interface
}
func NewClient(config *rest.Config) (*Client, error) {
kubernetesClient, err := kubernetes.NewForConfig(config)
if err != nil {
return nil, fmt.Errorf("create Kubernetes client: %w", err)
}
dynamicClient, err := dynamic.NewForConfig(config)
if err != nil {
return nil, fmt.Errorf("create Kubernetes dynamic client: %w", err)
}
return &Client{Kubernetes: kubernetesClient, Dynamic: dynamicClient}, nil
}
func NewInClusterClient() (*Client, error) {
config, err := rest.InClusterConfig()
if err != nil {
return nil, fmt.Errorf("load in-cluster Kubernetes config: %w", err)
}
return NewClient(config)
}
func (c *Client) ListPods(ctx context.Context, namespace, selector string) ([]Pod, error) {
list, err := c.Kubernetes.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: selector})
if err != nil {
return nil, err
}
pods := make([]Pod, 0, len(list.Items))
for _, item := range list.Items {
pods = append(pods, podFromKubernetes(item))
}
return pods, nil
}
func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, error) {
document := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: manifest.Name, Namespace: manifest.Namespace,
Labels: manifest.Labels, Annotations: manifest.Annotations,
},
Spec: corev1.PodSpec{
ServiceAccountName: manifest.ServiceAccount,
RestartPolicy: corev1.RestartPolicyNever,
Containers: []corev1.Container{{
Name: "executor", Image: manifest.Image, Args: manifest.Args,
SecurityContext: &corev1.SecurityContext{Privileged: boolPointer(true)},
VolumeMounts: []corev1.VolumeMount{{
Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true,
}},
}},
Volumes: []corev1.Volume{{
Name: "spire-agent-socket",
VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
}},
}},
},
}
created, err := c.Kubernetes.CoreV1().Pods(manifest.Namespace).Create(ctx, document, metav1.CreateOptions{})
if err != nil {
return Pod{}, err
}
return podFromKubernetes(*created), nil
}
func (c *Client) DeletePod(ctx context.Context, namespace, name string) error {
policy := metav1.DeletePropagationBackground
err := c.Kubernetes.CoreV1().Pods(namespace).Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &policy})
if apierrors.IsNotFound(err) {
return nil
}
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{})
if err == nil {
existingSpec, _, nestedErr := unstructured.NestedMap(existing.Object, "spec")
if nestedErr != nil {
return nestedErr
}
if !reflect.DeepEqual(existingSpec, identityEntryObject(entry).Object["spec"]) {
return fmt.Errorf("identity entry %s exists with different selectors or SPIFFE ID", entry.Name)
}
return nil
}
if !apierrors.IsNotFound(err) {
return err
}
_, err = resource.Create(ctx, identityEntryObject(entry), metav1.CreateOptions{})
return err
}
func (c *Client) DeleteIdentityEntry(ctx context.Context, name string) error {
policy := metav1.DeletePropagationBackground
err := c.Dynamic.Resource(identityEntryResource).Delete(ctx, name, metav1.DeleteOptions{PropagationPolicy: &policy})
if apierrors.IsNotFound(err) {
return nil
}
return err
}
func identityEntryObject(entry IdentityEntry) *unstructured.Unstructured {
selectors := make([]any, len(entry.Selectors))
for index, selector := range entry.Selectors {
selectors[index] = selector
}
return &unstructured.Unstructured{Object: map[string]any{
"apiVersion": "spire.spiffe.io/v1alpha1",
"kind": "ClusterStaticEntry",
"metadata": map[string]any{
"name": entry.Name, "labels": stringMap(entry.Labels),
},
"spec": map[string]any{
"className": entry.ClassName, "parentID": entry.ParentID,
"spiffeID": entry.SPIFFEID, "selectors": selectors,
},
}}
}
func podFromKubernetes(pod corev1.Pod) Pod {
return Pod{
Name: pod.Name, Namespace: pod.Namespace, UID: string(pod.UID),
Phase: string(pod.Status.Phase), Labels: pod.Labels, Annotations: pod.Annotations,
}
}
func boolPointer(value bool) *bool { return &value }
func stringMap(values map[string]string) map[string]any {
result := make(map[string]any, len(values))
for key, value := range values {
result[key] = value
}
return result
}
+72
View File
@@ -0,0 +1,72 @@
package podbackend
import (
"context"
"testing"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
dynamicfake "k8s.io/client-go/dynamic/fake"
"k8s.io/client-go/kubernetes/fake"
)
func testClient(objects ...runtime.Object) *Client {
return &Client{
Kubernetes: fake.NewSimpleClientset(objects...),
Dynamic: dynamicfake.NewSimpleDynamicClient(runtime.NewScheme()),
}
}
func TestClientPodLifecycleUsesTypedClient(t *testing.T) {
client := testClient()
created, err := client.CreatePod(context.Background(), PodManifest{
Name: "gitea-task-42", Namespace: "gitea-actions",
Labels: map[string]string{assignmentLabel: "gitea-task-42"},
Image: "zot/ci-executor:main", ServiceAccount: "gitea-task-executor",
Args: []string{"execute", "gitea-task-42"},
})
if err != nil {
t.Fatal(err)
}
pod, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Get(context.Background(), created.Name, metav1.GetOptions{})
if err != nil {
t.Fatal(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 {
t.Fatal(err)
}
pods, err := client.ListPods(context.Background(), "gitea-actions", assignmentLabel+"=gitea-task-42")
if err != nil || len(pods) != 1 || pods[0].UID != "pod-uid" || pods[0].Phase != "Running" {
t.Fatalf("pods=%#v err=%v", pods, err)
}
if err := client.DeletePod(context.Background(), "gitea-actions", created.Name); err != nil {
t.Fatal(err)
}
}
func TestClientEnsuresIdempotentClusterStaticEntry(t *testing.T) {
client := testClient()
entry := IdentityEntry{
Name: "gitea-task-42", Labels: map[string]string{assignmentLabel: "gitea-task-42"},
ClassName: "spire-mgmt-spire",
ParentID: "spiffe://ddupan.top/spire/agent/k8s_psat/homelab/pod/pod-uid",
SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish",
Selectors: []string{"unix:uid:2000"},
}
if err := client.EnsureIdentityEntry(context.Background(), entry); err != nil {
t.Fatal(err)
}
if err := client.EnsureIdentityEntry(context.Background(), entry); err != nil {
t.Fatal(err)
}
if err := client.DeleteIdentityEntry(context.Background(), entry.Name); err != nil {
t.Fatal(err)
}
if err := client.DeleteIdentityEntry(context.Background(), entry.Name); err != nil {
t.Fatal(err)
}
}