Compare commits
27
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0bf39b4751
|
||
|
|
d776fa71e9
|
||
|
|
adb5af1486
|
||
|
|
94fc84a47c | ||
|
|
cc94438bad
|
||
|
|
7d90f28b73 | ||
|
|
8f8ec04b18
|
||
|
|
74ffd49d08 | ||
|
|
e3ff308772
|
||
|
|
feb0b84b7c | ||
|
|
66b90146f8
|
||
|
|
3d8a04e4f7 | ||
|
|
fef7e5a214
|
||
|
|
51b940468e
|
||
|
|
b614ef4c2f
|
||
|
|
8fa8e46320
|
||
|
|
cc4405f788
|
||
|
|
4a428c4384 | ||
|
|
9a28a1573e
|
||
|
|
01995bf084 | ||
|
|
3d787b2dd3
|
||
|
|
54661411e3 | ||
|
|
53a080b310
|
||
|
|
38e8d59541
|
||
|
|
661b5e9218 | ||
|
|
48b6b8038e
|
||
|
|
70c5ff422f |
@@ -1,5 +1,7 @@
|
|||||||
|
---
|
||||||
name: dynamic Pod smoke test
|
name: dynamic Pod smoke test
|
||||||
|
|
||||||
|
# yamllint disable-line rule:truthy
|
||||||
on:
|
on:
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
@@ -17,3 +19,24 @@ jobs:
|
|||||||
-socketPath /run/spire/agent-sockets/spire-agent.sock \
|
-socketPath /run/spire/agent-sockets/spire-agent.sock \
|
||||||
>/dev/null
|
>/dev/null
|
||||||
test "$(id -u)" = 2000
|
test "$(id -u)" = 2000
|
||||||
|
|
||||||
|
- name: Build and run image
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
findmnt /var/lib/docker
|
||||||
|
context=$(mktemp -d)
|
||||||
|
cleanup() {
|
||||||
|
docker image rm --force pod-docker-smoke:test \
|
||||||
|
>/dev/null 2>&1 || true
|
||||||
|
rm -rf -- "$context"
|
||||||
|
}
|
||||||
|
trap cleanup EXIT
|
||||||
|
printf '%s\n' \
|
||||||
|
'FROM alpine:3.22' \
|
||||||
|
'RUN printf pod-docker-ok >/result' \
|
||||||
|
>"$context/Dockerfile"
|
||||||
|
docker build --tag pod-docker-smoke:test "$context"
|
||||||
|
output=$(docker run --rm pod-docker-smoke:test cat /result)
|
||||||
|
test "$output" = pod-docker-ok
|
||||||
|
test "$(docker info --format '{{.Driver}}')" = overlay2
|
||||||
|
|||||||
@@ -34,17 +34,27 @@ jobs:
|
|||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v4
|
- uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- uses: actions/setup-go@v5
|
||||||
|
with:
|
||||||
|
go-version-file: go.mod
|
||||||
|
|
||||||
|
- uses: actions/setup-python@v5
|
||||||
|
with:
|
||||||
|
python-version: '3.12'
|
||||||
|
|
||||||
- name: Test source
|
- name: Test source
|
||||||
shell: bash
|
shell: bash
|
||||||
run: |
|
run: |
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
go test ./...
|
go test ./...
|
||||||
go vet ./...
|
go vet ./...
|
||||||
python3 -m pip install --break-system-packages -e '.[test]'
|
python3 -m venv /tmp/test-venv
|
||||||
|
source /tmp/test-venv/bin/activate
|
||||||
|
python3 -m pip install -e '.[test]'
|
||||||
pytest -q
|
pytest -q
|
||||||
python3 -m compileall -q src tests
|
python3 -m compileall -q src tests
|
||||||
apt-get update
|
sudo apt-get update
|
||||||
apt-get install --yes --no-install-recommends shellcheck
|
sudo apt-get install --yes --no-install-recommends shellcheck
|
||||||
shellcheck scripts/*
|
shellcheck scripts/*
|
||||||
|
|
||||||
- name: Build and publish
|
- name: Build and publish
|
||||||
@@ -53,6 +63,10 @@ jobs:
|
|||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
set +x
|
set +x
|
||||||
|
|
||||||
|
# Bootstrap the image that first introduces automatic Docker setup.
|
||||||
|
# Once deployed, the job-started hook makes this an idempotent no-op.
|
||||||
|
sudo scripts/setup-job-docker
|
||||||
|
|
||||||
: "${GITHUB_SHA:?GITHUB_SHA is required}"
|
: "${GITHUB_SHA:?GITHUB_SHA is required}"
|
||||||
image_tag="sha-${GITHUB_SHA}"
|
image_tag="sha-${GITHUB_SHA}"
|
||||||
docker_config=$(mktemp -d)
|
docker_config=$(mktemp -d)
|
||||||
|
|||||||
@@ -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,5 +1,7 @@
|
|||||||
|
---
|
||||||
name: VM kind smoke
|
name: VM kind smoke
|
||||||
|
|
||||||
|
# yamllint disable-line rule:truthy
|
||||||
on:
|
on:
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
@@ -7,25 +9,123 @@ jobs:
|
|||||||
kind:
|
kind:
|
||||||
runs-on: [self-hosted, vm]
|
runs-on: [self-hosted, vm]
|
||||||
steps:
|
steps:
|
||||||
|
- name: Prepare nested kubelet device
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
if [[ ! -e /dev/kmsg ]]; then
|
||||||
|
sudo mknod /dev/kmsg c 1 11
|
||||||
|
fi
|
||||||
|
|
||||||
- name: Verify Docker
|
- name: Verify Docker
|
||||||
run: docker info
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
findmnt /var/lib/docker
|
||||||
|
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]
|
||||||
|
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
|
||||||
@@ -32,8 +36,9 @@ runs-on: [self-hosted, vm]
|
|||||||
entry;不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见
|
entry;不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见
|
||||||
[`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。
|
[`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。
|
||||||
- Pod executor:在 Kubernetes 中创建一次性 privileged Pod;Pod 内的 workflow 使用
|
- Pod executor:在 Kubernetes 中创建一次性 privileged Pod;Pod 内的 workflow 使用
|
||||||
host executor,Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
|
host executor。Runner 固定在支持原生 job hooks 的 3.x 版本,在 workflow 第一步前
|
||||||
支持原生 job hooks 的 3.x 版本,在 workflow 第一步前等待实际任务对应的 SVID。
|
等待实际任务对应的 SVID,并启动 job-local Docker daemon;workflow 可直接使用与
|
||||||
|
GitHub-hosted runner 相同的 Docker/BuildKit action。
|
||||||
- `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个
|
- `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个
|
||||||
动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
|
动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -29,6 +29,7 @@ COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
|
|||||||
COPY config/runner.yaml /etc/gitea-runner/config.yaml
|
COPY config/runner.yaml /etc/gitea-runner/config.yaml
|
||||||
COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
|
COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
|
||||||
COPY --chmod=0755 scripts/gitea-opensandbox-runner /usr/local/libexec/gitea-opensandbox-runner
|
COPY --chmod=0755 scripts/gitea-opensandbox-runner /usr/local/libexec/gitea-opensandbox-runner
|
||||||
|
COPY --chmod=0755 scripts/setup-job-docker /usr/local/libexec/setup-job-docker
|
||||||
|
|
||||||
VOLUME ["/data"]
|
VOLUME ["/data"]
|
||||||
ENV HOME=/home/runner
|
ENV HOME=/home/runner
|
||||||
|
|||||||
@@ -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 独立运行;job-started hook 在第一步 workflow 之前
|
||||||
|
启动 job-local Docker daemon,业务 workflow 不负责 runner 基础设施初始化;
|
||||||
|
- 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)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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) {
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
|
|
||||||
corev1 "k8s.io/api/core/v1"
|
corev1 "k8s.io/api/core/v1"
|
||||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||||
|
"k8s.io/apimachinery/pkg/api/resource"
|
||||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||||
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
|
||||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||||
@@ -76,17 +77,26 @@ func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, erro
|
|||||||
Containers: []corev1.Container{{
|
Containers: []corev1.Container{{
|
||||||
Name: "executor", Image: manifest.Image, Args: manifest.Args, Env: environment,
|
Name: "executor", Image: manifest.Image, Args: manifest.Args, Env: environment,
|
||||||
SecurityContext: &corev1.SecurityContext{Privileged: boolPointer(true)},
|
SecurityContext: &corev1.SecurityContext{Privileged: boolPointer(true)},
|
||||||
VolumeMounts: []corev1.VolumeMount{{
|
VolumeMounts: []corev1.VolumeMount{
|
||||||
Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true,
|
{Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true},
|
||||||
|
{Name: "docker-data", MountPath: "/var/lib/docker"},
|
||||||
|
},
|
||||||
}},
|
}},
|
||||||
}},
|
Volumes: []corev1.Volume{
|
||||||
Volumes: []corev1.Volume{{
|
{
|
||||||
Name: "spire-agent-socket",
|
Name: "spire-agent-socket",
|
||||||
VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
|
VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
|
||||||
Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
|
Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
|
||||||
}},
|
}},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
Name: "docker-data",
|
||||||
|
VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{
|
||||||
|
SizeLimit: resourceQuantity("20Gi"),
|
||||||
}},
|
}},
|
||||||
},
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
}
|
}
|
||||||
created, err := c.Kubernetes.CoreV1().Pods(manifest.Namespace).Create(ctx, document, metav1.CreateOptions{})
|
created, err := c.Kubernetes.CoreV1().Pods(manifest.Namespace).Create(ctx, document, metav1.CreateOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -169,6 +179,11 @@ func podFromKubernetes(pod corev1.Pod) Pod {
|
|||||||
|
|
||||||
func boolPointer(value bool) *bool { return &value }
|
func boolPointer(value bool) *bool { return &value }
|
||||||
|
|
||||||
|
func resourceQuantity(value string) *resource.Quantity {
|
||||||
|
quantity := resource.MustParse(value)
|
||||||
|
return &quantity
|
||||||
|
}
|
||||||
|
|
||||||
func stringMap(values map[string]string) map[string]any {
|
func stringMap(values map[string]string) map[string]any {
|
||||||
result := make(map[string]any, len(values))
|
result := make(map[string]any, len(values))
|
||||||
for key, value := range values {
|
for key, value := range values {
|
||||||
|
|||||||
@@ -37,6 +37,12 @@ func TestClientPodLifecycleUsesTypedClient(t *testing.T) {
|
|||||||
if got := pod.Spec.Containers[0].Env; len(got) != 1 || got[0].Name != "CI_RUNNER_CAPABILITY" || got[0].Value != "capability" {
|
if got := pod.Spec.Containers[0].Env; len(got) != 1 || got[0].Name != "CI_RUNNER_CAPABILITY" || got[0].Value != "capability" {
|
||||||
t.Fatalf("environment = %#v", got)
|
t.Fatalf("environment = %#v", got)
|
||||||
}
|
}
|
||||||
|
if got := pod.Spec.Containers[0].VolumeMounts; len(got) != 2 || got[1].Name != "docker-data" || got[1].MountPath != "/var/lib/docker" {
|
||||||
|
t.Fatalf("volume mounts = %#v", got)
|
||||||
|
}
|
||||||
|
if got := pod.Spec.Volumes; len(got) != 2 || got[1].EmptyDir == nil || got[1].EmptyDir.SizeLimit == nil || got[1].EmptyDir.SizeLimit.String() != "20Gi" {
|
||||||
|
t.Fatalf("volumes = %#v", got)
|
||||||
|
}
|
||||||
if err := client.LabelPod(context.Background(), "gitea-actions", created.Name, map[string]string{terminalLabel: "true"}); err != nil {
|
if err := client.LabelPod(context.Background(), "gitea-actions", created.Name, map[string]string{terminalLabel: "true"}); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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) {
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ while [ "$(date +%s)" -lt "$deadline" ]; do
|
|||||||
-audience ci-job-ready \
|
-audience ci-job-ready \
|
||||||
-socketPath "$socket" \
|
-socketPath "$socket" \
|
||||||
>/dev/null 2>&1; then
|
>/dev/null 2>&1; then
|
||||||
|
/usr/local/libexec/setup-job-docker
|
||||||
exit 0
|
exit 0
|
||||||
fi
|
fi
|
||||||
sleep 1
|
sleep 1
|
||||||
|
|||||||
Executable
+60
@@ -0,0 +1,60 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
set -euo pipefail
|
||||||
|
|
||||||
|
if docker info >/dev/null 2>&1; then
|
||||||
|
printf '%s\n' 'job Docker daemon is already ready'
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
|
||||||
|
storage_size=${DOCKER_DATA_SIZE:-20G}
|
||||||
|
storage_driver=${DOCKER_STORAGE_DRIVER:-overlay2}
|
||||||
|
wait_seconds=${DOCKER_START_WAIT_SECONDS:-60}
|
||||||
|
|
||||||
|
sudo install -d /var/lib/docker
|
||||||
|
if ! mountpoint --quiet /var/lib/docker; then
|
||||||
|
printf '%s\n' "preparing ${storage_size} loop-backed Docker storage"
|
||||||
|
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
|
||||||
|
sudo truncate -s "$storage_size" /tmp/docker-data.img
|
||||||
|
sudo mkfs.ext4 -F /tmp/docker-data.img
|
||||||
|
loop_device=$(sudo losetup --find --show /tmp/docker-data.img)
|
||||||
|
sudo mount "$loop_device" /var/lib/docker
|
||||||
|
else
|
||||||
|
printf '%s\n' 'using mounted Docker storage at /var/lib/docker'
|
||||||
|
fi
|
||||||
|
|
||||||
|
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 sh -c 'exec nohup dockerd "$@" >/tmp/dockerd.log 2>&1' sh \
|
||||||
|
--host=unix:///var/run/docker.sock \
|
||||||
|
--storage-driver="$storage_driver" &
|
||||||
|
printf '%s\n' 'waiting for job Docker daemon'
|
||||||
|
for ((attempt = 0; attempt < wait_seconds; attempt++)); do
|
||||||
|
if docker info >/dev/null 2>&1; then
|
||||||
|
findmnt /var/lib/docker
|
||||||
|
docker info --format \
|
||||||
|
'{{json .ServerVersion}} {{json .Driver}} {{json .CgroupVersion}}'
|
||||||
|
exit 0
|
||||||
|
fi
|
||||||
|
sleep 1
|
||||||
|
done
|
||||||
|
|
||||||
|
cat /tmp/dockerd.log
|
||||||
|
exit 1
|
||||||
Reference in New Issue
Block a user