Compare commits
29
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8f8ec04b18
|
||
|
|
74ffd49d08 | ||
|
|
e3ff308772
|
||
|
|
feb0b84b7c | ||
|
|
66b90146f8
|
||
|
|
3d8a04e4f7 | ||
|
|
fef7e5a214
|
||
|
|
51b940468e
|
||
|
|
b614ef4c2f
|
||
|
|
8fa8e46320
|
||
|
|
cc4405f788
|
||
|
|
4a428c4384 | ||
|
|
9a28a1573e
|
||
|
|
01995bf084 | ||
|
|
3d787b2dd3
|
||
|
|
54661411e3 | ||
|
|
53a080b310
|
||
|
|
38e8d59541
|
||
|
|
661b5e9218 | ||
|
|
48b6b8038e
|
||
|
|
70c5ff422f | ||
|
|
906e6a2e18
|
||
|
|
ace84373f6 | ||
|
|
87cf359ef6
|
||
|
|
ad3934e7e1 | ||
|
|
d82687382a
|
||
|
|
7f52cf393f | ||
|
|
8a186dcd86 | ||
|
|
327c73e744 |
@@ -17,7 +17,7 @@ jobs:
|
|||||||
- run: go vet ./...
|
- run: go vet ./...
|
||||||
|
|
||||||
python:
|
python:
|
||||||
runs-on: self-hosted
|
runs-on: [self-hosted, pod]
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v4
|
- uses: actions/checkout@v4
|
||||||
- uses: actions/setup-python@v5
|
- uses: actions/setup-python@v5
|
||||||
@@ -28,7 +28,7 @@ jobs:
|
|||||||
- run: python -m compileall -q src
|
- run: python -m compileall -q src
|
||||||
|
|
||||||
shell:
|
shell:
|
||||||
runs-on: self-hosted
|
runs-on: [self-hosted, pod]
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v4
|
- uses: actions/checkout@v4
|
||||||
- run: |
|
- run: |
|
||||||
|
|||||||
@@ -1,31 +1,174 @@
|
|||||||
|
---
|
||||||
name: VM kind smoke
|
name: VM kind smoke
|
||||||
|
|
||||||
|
# yamllint disable-line rule:truthy
|
||||||
on:
|
on:
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
kind:
|
kind:
|
||||||
runs-on: [self-hosted, vm]
|
runs-on: [self-hosted, vm-dev]
|
||||||
steps:
|
steps:
|
||||||
|
- name: Start job-local Docker
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
if [[ ! -e /dev/kmsg ]]; then
|
||||||
|
sudo mknod /dev/kmsg c 1 11
|
||||||
|
fi
|
||||||
|
sudo install -d /var/lib/docker
|
||||||
|
sudo truncate -s 20G /tmp/docker-data.img
|
||||||
|
sudo mkfs.ext4 -F /tmp/docker-data.img
|
||||||
|
if [[ ! -e /dev/loop-control ]]; then
|
||||||
|
sudo mknod /dev/loop-control c 10 237
|
||||||
|
fi
|
||||||
|
for minor in {0..7}; do
|
||||||
|
if [[ ! -e "/dev/loop${minor}" ]]; then
|
||||||
|
sudo mknod "/dev/loop${minor}" b 7 "$minor"
|
||||||
|
fi
|
||||||
|
done
|
||||||
|
loop_device=$(sudo losetup --find --show /tmp/docker-data.img)
|
||||||
|
sudo mount "$loop_device" /var/lib/docker
|
||||||
|
findmnt /var/lib/docker
|
||||||
|
|
||||||
|
# A nested systemd needs a domain cgroup namespace. This is the
|
||||||
|
# cgroup v2 nesting initialization performed by the official DinD
|
||||||
|
# entrypoint, kept here because the shared runner image deliberately
|
||||||
|
# does not carry a second DinD-specific entrypoint.
|
||||||
|
if [[ -f /sys/fs/cgroup/cgroup.controllers ]]; then
|
||||||
|
sudo sh -c '
|
||||||
|
mkdir -p /sys/fs/cgroup/init
|
||||||
|
while read -r pid; do
|
||||||
|
printf "%s\n" "$pid" \
|
||||||
|
>/sys/fs/cgroup/init/cgroup.procs 2>/dev/null || true
|
||||||
|
done </sys/fs/cgroup/cgroup.procs
|
||||||
|
sed -e "s/ / +/g" -e "s/^/+/" \
|
||||||
|
/sys/fs/cgroup/cgroup.controllers \
|
||||||
|
>/sys/fs/cgroup/cgroup.subtree_control
|
||||||
|
'
|
||||||
|
fi
|
||||||
|
|
||||||
|
sudo nohup dockerd \
|
||||||
|
--host=unix:///var/run/docker.sock \
|
||||||
|
--storage-driver=overlay2 \
|
||||||
|
>/tmp/dockerd.log 2>&1 &
|
||||||
|
for _ in {1..60}; do
|
||||||
|
if docker info >/dev/null 2>&1; then
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
sleep 1
|
||||||
|
done
|
||||||
|
cat /tmp/dockerd.log
|
||||||
|
exit 1
|
||||||
|
|
||||||
- name: Verify Docker
|
- name: Verify Docker
|
||||||
run: docker info
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
docker info
|
||||||
|
echo '### runner cgroup'
|
||||||
|
cat /proc/self/cgroup
|
||||||
|
cat /sys/fs/cgroup/cgroup.type
|
||||||
|
cat /sys/fs/cgroup/cgroup.controllers
|
||||||
|
cat /sys/fs/cgroup/cgroup.subtree_control
|
||||||
|
echo '### nested private cgroup namespace'
|
||||||
|
docker run --rm --privileged --cgroupns=private alpine:3.22 \
|
||||||
|
sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
|
||||||
|
echo '### nested host cgroup namespace'
|
||||||
|
docker run --rm --privileged --cgroupns=host alpine:3.22 \
|
||||||
|
sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
|
||||||
|
|
||||||
- name: Install kind
|
- name: Install kind
|
||||||
shell: bash
|
shell: bash
|
||||||
run: |
|
run: |
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
version=v0.33.0
|
version=v0.33.0
|
||||||
|
base_url="https://kind.sigs.k8s.io/dl/${version}"
|
||||||
curl --fail --location --silent --show-error \
|
curl --fail --location --silent --show-error \
|
||||||
--output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64"
|
--output /tmp/kind "${base_url}/kind-linux-amd64"
|
||||||
curl --fail --location --silent --show-error \
|
curl --fail --location --silent --show-error \
|
||||||
--output /tmp/kind.sha256sum "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64.sha256sum"
|
--output /tmp/kind.sha256sum \
|
||||||
printf '%s %s\n' "$(cut -d ' ' -f1 /tmp/kind.sha256sum)" /tmp/kind | sha256sum --check
|
"${base_url}/kind-linux-amd64.sha256sum"
|
||||||
|
checksum=$(cut -d ' ' -f1 /tmp/kind.sha256sum)
|
||||||
|
printf '%s %s\n' "$checksum" /tmp/kind | sha256sum --check
|
||||||
chmod 0755 /tmp/kind
|
chmod 0755 /tmp/kind
|
||||||
|
|
||||||
- name: Create and delete kind cluster
|
- name: Create and delete kind cluster
|
||||||
shell: bash
|
shell: bash
|
||||||
run: |
|
run: |
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
trap '/tmp/kind delete cluster --name smoke' EXIT
|
diagnose_and_cleanup() {
|
||||||
/tmp/kind create cluster --name smoke --wait 180s
|
status=$?
|
||||||
|
node=smoke-control-plane
|
||||||
|
if (( status != 0 )) && docker inspect "$node" >/dev/null 2>&1; then
|
||||||
|
echo '::group::kind node inspect'
|
||||||
|
docker inspect "$node"
|
||||||
|
echo '::endgroup::'
|
||||||
|
echo '::group::kind node logs'
|
||||||
|
docker logs "$node" 2>&1 || true
|
||||||
|
echo '::endgroup::'
|
||||||
|
echo '::group::kind node guest state'
|
||||||
|
docker exec "$node" bash -c '
|
||||||
|
set +e
|
||||||
|
echo "### pid 1"
|
||||||
|
ps -p 1 -o pid,ppid,user,stat,comm,args
|
||||||
|
cat /proc/1/status
|
||||||
|
echo "### cgroup"
|
||||||
|
cat /proc/1/cgroup
|
||||||
|
findmnt -R /sys/fs/cgroup
|
||||||
|
stat -fc "%T %a" /sys/fs/cgroup
|
||||||
|
echo "### systemd"
|
||||||
|
systemctl --no-pager --failed
|
||||||
|
systemctl --no-pager status \
|
||||||
|
multi-user.target containerd.service kubelet.service
|
||||||
|
echo "### CRI containers"
|
||||||
|
endpoint=unix:///run/containerd/containerd.sock
|
||||||
|
crictl --runtime-endpoint "$endpoint" ps --all
|
||||||
|
for id in $(
|
||||||
|
crictl --runtime-endpoint "$endpoint" ps --all --quiet
|
||||||
|
); do
|
||||||
|
echo "### CRI container $id"
|
||||||
|
crictl --runtime-endpoint "$endpoint" inspect "$id"
|
||||||
|
crictl --runtime-endpoint "$endpoint" logs "$id"
|
||||||
|
done
|
||||||
|
echo "### containerd metadata"
|
||||||
|
timeout 10 ctr --namespace k8s.io containers list
|
||||||
|
timeout 10 ctr --namespace k8s.io snapshots list
|
||||||
|
echo "### runtime process stacks"
|
||||||
|
ps -e -o pid,ppid,stat,wchan:32,comm,args
|
||||||
|
for pid in $(pidof containerd containerd-shim-runc-v2); do
|
||||||
|
echo "### kernel stack $pid"
|
||||||
|
cat "/proc/$pid/stack"
|
||||||
|
done
|
||||||
|
kill -USR1 "$(pidof containerd)"
|
||||||
|
sleep 2
|
||||||
|
journalctl --no-pager -b -n 500
|
||||||
|
' 2>&1 || true
|
||||||
|
echo '::endgroup::'
|
||||||
|
fi
|
||||||
|
/tmp/kind delete cluster --name smoke || true
|
||||||
|
exit "$status"
|
||||||
|
}
|
||||||
|
trap diagnose_and_cleanup EXIT
|
||||||
|
# Nested Kata + Docker + kind cold starts can take longer than
|
||||||
|
# kubeadm's one-minute API-call default even after the static pods
|
||||||
|
# have been accepted. Give the API server enough time to become
|
||||||
|
# responsive before kubeadm creates its initial RBAC objects.
|
||||||
|
cat >/tmp/kind-config.yml <<'EOF'
|
||||||
|
kind: Cluster
|
||||||
|
apiVersion: kind.x-k8s.io/v1alpha4
|
||||||
|
nodes:
|
||||||
|
- role: control-plane
|
||||||
|
kubeadmConfigPatches:
|
||||||
|
- |
|
||||||
|
apiVersion: kubeadm.k8s.io/v1beta4
|
||||||
|
kind: InitConfiguration
|
||||||
|
timeouts:
|
||||||
|
kubernetesAPICall: 5m0s
|
||||||
|
EOF
|
||||||
|
/tmp/kind create cluster \
|
||||||
|
--name smoke \
|
||||||
|
--config /tmp/kind-config.yml \
|
||||||
|
--wait 300s \
|
||||||
|
--retain
|
||||||
/tmp/kind get clusters | grep -Fx smoke
|
/tmp/kind get clusters | grep -Fx smoke
|
||||||
|
|||||||
@@ -0,0 +1,36 @@
|
|||||||
|
---
|
||||||
|
name: VM runtime smoke
|
||||||
|
|
||||||
|
# yamllint disable-line rule:truthy
|
||||||
|
on:
|
||||||
|
workflow_dispatch:
|
||||||
|
|
||||||
|
jobs:
|
||||||
|
runtime:
|
||||||
|
runs-on: [self-hosted, vm-dev]
|
||||||
|
steps:
|
||||||
|
- name: Verify workload identity socket
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
: "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}"
|
||||||
|
socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://}
|
||||||
|
test -S "$socket_path"
|
||||||
|
|
||||||
|
- name: Verify Docker daemon
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
format='{{json .ServerVersion}} {{json .Driver}}'
|
||||||
|
format="$format {{json .CgroupVersion}}"
|
||||||
|
docker info --format "$format"
|
||||||
|
|
||||||
|
- name: Run and clean nested container
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
name=vm-runtime-smoke
|
||||||
|
trap 'docker rm --force "$name" >/dev/null 2>&1 || true' EXIT
|
||||||
|
output=$(docker run --name "$name" alpine:3.22 /bin/sh -c \
|
||||||
|
'test "$(uname -m)" = x86_64 && printf vm-runtime-ok')
|
||||||
|
test "$output" = vm-runtime-ok
|
||||||
@@ -15,6 +15,10 @@ runs-on: [self-hosted, vm]
|
|||||||
只执行一个 job,并在 job 结束后连同本地状态一起销毁。完整的设计约束见
|
只执行一个 job,并在 job 结束后连同本地状态一起销毁。完整的设计约束见
|
||||||
[`docs/design-principles.md`](docs/design-principles.md)。
|
[`docs/design-principles.md`](docs/design-principles.md)。
|
||||||
|
|
||||||
|
集成期间可将 `VM_RUNNER_LABEL=vm-dev`,只接取显式使用
|
||||||
|
`runs-on: [self-hosted, vm-dev]` 的测试任务;生产 `vm` job 将保持在 Gitea pending,
|
||||||
|
不会在 backend 修复过程中继续涌入。
|
||||||
|
|
||||||
目标 Go controller 组件:
|
目标 Go controller 组件:
|
||||||
|
|
||||||
- `scheduler`:以常驻 Gitea RunnerService 身份直接领取 task,并把完整 assignment
|
- `scheduler`:以常驻 Gitea RunnerService 身份直接领取 task,并把完整 assignment
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
"log/slog"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
"slices"
|
"slices"
|
||||||
@@ -49,7 +49,7 @@ type controllerConfig struct {
|
|||||||
PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
|
PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
|
||||||
SPIREAgentID string
|
SPIREAgentID string
|
||||||
PodExecutorUID, PodCapacity int
|
PodExecutorUID, PodCapacity int
|
||||||
OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool string
|
OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool, VMRunnerLabel string
|
||||||
VMTimeout, VMCapacity int
|
VMTimeout, VMCapacity int
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -131,7 +131,7 @@ func runController(ctx context.Context) error {
|
|||||||
labels = append(labels, string(taskassignment.BackendPod))
|
labels = append(labels, string(taskassignment.BackendPod))
|
||||||
}
|
}
|
||||||
if slices.Contains(config.Components, controller.VMWorker) {
|
if slices.Contains(config.Components, controller.VMWorker) {
|
||||||
labels = append(labels, string(taskassignment.BackendVM))
|
labels = append(labels, config.VMRunnerLabel)
|
||||||
}
|
}
|
||||||
poller := taskscheduler.Poller{
|
poller := taskscheduler.Poller{
|
||||||
Client: giteaClient,
|
Client: giteaClient,
|
||||||
@@ -139,7 +139,7 @@ func runController(ctx context.Context) error {
|
|||||||
JetStream: producerJS, SubjectBase: config.SubjectBase,
|
JetStream: producerJS, SubjectBase: config.SubjectBase,
|
||||||
}},
|
}},
|
||||||
Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity},
|
Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity},
|
||||||
OnError: func(err error) { log.Printf("scheduler: %v", err) },
|
OnError: func(err error) { slog.Error("scheduler error", "component", "scheduler", "error", err) },
|
||||||
}
|
}
|
||||||
kubernetesConfig, err := rest.InClusterConfig()
|
kubernetesConfig, err := rest.InClusterConfig()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -190,11 +190,13 @@ func runController(ctx context.Context) error {
|
|||||||
return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
|
return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, podPool)
|
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendPod)}, registry, podPool)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }}
|
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) {
|
||||||
|
slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendPod, "error", err)
|
||||||
|
}}
|
||||||
components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
|
components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
|
||||||
group, groupContext := errgroup.WithContext(ctx)
|
group, groupContext := errgroup.WithContext(ctx)
|
||||||
group.Go(func() error { return component.Run(groupContext) })
|
group.Go(func() error { return component.Run(groupContext) })
|
||||||
@@ -220,11 +222,13 @@ func runController(ctx context.Context) error {
|
|||||||
return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
|
return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, vmPool)
|
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendVM)}, registry, vmPool)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { log.Printf("VM lifecycle: %v", err) }}
|
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) {
|
||||||
|
slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendVM, "error", err)
|
||||||
|
}}
|
||||||
components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
|
components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
|
||||||
group, groupContext := errgroup.WithContext(ctx)
|
group, groupContext := errgroup.WithContext(ctx)
|
||||||
group.Go(func() error { return component.Run(groupContext) })
|
group.Go(func() error { return component.Run(groupContext) })
|
||||||
@@ -294,11 +298,21 @@ func workerComponent(ctx context.Context, js jetstream.JetStream, config control
|
|||||||
}
|
}
|
||||||
return assignmentqueue.ConsumerComponent{
|
return assignmentqueue.ConsumerComponent{
|
||||||
Consumer: consumer, Capacity: capacity,
|
Consumer: consumer, Capacity: capacity,
|
||||||
Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission},
|
Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission, OnEvent: func(event assignmentqueue.Event) {
|
||||||
OnError: func(err error) { log.Printf("%s worker: %v", backend, err) },
|
slog.Info("assignment transition", "component", "worker", "event", event.Name, "backend", event.Backend, "assignment", event.AssignmentID, "retry_delay", event.RetryDelay)
|
||||||
|
}},
|
||||||
|
OnError: func(err error) {
|
||||||
|
slog.Error("assignment processing error", "component", "worker", "backend", backend, "error", err)
|
||||||
|
},
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func workerEventLogger(backend taskassignment.Backend) func(taskworker.Event) {
|
||||||
|
return func(event taskworker.Event) {
|
||||||
|
slog.Info("executor transition", "component", "worker", "event", event.Name, "backend", backend, "assignment", event.AssignmentID, "executor", event.Executor, "phase", event.Phase)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func loadControllerConfig() (controllerConfig, error) {
|
func loadControllerConfig() (controllerConfig, error) {
|
||||||
selection, err := controller.ParseSelection(os.Getenv("COMPONENTS"))
|
selection, err := controller.ParseSelection(os.Getenv("COMPONENTS"))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -345,7 +359,7 @@ func loadControllerConfig() (controllerConfig, error) {
|
|||||||
FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey),
|
FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey),
|
||||||
PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"),
|
PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"),
|
||||||
SPIRECluster: env("SPIRE_CLUSTER", "homelab"), SPIREClass: env("SPIRE_CLASS", "spire-mgmt-spire"), SPIREAgentID: os.Getenv("SPIRE_AGENT_ID"), PodExecutorUID: envInt("POD_EXECUTOR_UID", 2000), PodCapacity: envInt("POD_CAPACITY", 4),
|
SPIRECluster: env("SPIRE_CLUSTER", "homelab"), SPIREClass: env("SPIRE_CLASS", "spire-mgmt-spire"), SPIREAgentID: os.Getenv("SPIRE_AGENT_ID"), PodExecutorUID: envInt("POD_EXECUTOR_UID", 2000), PodCapacity: envInt("POD_CAPACITY", 4),
|
||||||
OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
|
OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-vm"), VMRunnerLabel: env("VM_RUNNER_LABEL", "vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
|
||||||
}
|
}
|
||||||
if config.WorkloadAPIAddr == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
|
if config.WorkloadAPIAddr == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
|
||||||
return controllerConfig{}, errors.New("SPIFFE_ENDPOINT_SOCKET, RUNNER_FACADE_URL, and RUNNER_FACADE_SPIFFE_ID are required")
|
return controllerConfig{}, errors.New("SPIFFE_ENDPOINT_SOCKET, RUNNER_FACADE_URL, and RUNNER_FACADE_SPIFFE_ID are required")
|
||||||
@@ -357,6 +371,9 @@ func loadControllerConfig() (controllerConfig, error) {
|
|||||||
return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker")
|
return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker")
|
||||||
}
|
}
|
||||||
if slices.Contains(selection, controller.VMWorker) {
|
if slices.Contains(selection, controller.VMWorker) {
|
||||||
|
if config.VMRunnerLabel != "vm" && config.VMRunnerLabel != "vm-dev" {
|
||||||
|
return controllerConfig{}, errors.New("VM_RUNNER_LABEL must be vm or vm-dev")
|
||||||
|
}
|
||||||
if config.OpenSandboxURL == "" {
|
if config.OpenSandboxURL == "" {
|
||||||
return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker")
|
return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -56,6 +56,7 @@ func TestLoadControllerConfigRequiresOpenSandboxSecretOnlyForVM(t *testing.T) {
|
|||||||
t.Setenv("RUNNER_FACADE_URL", "https://facade:8443")
|
t.Setenv("RUNNER_FACADE_URL", "https://facade:8443")
|
||||||
t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/controller")
|
t.Setenv("RUNNER_FACADE_SPIFFE_ID", "spiffe://ddupan.top/controller")
|
||||||
t.Setenv("OPENSANDBOX_API", "http://opensandbox.internal")
|
t.Setenv("OPENSANDBOX_API", "http://opensandbox.internal")
|
||||||
|
t.Setenv("VM_RUNNER_LABEL", "vm-dev")
|
||||||
if _, err := loadControllerConfig(); err == nil {
|
if _, err := loadControllerConfig(); err == nil {
|
||||||
t.Fatal("expected missing OpenSandbox API key file error")
|
t.Fatal("expected missing OpenSandbox API key file error")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"log/slog"
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"syscall"
|
"syscall"
|
||||||
@@ -12,8 +13,9 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, nil)))
|
||||||
if err := run(); err != nil {
|
if err := run(); err != nil {
|
||||||
fmt.Fprintln(os.Stderr, err)
|
slog.Error("runner stopped", "error", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,9 +40,11 @@ UID attestation 的临时 Agent 失去父级。
|
|||||||
|
|
||||||
`ci-vm` 使用 `kata-clh-runtime-rs`;`ci-pod` 使用默认 runc。两者都要求:
|
`ci-vm` 使用 `kata-clh-runtime-rs`;`ci-pod` 使用默认 runc。两者都要求:
|
||||||
|
|
||||||
- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI 和 identity gate;
|
- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI、Docker 工具和
|
||||||
- runner UID 2000,SPIRE Agent 与 privileged dockerd 使用不同 UID;
|
identity gate;Pod 与 VM backend 使用同一个镜像;
|
||||||
- Docker socket 通过 group 2000 共享,Docker 数据仅存在于 sandbox emptyDir;
|
- runner UID 2000;SPIRE Agent 独立运行,需要 Docker 的 workflow 通过 sudo 在
|
||||||
|
privileged executor 内启动 job-local daemon;
|
||||||
|
- Kata VM 中 Docker 数据使用 guest 内的 loop-backed ext4,并随 sandbox 一起删除;
|
||||||
- `self-hosted` 必须是所有 runner labels 的前缀;
|
- `self-hosted` 必须是所有 runner labels 的前缀;
|
||||||
- ephemeral/once runner 完成一项任务后退出。
|
- ephemeral/once runner 完成一项任务后退出。
|
||||||
|
|
||||||
|
|||||||
@@ -78,6 +78,22 @@ type Processor struct {
|
|||||||
Admission Admission
|
Admission Admission
|
||||||
RetryDelay time.Duration
|
RetryDelay time.Duration
|
||||||
ClaimTimeout time.Duration
|
ClaimTimeout time.Duration
|
||||||
|
OnEvent func(Event)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Event describes a non-sensitive assignment handoff transition. It never
|
||||||
|
// contains task payloads, credentials, capabilities, or workload identities.
|
||||||
|
type Event struct {
|
||||||
|
Name string
|
||||||
|
AssignmentID string
|
||||||
|
Backend taskassignment.Backend
|
||||||
|
RetryDelay time.Duration
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p Processor) event(name string, assignment taskassignment.Assignment, retryDelay time.Duration) {
|
||||||
|
if p.OnEvent != nil {
|
||||||
|
p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Backend: assignment.Backend, RetryDelay: retryDelay})
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p Processor) Process(ctx context.Context, message Message) error {
|
func (p Processor) Process(ctx context.Context, message Message) error {
|
||||||
@@ -88,6 +104,7 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return errors.Join(err, message.TermWithReason("invalid assignment"))
|
return errors.Join(err, message.TermWithReason("invalid assignment"))
|
||||||
}
|
}
|
||||||
|
p.event("received", assignment, 0)
|
||||||
if _, err := p.Claims.Offer(assignment); err != nil {
|
if _, err := p.Claims.Offer(assignment); err != nil {
|
||||||
return errors.Join(err, message.TermWithReason("conflicting assignment"))
|
return errors.Join(err, message.TermWithReason("conflicting assignment"))
|
||||||
}
|
}
|
||||||
@@ -96,8 +113,10 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
|||||||
if delay <= 0 {
|
if delay <= 0 {
|
||||||
delay = 2 * time.Second
|
delay = 2 * time.Second
|
||||||
}
|
}
|
||||||
|
p.event("capacity_wait", assignment, delay)
|
||||||
return message.NakWithDelay(delay)
|
return message.NakWithDelay(delay)
|
||||||
}
|
}
|
||||||
|
p.event("capacity_acquired", assignment, 0)
|
||||||
accepted, err := p.Accepter.Accept(ctx, assignment)
|
accepted, err := p.Accepter.Accept(ctx, assignment)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
p.Admission.Release(assignment.ID)
|
p.Admission.Release(assignment.ID)
|
||||||
@@ -105,9 +124,11 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
|||||||
if delay <= 0 {
|
if delay <= 0 {
|
||||||
delay = 15 * time.Second
|
delay = 15 * time.Second
|
||||||
}
|
}
|
||||||
|
p.event("backend_retry", assignment, delay)
|
||||||
return errors.Join(err, message.NakWithDelay(delay))
|
return errors.Join(err, message.NakWithDelay(delay))
|
||||||
}
|
}
|
||||||
if accepted {
|
if accepted {
|
||||||
|
p.event("backend_ready", assignment, 0)
|
||||||
timeout := p.ClaimTimeout
|
timeout := p.ClaimTimeout
|
||||||
if timeout <= 0 {
|
if timeout <= 0 {
|
||||||
timeout = 4 * time.Minute
|
timeout = 4 * time.Minute
|
||||||
@@ -120,17 +141,21 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
|||||||
if delay <= 0 {
|
if delay <= 0 {
|
||||||
delay = 2 * time.Second
|
delay = 2 * time.Second
|
||||||
}
|
}
|
||||||
|
p.event("claim_timeout", assignment, delay)
|
||||||
return errors.Join(err, message.NakWithDelay(delay))
|
return errors.Join(err, message.NakWithDelay(delay))
|
||||||
}
|
}
|
||||||
|
p.event("runner_claimed", assignment, 0)
|
||||||
if err := message.DoubleAck(ctx); err != nil {
|
if err := message.DoubleAck(ctx); err != nil {
|
||||||
return fmt.Errorf("ack assignment %s: %w", assignment.ID, err)
|
return fmt.Errorf("ack assignment %s: %w", assignment.ID, err)
|
||||||
}
|
}
|
||||||
|
p.event("acked", assignment, 0)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
delay := p.RetryDelay
|
delay := p.RetryDelay
|
||||||
if delay <= 0 {
|
if delay <= 0 {
|
||||||
delay = 2 * time.Second
|
delay = 2 * time.Second
|
||||||
}
|
}
|
||||||
|
p.event("backend_pending", assignment, delay)
|
||||||
return message.NakWithDelay(delay)
|
return message.NakWithDelay(delay)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -167,7 +192,7 @@ func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectB
|
|||||||
AckPolicy: jetstream.AckExplicitPolicy,
|
AckPolicy: jetstream.AckExplicitPolicy,
|
||||||
AckWait: 5 * time.Minute,
|
AckWait: 5 * time.Minute,
|
||||||
MaxAckPending: capacity,
|
MaxAckPending: capacity,
|
||||||
MaxDeliver: 20,
|
MaxDeliver: 1000,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)
|
return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)
|
||||||
|
|||||||
@@ -126,13 +126,23 @@ func encodedAssignment(t *testing.T) []byte {
|
|||||||
|
|
||||||
func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
|
func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
|
||||||
message := &fakeMessage{data: encodedAssignment(t)}
|
message := &fakeMessage{data: encodedAssignment(t)}
|
||||||
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}}
|
var events []Event
|
||||||
|
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}, OnEvent: func(event Event) { events = append(events, event) }}
|
||||||
if err := processor.Process(context.Background(), message); err != nil {
|
if err := processor.Process(context.Background(), message); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if message.acked != 1 || message.nacked != 0 {
|
if message.acked != 1 || message.nacked != 0 {
|
||||||
t.Fatalf("message = %#v", message)
|
t.Fatalf("message = %#v", message)
|
||||||
}
|
}
|
||||||
|
want := []string{"received", "capacity_acquired", "backend_ready", "runner_claimed", "acked"}
|
||||||
|
if len(events) != len(want) {
|
||||||
|
t.Fatalf("events = %#v", events)
|
||||||
|
}
|
||||||
|
for index := range want {
|
||||||
|
if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Backend != taskassignment.BackendPod {
|
||||||
|
t.Fatalf("event[%d] = %#v", index, events[index])
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
|
func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
|
||||||
@@ -205,7 +215,7 @@ func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
|
|||||||
if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
|
if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 {
|
if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
|
||||||
t.Fatalf("config = %#v", manager.config)
|
t.Fatalf("config = %#v", manager.config)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,9 +3,13 @@ package opensandboxbackend
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"encoding/base64"
|
||||||
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"sort"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
|
opensandbox "github.com/alibaba/OpenSandbox/sdks/sandbox/go"
|
||||||
@@ -17,6 +21,8 @@ import (
|
|||||||
|
|
||||||
const assignmentMetadata = "ci.ddupan.top/assignment-id"
|
const assignmentMetadata = "ci.ddupan.top/assignment-id"
|
||||||
const terminalMetadata = "ci.ddupan.top/terminal"
|
const terminalMetadata = "ci.ddupan.top/terminal"
|
||||||
|
const annotationsMetadataPrefix = "ci.ddupan.top/annotations-"
|
||||||
|
const metadataValueLimit = 63
|
||||||
|
|
||||||
type Lifecycle interface {
|
type Lifecycle interface {
|
||||||
ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error)
|
ListSandboxes(context.Context, opensandbox.ListOptions) (*opensandbox.ListSandboxesResponse, error)
|
||||||
@@ -123,7 +129,11 @@ func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]
|
|||||||
if sandbox.Metadata[terminalMetadata] == "true" {
|
if sandbox.Metadata[terminalMetadata] == "true" {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
assignment, err := taskassignment.FromMetadata(sandbox.Metadata, sandbox.Metadata, trustDomain)
|
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 {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err)
|
return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err)
|
||||||
}
|
}
|
||||||
@@ -169,8 +179,12 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen
|
|||||||
for key, value := range launch.Environment {
|
for key, value := range launch.Environment {
|
||||||
environment[key] = value
|
environment[key] = value
|
||||||
}
|
}
|
||||||
sandboxMetadata := clone(launch.Metadata.Annotations)
|
sandboxMetadata := clone(launch.Metadata.Labels)
|
||||||
for key, value := range 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
|
sandboxMetadata[key] = value
|
||||||
}
|
}
|
||||||
request := opensandbox.CreateSandboxRequest{
|
request := opensandbox.CreateSandboxRequest{
|
||||||
@@ -194,7 +208,11 @@ func (b Backend) BindIdentity(ctx context.Context, executor *taskworker.Executor
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("verify sandbox identity metadata: %w", err)
|
return fmt.Errorf("verify sandbox identity metadata: %w", err)
|
||||||
}
|
}
|
||||||
if sandbox.Metadata["ci.ddupan.top/spiffe-id"] != identity.SPIFFEID {
|
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 fmt.Errorf("sandbox %s has inconsistent SPIFFE identity metadata", executor.Name)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
@@ -243,3 +261,50 @@ func clone(source map[string]string) map[string]string {
|
|||||||
}
|
}
|
||||||
return result
|
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
|
||||||
|
}
|
||||||
|
|||||||
@@ -31,7 +31,8 @@ func (f *fakeLifecycle) GetSandbox(_ context.Context, id string) (*opensandbox.S
|
|||||||
return &item, nil
|
return &item, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return &opensandbox.SandboxInfo{ID: id, Metadata: map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID}}, nil
|
metadata, _ := encodeAnnotations(map[string]string{"ci.ddupan.top/spiffe-id": assignment().Identity.SPIFFEID})
|
||||||
|
return &opensandbox.SandboxInfo{ID: id, Metadata: metadata}, nil
|
||||||
}
|
}
|
||||||
func (f *fakeLifecycle) PatchSandboxMetadata(_ context.Context, id string, patch opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error) {
|
func (f *fakeLifecycle) PatchSandboxMetadata(_ context.Context, id string, patch opensandbox.MetadataPatch) (*opensandbox.SandboxInfo, error) {
|
||||||
for index := range f.items {
|
for index := range f.items {
|
||||||
@@ -89,12 +90,38 @@ func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
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" {
|
if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Metadata["ci.ddupan.top/runner"] != "true" || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" {
|
||||||
t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor)
|
t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor)
|
||||||
}
|
}
|
||||||
if lifecycle.created.Env["CI_RUNNER_CAPABILITY"] != "capability" {
|
if lifecycle.created.Env["CI_RUNNER_CAPABILITY"] != "capability" {
|
||||||
t.Fatalf("environment = %#v", lifecycle.created.Env)
|
t.Fatalf("environment = %#v", lifecycle.created.Env)
|
||||||
}
|
}
|
||||||
|
annotations, err := decodeAnnotations(lifecycle.created.Metadata)
|
||||||
|
if err != nil || annotations["ci.ddupan.top/repository"] != "owner/repo" || annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID {
|
||||||
|
t.Fatalf("annotations=%#v err=%v", annotations, err)
|
||||||
|
}
|
||||||
|
for _, value := range lifecycle.created.Metadata {
|
||||||
|
if len(value) > metadataValueLimit {
|
||||||
|
t.Fatalf("metadata value exceeds %d characters: %q", metadataValueLimit, value)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestAnnotationMetadataRoundTripPreservesSlashValues(t *testing.T) {
|
||||||
|
want := taskworker.BackendMetadata(assignment()).Annotations
|
||||||
|
encoded, err := encodeAnnotations(want)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
got, err := decodeAnnotations(encoded)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
for key, value := range want {
|
||||||
|
if got[key] != value {
|
||||||
|
t.Fatalf("%s=%q, want %q", key, got[key], value)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) {
|
func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) {
|
||||||
|
|||||||
@@ -102,7 +102,7 @@ func backendFromTask(task *runnerv1.Task) (Backend, error) {
|
|||||||
return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
|
return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
|
||||||
}
|
}
|
||||||
hasPod := slices.Contains(labels, string(BackendPod))
|
hasPod := slices.Contains(labels, string(BackendPod))
|
||||||
hasVM := slices.Contains(labels, string(BackendVM))
|
hasVM := slices.Contains(labels, string(BackendVM)) || slices.Contains(labels, "vm-dev")
|
||||||
if hasPod == hasVM {
|
if hasPod == hasVM {
|
||||||
return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels)
|
return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ func TestNewSelectsBackendFromRunsOn(t *testing.T) {
|
|||||||
}{
|
}{
|
||||||
{"[self-hosted, pod]", BackendPod},
|
{"[self-hosted, pod]", BackendPod},
|
||||||
{"[self-hosted, vm]", BackendVM},
|
{"[self-hosted, vm]", BackendVM},
|
||||||
|
{"[self-hosted, vm-dev]", BackendVM},
|
||||||
} {
|
} {
|
||||||
assignment, err := New(task(t, test.labels), "ddupan.top")
|
assignment, err := New(task(t, test.labels), "ddupan.top")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -71,6 +71,28 @@ type Worker struct {
|
|||||||
Backend Backend
|
Backend Backend
|
||||||
Tasks TaskState
|
Tasks TaskState
|
||||||
Bootstrap Bootstrap
|
Bootstrap Bootstrap
|
||||||
|
OnEvent func(Event)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Event describes a backend lifecycle transition without exposing launch
|
||||||
|
// environment values or other credentials.
|
||||||
|
type Event struct {
|
||||||
|
Name string
|
||||||
|
AssignmentID string
|
||||||
|
Executor string
|
||||||
|
Phase Phase
|
||||||
|
}
|
||||||
|
|
||||||
|
func (w Worker) event(name string, assignment taskassignment.Assignment, executor *Executor) {
|
||||||
|
if w.OnEvent == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
event := Event{Name: name, AssignmentID: assignment.ID}
|
||||||
|
if executor != nil {
|
||||||
|
event.Executor = executor.Name
|
||||||
|
event.Phase = executor.Phase
|
||||||
|
}
|
||||||
|
w.OnEvent(event)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Accept completes the durable handoff from JetStream to the backend. Once it
|
// Accept completes the durable handoff from JetStream to the backend. Once it
|
||||||
@@ -88,6 +110,7 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment
|
|||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
if executor == nil {
|
if executor == nil {
|
||||||
|
w.event("executor_absent", assignment, nil)
|
||||||
launch, launchErr := w.launchSpec(assignment)
|
launch, launchErr := w.launchSpec(assignment)
|
||||||
if launchErr != nil {
|
if launchErr != nil {
|
||||||
return false, launchErr
|
return false, launchErr
|
||||||
@@ -96,13 +119,18 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
w.event("executor_created", assignment, executor)
|
||||||
|
} else {
|
||||||
|
w.event("executor_found", assignment, executor)
|
||||||
}
|
}
|
||||||
if executor.IdentityTarget == "" {
|
if executor.IdentityTarget == "" {
|
||||||
|
w.event("identity_target_pending", assignment, executor)
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
|
if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
w.event("identity_bound", assignment, executor)
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -178,6 +206,7 @@ func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, er
|
|||||||
func BackendMetadata(assignment taskassignment.Assignment) Metadata {
|
func BackendMetadata(assignment taskassignment.Assignment) Metadata {
|
||||||
return Metadata{
|
return Metadata{
|
||||||
Labels: map[string]string{
|
Labels: map[string]string{
|
||||||
|
"ci.ddupan.top/runner": "true",
|
||||||
"ci.ddupan.top/assignment-id": assignment.ID,
|
"ci.ddupan.top/assignment-id": assignment.ID,
|
||||||
"ci.ddupan.top/task-id": strconv.FormatInt(assignment.Task.GetId(), 10),
|
"ci.ddupan.top/task-id": strconv.FormatInt(assignment.Task.GetId(), 10),
|
||||||
"ci.ddupan.top/backend": string(assignment.Backend),
|
"ci.ddupan.top/backend": string(assignment.Backend),
|
||||||
|
|||||||
@@ -79,7 +79,8 @@ func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
|
|||||||
|
|
||||||
func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
||||||
backend := &fakeBackend{}
|
backend := &fakeBackend{}
|
||||||
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
|
var events []Event
|
||||||
|
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}, OnEvent: func(event Event) { events = append(events, event) }}
|
||||||
|
|
||||||
accepted, err := worker.Accept(context.Background(), assignment())
|
accepted, err := worker.Accept(context.Background(), assignment())
|
||||||
if err != nil || !accepted {
|
if err != nil || !accepted {
|
||||||
@@ -88,6 +89,18 @@ func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
|||||||
if backend.created != 1 || backend.bound != 1 || backend.deleted != 0 {
|
if backend.created != 1 || backend.bound != 1 || backend.deleted != 0 {
|
||||||
t.Fatalf("created=%d bound=%d deleted=%d", backend.created, backend.bound, backend.deleted)
|
t.Fatalf("created=%d bound=%d deleted=%d", backend.created, backend.bound, backend.deleted)
|
||||||
}
|
}
|
||||||
|
want := []string{"executor_absent", "executor_created", "identity_bound"}
|
||||||
|
if len(events) != len(want) {
|
||||||
|
t.Fatalf("events = %#v", events)
|
||||||
|
}
|
||||||
|
for index := range want {
|
||||||
|
if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" {
|
||||||
|
t.Fatalf("event[%d] = %#v", index, events[index])
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if events[1].Executor != "executor" || events[1].Phase != PhaseRunning {
|
||||||
|
t.Fatalf("created event = %#v", events[1])
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) {
|
func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user