203 lines
6.0 KiB
Go
203 lines
6.0 KiB
Go
// 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
|
|
Environment map[string]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
|
|
SPIREAgentID 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, launch taskworker.LaunchSpec) (*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(launch.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(launch.Metadata.Annotations),
|
|
Image: b.Config.Image,
|
|
ServiceAccount: b.Config.ServiceAccount,
|
|
Args: append([]string{}, b.Config.ExecutorArgs...),
|
|
Environment: clone(launch.Environment),
|
|
})
|
|
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: b.Config.SPIREAgentID,
|
|
SPIFFEID: identity.SPIFFEID,
|
|
Selectors: []string{"k8s:pod-uid:" + executor.IdentityTarget},
|
|
})
|
|
}
|
|
|
|
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.SPIREAgentID == "" || 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
|
|
}
|