Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c1d0f4cec2
|
@@ -1,7 +1,5 @@
|
|||||||
---
|
|
||||||
name: dynamic Pod smoke test
|
name: dynamic Pod smoke test
|
||||||
|
|
||||||
# yamllint disable-line rule:truthy
|
|
||||||
on:
|
on:
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
@@ -19,24 +17,3 @@ 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
|
|
||||||
|
|||||||
@@ -1,36 +1,128 @@
|
|||||||
---
|
---
|
||||||
name: publish controller image
|
name: publish images
|
||||||
|
|
||||||
on:
|
on:
|
||||||
push:
|
push:
|
||||||
branches: [main]
|
branches: [main]
|
||||||
paths:
|
paths:
|
||||||
- '.gitea/workflows/publish-images.yml'
|
- '.gitea/workflows/publish-images.yml'
|
||||||
- 'container/controller.Dockerfile'
|
- 'config/**'
|
||||||
|
- 'container/**'
|
||||||
- 'cmd/**'
|
- 'cmd/**'
|
||||||
- 'internal/**'
|
- 'internal/**'
|
||||||
- 'scripts/publish-image'
|
- 'scripts/**'
|
||||||
|
- 'src/**'
|
||||||
- 'go.mod'
|
- 'go.mod'
|
||||||
- 'go.sum'
|
- 'go.sum'
|
||||||
|
- 'pyproject.toml'
|
||||||
|
- 'README.md'
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
publish-images:
|
publish-images:
|
||||||
name: publish-controller
|
name: publish-images
|
||||||
runs-on: [self-hosted, pod]
|
runs-on: [self-hosted, vm]
|
||||||
timeout-minutes: 45
|
timeout-minutes: 45
|
||||||
permissions:
|
permissions:
|
||||||
contents: read
|
contents: read
|
||||||
env:
|
env:
|
||||||
PUSH_REGISTRY: zot-push.ad.ddupan.top
|
PUSH_REGISTRY: zot-push.ad.ddupan.top
|
||||||
PULL_REGISTRY: zot.ad.ddupan.top
|
PULL_REGISTRY: zot.ad.ddupan.top
|
||||||
IMAGE_NAME: controller
|
CONTROLLER_REPOSITORY: panxiao81/gitea-dynamic-runner-controller
|
||||||
IMAGE_REPOSITORY: panxiao81/gitea-dynamic-runner-controller
|
RUNNER_REPOSITORY: panxiao81/gitea-dynamic-runner-runner
|
||||||
IMAGE_DOCKERFILE: container/controller.Dockerfile
|
|
||||||
SPIRE_AGENT_SOCKET: /run/spire/agent-sockets/spire-agent.sock
|
SPIRE_AGENT_SOCKET: /run/spire/agent-sockets/spire-agent.sock
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v4
|
- uses: actions/checkout@v4
|
||||||
|
|
||||||
|
- name: Test source
|
||||||
|
shell: bash
|
||||||
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
go test ./...
|
||||||
|
go vet ./...
|
||||||
|
python3 -m pip install --break-system-packages -e '.[test]'
|
||||||
|
pytest -q
|
||||||
|
python3 -m compileall -q src tests
|
||||||
|
apt-get update
|
||||||
|
apt-get install --yes --no-install-recommends shellcheck
|
||||||
|
shellcheck scripts/*
|
||||||
|
|
||||||
- name: Build and publish
|
- name: Build and publish
|
||||||
shell: bash
|
shell: bash
|
||||||
run: scripts/publish-image
|
run: |
|
||||||
|
set -euo pipefail
|
||||||
|
set +x
|
||||||
|
|
||||||
|
: "${GITHUB_SHA:?GITHUB_SHA is required}"
|
||||||
|
image_tag="sha-${GITHUB_SHA}"
|
||||||
|
docker_config=$(mktemp -d)
|
||||||
|
jwt_file=$(mktemp)
|
||||||
|
cleanup() {
|
||||||
|
docker buildx rm ci-builder >/dev/null 2>&1 || true
|
||||||
|
rm -rf -- "$docker_config" "$jwt_file"
|
||||||
|
}
|
||||||
|
trap cleanup EXIT
|
||||||
|
export DOCKER_CONFIG="$docker_config"
|
||||||
|
|
||||||
|
/opt/spire/bin/spire-agent api fetch jwt \
|
||||||
|
-audience zot \
|
||||||
|
-socketPath "$SPIRE_AGENT_SOCKET" \
|
||||||
|
-output json >"$jwt_file"
|
||||||
|
# shellcheck disable=SC2016
|
||||||
|
jq -er '.[0].svids[0].svid' "$jwt_file" | \
|
||||||
|
docker login "$PUSH_REGISTRY" --username zot --password-stdin
|
||||||
|
|
||||||
|
docker buildx create \
|
||||||
|
--name ci-builder \
|
||||||
|
--driver docker-container \
|
||||||
|
--use
|
||||||
|
|
||||||
|
publish() {
|
||||||
|
local repository=$1
|
||||||
|
local dockerfile=$2
|
||||||
|
local metadata=$3
|
||||||
|
docker buildx build \
|
||||||
|
--builder ci-builder \
|
||||||
|
--platform linux/amd64 \
|
||||||
|
--file "$dockerfile" \
|
||||||
|
--tag "${PUSH_REGISTRY}/${repository}:${image_tag}" \
|
||||||
|
--tag "${PUSH_REGISTRY}/${repository}:main" \
|
||||||
|
--provenance=mode=max \
|
||||||
|
--sbom=true \
|
||||||
|
--metadata-file "$metadata" \
|
||||||
|
--push \
|
||||||
|
.
|
||||||
|
}
|
||||||
|
|
||||||
|
publish \
|
||||||
|
"$CONTROLLER_REPOSITORY" \
|
||||||
|
container/controller.Dockerfile \
|
||||||
|
controller-metadata.json
|
||||||
|
publish \
|
||||||
|
"$RUNNER_REPOSITORY" \
|
||||||
|
container/runner.Dockerfile \
|
||||||
|
runner-metadata.json
|
||||||
|
|
||||||
|
controller_digest=$(
|
||||||
|
# shellcheck disable=SC2016
|
||||||
|
jq -er '."containerimage.digest"' controller-metadata.json
|
||||||
|
)
|
||||||
|
runner_digest=$(
|
||||||
|
# shellcheck disable=SC2016
|
||||||
|
jq -er '."containerimage.digest"' runner-metadata.json
|
||||||
|
)
|
||||||
|
controller_ref="${PULL_REGISTRY}/${CONTROLLER_REPOSITORY}@${controller_digest}"
|
||||||
|
runner_ref="${PULL_REGISTRY}/${RUNNER_REPOSITORY}@${runner_digest}"
|
||||||
|
|
||||||
|
printf 'controller=%s\nrunner=%s\n' "$controller_ref" "$runner_ref"
|
||||||
|
if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then
|
||||||
|
{
|
||||||
|
printf '## Published images\n\n'
|
||||||
|
# shellcheck disable=SC2016
|
||||||
|
printf -- '- Controller: `%s`\n' "$controller_ref"
|
||||||
|
# shellcheck disable=SC2016
|
||||||
|
printf -- '- Runner: `%s`\n' "$runner_ref"
|
||||||
|
# shellcheck disable=SC2016
|
||||||
|
printf -- '- Source: `%s`\n' "$GITHUB_SHA"
|
||||||
|
} >>"$GITHUB_STEP_SUMMARY"
|
||||||
|
fi
|
||||||
|
|||||||
@@ -1,26 +0,0 @@
|
|||||||
---
|
|
||||||
name: publish runner image
|
|
||||||
|
|
||||||
on:
|
|
||||||
workflow_dispatch:
|
|
||||||
|
|
||||||
jobs:
|
|
||||||
publish-images:
|
|
||||||
name: publish-runner
|
|
||||||
runs-on: [self-hosted, pod]
|
|
||||||
timeout-minutes: 45
|
|
||||||
permissions:
|
|
||||||
contents: read
|
|
||||||
env:
|
|
||||||
PUSH_REGISTRY: zot-push.ad.ddupan.top
|
|
||||||
PULL_REGISTRY: zot.ad.ddupan.top
|
|
||||||
IMAGE_NAME: runner
|
|
||||||
IMAGE_REPOSITORY: panxiao81/gitea-dynamic-runner-runner
|
|
||||||
IMAGE_DOCKERFILE: container/runner.Dockerfile
|
|
||||||
SPIRE_AGENT_SOCKET: /run/spire/agent-sockets/spire-agent.sock
|
|
||||||
steps:
|
|
||||||
- uses: actions/checkout@v4
|
|
||||||
|
|
||||||
- name: Build and publish
|
|
||||||
shell: bash
|
|
||||||
run: scripts/publish-image
|
|
||||||
@@ -5,20 +5,29 @@ on:
|
|||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
jwt-svid:
|
jwt-svid:
|
||||||
runs-on: [self-hosted, pod]
|
runs-on: self-hosted
|
||||||
|
container:
|
||||||
|
volumes:
|
||||||
|
- /run/spire/agent-sockets:/run/spire/agent-sockets:ro
|
||||||
steps:
|
steps:
|
||||||
- name: Verify bundled SPIRE CLI
|
- name: Fetch pinned SPIRE CLI
|
||||||
shell: bash
|
shell: bash
|
||||||
run: |
|
run: |
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
command -v spire-agent
|
archive=/tmp/spire.tar.gz
|
||||||
spire-agent -version
|
curl --fail --location --silent --show-error \
|
||||||
|
--output "$archive" \
|
||||||
|
https://github.com/spiffe/spire/releases/download/v1.15.3/spire-1.15.3-linux-amd64-musl.tar.gz
|
||||||
|
printf '%s %s\n' \
|
||||||
|
ca1a4d1155317bdd2afc7f36663828a10410c7c840e54725b90b4064b0a301c7 \
|
||||||
|
"$archive" | sha256sum --check --status
|
||||||
|
tar -xzf "$archive" -C /tmp spire-1.15.3/bin/spire-agent
|
||||||
|
|
||||||
- name: Fetch short-lived zot JWT-SVID
|
- name: Fetch short-lived zot JWT-SVID
|
||||||
shell: bash
|
shell: bash
|
||||||
run: |
|
run: |
|
||||||
set -euo pipefail
|
set -euo pipefail
|
||||||
spire-agent api fetch jwt \
|
/tmp/spire-1.15.3/bin/spire-agent api fetch jwt \
|
||||||
-audience zot \
|
-audience zot \
|
||||||
-socketPath /run/spire/agent-sockets/spire-agent.sock \
|
-socketPath /run/spire/agent-sockets/spire-agent.sock \
|
||||||
>/dev/null
|
>/dev/null
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ jobs:
|
|||||||
- run: go vet ./...
|
- run: go vet ./...
|
||||||
|
|
||||||
python:
|
python:
|
||||||
runs-on: [self-hosted, pod]
|
runs-on: self-hosted
|
||||||
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, pod]
|
runs-on: self-hosted
|
||||||
steps:
|
steps:
|
||||||
- uses: actions/checkout@v4
|
- uses: actions/checkout@v4
|
||||||
- run: |
|
- run: |
|
||||||
|
|||||||
@@ -1,7 +1,5 @@
|
|||||||
---
|
|
||||||
name: VM kind smoke
|
name: VM kind smoke
|
||||||
|
|
||||||
# yamllint disable-line rule:truthy
|
|
||||||
on:
|
on:
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
|
|
||||||
@@ -9,123 +7,25 @@ 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
|
||||||
shell: bash
|
run: docker info
|
||||||
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 "${base_url}/kind-linux-amd64"
|
--output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64"
|
||||||
curl --fail --location --silent --show-error \
|
curl --fail --location --silent --show-error \
|
||||||
--output /tmp/kind.sha256sum \
|
--output /tmp/kind.sha256sum "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64.sha256sum"
|
||||||
"${base_url}/kind-linux-amd64.sha256sum"
|
printf '%s %s\n' "$(cut -d ' ' -f1 /tmp/kind.sha256sum)" /tmp/kind | sha256sum --check
|
||||||
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
|
||||||
diagnose_and_cleanup() {
|
trap '/tmp/kind delete cluster --name smoke' EXIT
|
||||||
status=$?
|
/tmp/kind create cluster --name smoke --wait 180s
|
||||||
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
|
||||||
|
|||||||
@@ -1,36 +0,0 @@
|
|||||||
---
|
|
||||||
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
|
|
||||||
@@ -1,33 +1,27 @@
|
|||||||
# Gitea dynamic runner
|
# Gitea dynamic runner
|
||||||
|
|
||||||
为 Gitea Actions 按需创建一次性执行环境。workflow 分别声明 workload class 与
|
为 Gitea Actions 按需创建一次性执行环境。对 workflow 提供两种稳定的 runner
|
||||||
placement driver:
|
接口:
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
runs-on: [self-hosted, container, kubernetes]
|
runs-on: [self-hosted, pod]
|
||||||
```
|
```
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
runs-on: [self-hosted, vm, opensandbox]
|
runs-on: [self-hosted, vm]
|
||||||
```
|
```
|
||||||
|
|
||||||
兼容标签 `[self-hosted, pod]` 严格映射为 `container+kubernetes`,
|
`pod` 使用动态 Kubernetes Pod,`vm` 使用动态 Cloud Hypervisor microVM。每个环境
|
||||||
`[self-hosted, vm]` 和 `vm-dev` 严格映射为 `vm+opensandbox`;显式 driver 容量耗尽时
|
|
||||||
不会回退到其他 driver。每个环境
|
|
||||||
只执行一个 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
|
||||||
持久化到 JetStream;一个 registration 下按配置启动多个并发 `FetchTask` goroutine,
|
持久化到 JetStream;一个 registration 下按配置启动多个并发 `FetchTask` goroutine,
|
||||||
同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
|
同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
|
||||||
- `kubernetes-worker`:直接在 homelab Kubernetes 创建一次性 container workload。
|
- `pod-worker`:直接在 homelab Kubernetes 创建一次性 Pod。
|
||||||
- `opensandbox-worker`:通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 VM workload。
|
- `vm-worker`:通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。
|
||||||
- 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker,因为 facade
|
- 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker,因为 facade
|
||||||
的 assignment claim registry 仍是进程内状态;支持安全拆分前进程会明确拒绝该配置。
|
的 assignment claim registry 仍是进程内状态;支持安全拆分前进程会明确拒绝该配置。
|
||||||
- `microvm-runner-launch`:为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP,运行
|
- `microvm-runner-launch`:为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP,运行
|
||||||
@@ -38,14 +32,13 @@ runs-on: [self-hosted, vm, opensandbox]
|
|||||||
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。Runner 固定在支持原生 job hooks 的 3.x 版本,在 workflow 第一步前
|
host executor,Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
|
||||||
等待实际任务对应的 SVID,并启动 job-local Docker daemon;workflow 可直接使用与
|
支持原生 job hooks 的 3.x 版本,在 workflow 第一步前等待实际任务对应的 SVID。
|
||||||
GitHub-hosted runner 相同的 Docker/BuildKit action。
|
|
||||||
- `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个
|
- `jwt-broker`:早期共享 Kubernetes runner 的过渡实验;目标架构不部署它,每个
|
||||||
动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
|
动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
|
||||||
|
|
||||||
Container/Kubernetes 路径由 homelab 集群中的 `kubernetes-worker` 直接创建 Pod。
|
Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于
|
||||||
OpenSandbox 只用于 VM/Kata workload;每个 placement 使用独立 durable consumer 和容量池。assignment 根据
|
VM/Kata workload;两个 backend 使用独立 durable consumer 和独立容量池。assignment 根据
|
||||||
`runs-on` 进入对应池,池满时留在 JetStream pending,不会创建超出容量的 workload;任一
|
`runs-on` 进入对应池,池满时留在 JetStream pending,不会创建超出容量的 workload;任一
|
||||||
执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见
|
执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见
|
||||||
[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
|
[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
|
||||||
@@ -72,13 +65,12 @@ credential 都从挂载文件读取,不接受明文环境变量:
|
|||||||
- `NATS_PRODUCER_PASSWORD_FILE`、`NATS_WORKER_PASSWORD_FILE`:分别使用现有最小权限的
|
- `NATS_PRODUCER_PASSWORD_FILE`、`NATS_WORKER_PASSWORD_FILE`:分别使用现有最小权限的
|
||||||
`ci-producer` publish 连接和 `ci-worker` pull/ACK 连接,controller 不合并权限。
|
`ci-producer` publish 连接和 `ci-worker` pull/ACK 连接,controller 不合并权限。
|
||||||
- `RUNNER_FACADE_CAPABILITY_KEY_FILE`:至少 32 字节的 controller HMAC key。
|
- `RUNNER_FACADE_CAPABILITY_KEY_FILE`:至少 32 字节的 controller HMAC key。
|
||||||
- `OPENSANDBOX_API_KEY_FILE`:仅启用 `opensandbox-worker` 时读取。
|
- `OPENSANDBOX_API_KEY_FILE`:仅启用 `vm-worker` 时读取。
|
||||||
|
|
||||||
必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`(应使用 digest)、`SPIRE_AGENT_ID`、
|
必要的非 secret 配置包括 `POD_EXECUTOR_IMAGE`(应使用 digest)、`SPIRE_AGENT_ID`、
|
||||||
`RUNNER_FACADE_URL`、`RUNNER_FACADE_SPIFFE_ID` 和 `SPIFFE_ENDPOINT_SOCKET`。默认
|
`RUNNER_FACADE_URL`、`RUNNER_FACADE_SPIFFE_ID` 和 `SPIFFE_ENDPOINT_SOCKET`。默认
|
||||||
`COMPONENTS=all`、Pod 并发 4、VM 并发 1;首次 smoke test 应显式设为
|
`COMPONENTS=all`、Pod 并发 4、VM 并发 1;首次 smoke test 应显式设为
|
||||||
`COMPONENTS=scheduler,kubernetes-worker`,先验证 Kubernetes 链路,避免同时消耗 VM
|
`COMPONENTS=scheduler,pod-worker`,先验证 Pod 链路,避免同时消耗 VM 容量。
|
||||||
容量。旧组件名 `pod-worker`、`vm-worker` 只作为配置兼容别名保留。
|
|
||||||
|
|
||||||
Pod task 的 terminal update 被 Gitea 接受后,controller 会在 Pod 上持久写入
|
Pod task 的 terminal update 被 Gitea 接受后,controller 会在 Pod 上持久写入
|
||||||
`ci.ddupan.top/terminal=true` label。生命周期 reconciler 只清理同时带该 label 且已经
|
`ci.ddupan.top/terminal=true` label。生命周期 reconciler 只清理同时带该 label 且已经
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
"slices"
|
"slices"
|
||||||
@@ -38,15 +38,6 @@ type runComponent func(context.Context) error
|
|||||||
|
|
||||||
func (function runComponent) Run(ctx context.Context) error { return function(ctx) }
|
func (function runComponent) Run(ctx context.Context) error { return function(ctx) }
|
||||||
|
|
||||||
type terminalBackend interface {
|
|
||||||
MarkTerminal(context.Context, string) error
|
|
||||||
}
|
|
||||||
|
|
||||||
type placementRuntime struct {
|
|
||||||
backend terminalBackend
|
|
||||||
pool *backendpool.Pool
|
|
||||||
}
|
|
||||||
|
|
||||||
type controllerConfig struct {
|
type controllerConfig struct {
|
||||||
Components controller.Selection
|
Components controller.Selection
|
||||||
TrustDomain, WorkloadAPIAddr string
|
TrustDomain, WorkloadAPIAddr string
|
||||||
@@ -58,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, VMRunnerLabel string
|
OpenSandboxURL, OpenSandboxAPIKey, OpenSandboxPool string
|
||||||
VMTimeout, VMCapacity int
|
VMTimeout, VMCapacity int
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -70,7 +61,7 @@ func runController(ctx context.Context) error {
|
|||||||
if !slices.Contains(config.Components, controller.Scheduler) {
|
if !slices.Contains(config.Components, controller.Scheduler) {
|
||||||
return errors.New("split worker deployment is not yet safe: scheduler/facade must be enabled with workers")
|
return errors.New("split worker deployment is not yet safe: scheduler/facade must be enabled with workers")
|
||||||
}
|
}
|
||||||
if !slices.Contains(config.Components, controller.KubernetesWorker) && !slices.Contains(config.Components, controller.OpenSandboxWorker) {
|
if !slices.Contains(config.Components, controller.PodWorker) && !slices.Contains(config.Components, controller.VMWorker) {
|
||||||
return errors.New("scheduler requires at least one local backend worker")
|
return errors.New("scheduler requires at least one local backend worker")
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -101,19 +92,32 @@ func runController(ctx context.Context) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
registry := runnerfacade.NewRegistry()
|
registry := runnerfacade.NewRegistry()
|
||||||
runtimes := make(map[taskassignment.Placement]placementRuntime)
|
podPool := backendpool.New(config.PodCapacity)
|
||||||
|
vmPool := backendpool.New(config.VMCapacity)
|
||||||
|
var podExecutorBackend *podbackend.Backend
|
||||||
|
var vmExecutorBackend *opensandboxbackend.Backend
|
||||||
giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken)
|
giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken)
|
||||||
facade := &runnerfacade.Facade{
|
facade := &runnerfacade.Facade{
|
||||||
Registry: registry, Capabilities: capabilities, Upstream: giteaClient,
|
Registry: registry, Capabilities: capabilities, Upstream: giteaClient,
|
||||||
OnTerminal: func(ctx context.Context, assignment taskassignment.Assignment) error {
|
OnTerminal: func(ctx context.Context, assignment taskassignment.Assignment) error {
|
||||||
runtime, ok := runtimes[assignment.Placement]
|
switch assignment.Backend {
|
||||||
if !ok {
|
case taskassignment.BackendPod:
|
||||||
return fmt.Errorf("placement runtime %s is not configured", assignment.Placement.Key())
|
if podExecutorBackend == nil {
|
||||||
|
return errors.New("Pod lifecycle backend is not configured")
|
||||||
|
}
|
||||||
|
if err := podExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
podPool.Release(assignment.ID)
|
||||||
|
case taskassignment.BackendVM:
|
||||||
|
if vmExecutorBackend == nil {
|
||||||
|
return errors.New("VM lifecycle backend is not configured")
|
||||||
|
}
|
||||||
|
if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
vmPool.Release(assignment.ID)
|
||||||
}
|
}
|
||||||
if err := runtime.backend.MarkTerminal(ctx, assignment.ID); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
runtime.pool.Release(assignment.ID)
|
|
||||||
return nil
|
return nil
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
@@ -123,11 +127,11 @@ func runController(ctx context.Context) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
labels := []string{"self-hosted"}
|
labels := []string{"self-hosted"}
|
||||||
if slices.Contains(config.Components, controller.KubernetesWorker) {
|
if slices.Contains(config.Components, controller.PodWorker) {
|
||||||
labels = append(labels, "pod", string(taskassignment.WorkloadContainer), string(taskassignment.DriverKubernetes))
|
labels = append(labels, string(taskassignment.BackendPod))
|
||||||
}
|
}
|
||||||
if slices.Contains(config.Components, controller.OpenSandboxWorker) {
|
if slices.Contains(config.Components, controller.VMWorker) {
|
||||||
labels = append(labels, config.VMRunnerLabel, string(taskassignment.DriverOpenSandbox))
|
labels = append(labels, string(taskassignment.BackendVM))
|
||||||
}
|
}
|
||||||
poller := taskscheduler.Poller{
|
poller := taskscheduler.Poller{
|
||||||
Client: giteaClient,
|
Client: giteaClient,
|
||||||
@@ -135,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) { slog.Error("scheduler error", "component", "scheduler", "error", err) },
|
OnError: func(err error) { log.Printf("scheduler: %v", err) },
|
||||||
}
|
}
|
||||||
kubernetesConfig, err := rest.InClusterConfig()
|
kubernetesConfig, err := rest.InClusterConfig()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -164,9 +168,7 @@ func runController(ctx context.Context) error {
|
|||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
|
|
||||||
if slices.Contains(config.Components, controller.KubernetesWorker) {
|
if slices.Contains(config.Components, controller.PodWorker) {
|
||||||
placement := taskassignment.KubernetesContainer
|
|
||||||
pool := backendpool.New(config.PodCapacity)
|
|
||||||
client, err := podbackend.NewInClusterClient()
|
client, err := podbackend.NewInClusterClient()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -177,59 +179,53 @@ func runController(ctx context.Context) error {
|
|||||||
SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass,
|
SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass,
|
||||||
SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID,
|
SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID,
|
||||||
}}
|
}}
|
||||||
runtimes[placement] = placementRuntime{backend: backend, pool: pool}
|
podExecutorBackend = &backend
|
||||||
assignments, err := backend.RecoverAssignments(ctx)
|
assignments, err := backend.RecoverAssignments(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
for _, assignment := range assignments {
|
for _, assignment := range assignments {
|
||||||
pool.Restore(assignment.ID)
|
podPool.Restore(assignment.ID)
|
||||||
if err := registry.RecoverClaimed(assignment); err != nil {
|
if err := registry.RecoverClaimed(assignment); err != nil {
|
||||||
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, placement, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(placement)}, registry, pool)
|
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, podPool)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) {
|
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }}
|
||||||
slog.Error("backend lifecycle error", "component", "lifecycle", "placement", placement.Key(), "error", err)
|
components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
|
||||||
}}
|
|
||||||
components[controller.KubernetesWorker] = 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) })
|
||||||
group.Go(func() error { return lifecycle.Run(groupContext) })
|
group.Go(func() error { return lifecycle.Run(groupContext) })
|
||||||
return group.Wait()
|
return group.Wait()
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
if slices.Contains(config.Components, controller.OpenSandboxWorker) {
|
if slices.Contains(config.Components, controller.VMWorker) {
|
||||||
placement := taskassignment.OpenSandboxVM
|
|
||||||
pool := backendpool.New(config.VMCapacity)
|
|
||||||
lifecycle := opensandboxbackend.NewLifecycleClient(config.OpenSandboxURL, config.OpenSandboxAPIKey, &http.Client{Timeout: 60 * time.Second})
|
lifecycle := opensandboxbackend.NewLifecycleClient(config.OpenSandboxURL, config.OpenSandboxAPIKey, &http.Client{Timeout: 60 * time.Second})
|
||||||
backend := opensandboxbackend.Backend{Lifecycle: lifecycle, Config: opensandboxbackend.Config{
|
backend := opensandboxbackend.Backend{Lifecycle: lifecycle, Config: opensandboxbackend.Config{
|
||||||
Pool: config.OpenSandboxPool, Timeout: config.VMTimeout,
|
Pool: config.OpenSandboxPool, Timeout: config.VMTimeout,
|
||||||
Entrypoint: []string{"/usr/local/bin/gitea-dynamic-runner", "executor"},
|
Entrypoint: []string{"/usr/local/bin/gitea-dynamic-runner", "executor"},
|
||||||
Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
|
Env: map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
|
||||||
}}
|
}}
|
||||||
runtimes[placement] = placementRuntime{backend: backend, pool: pool}
|
vmExecutorBackend = &backend
|
||||||
assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain)
|
assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
for _, assignment := range assignments {
|
for _, assignment := range assignments {
|
||||||
pool.Restore(assignment.ID)
|
vmPool.Restore(assignment.ID)
|
||||||
if err := registry.RecoverClaimed(assignment); err != nil {
|
if err := registry.RecoverClaimed(assignment); err != nil {
|
||||||
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, placement, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(placement)}, registry, pool)
|
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, vmPool)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) {
|
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { log.Printf("VM lifecycle: %v", err) }}
|
||||||
slog.Error("backend lifecycle error", "component", "lifecycle", "placement", placement.Key(), "error", err)
|
components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
|
||||||
}}
|
|
||||||
components[controller.OpenSandboxWorker] = 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) })
|
||||||
group.Go(func() error { return lifecycleReconciler.Run(groupContext) })
|
group.Go(func() error { return lifecycleReconciler.Run(groupContext) })
|
||||||
@@ -291,28 +287,18 @@ func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn,
|
|||||||
return nats.Connect(server, options...)
|
return nats.Connect(server, options...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, placement taskassignment.Placement, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) {
|
func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) {
|
||||||
consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, placement, capacity)
|
consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
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, OnEvent: func(event assignmentqueue.Event) {
|
Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission},
|
||||||
slog.Info("assignment transition", "component", "worker", "event", event.Name, "placement", event.Placement.Key(), "assignment", event.AssignmentID, "retry_delay", event.RetryDelay)
|
OnError: func(err error) { log.Printf("%s worker: %v", backend, err) },
|
||||||
}},
|
|
||||||
OnError: func(err error) {
|
|
||||||
slog.Error("assignment processing error", "component", "worker", "placement", placement.Key(), "error", err)
|
|
||||||
},
|
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func workerEventLogger(placement taskassignment.Placement) func(taskworker.Event) {
|
|
||||||
return func(event taskworker.Event) {
|
|
||||||
slog.Info("executor transition", "component", "worker", "event", event.Name, "placement", placement.Key(), "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 {
|
||||||
@@ -359,23 +345,20 @@ 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"), VMRunnerLabel: env("VM_RUNNER_LABEL", "vm"), VMTimeout: envInt("VM_TIMEOUT_SECONDS", 14400), VMCapacity: envInt("VM_CAPACITY", 1),
|
OpenSandboxURL: os.Getenv("OPENSANDBOX_API"), OpenSandboxPool: env("OPENSANDBOX_POOL", "ci-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")
|
||||||
}
|
}
|
||||||
if slices.Contains(selection, controller.KubernetesWorker) && config.PodImage == "" {
|
if slices.Contains(selection, controller.PodWorker) && config.PodImage == "" {
|
||||||
return controllerConfig{}, errors.New("POD_EXECUTOR_IMAGE is required for kubernetes-worker")
|
return controllerConfig{}, errors.New("POD_EXECUTOR_IMAGE is required for pod-worker")
|
||||||
}
|
}
|
||||||
if slices.Contains(selection, controller.KubernetesWorker) && config.SPIREAgentID == "" {
|
if slices.Contains(selection, controller.PodWorker) && config.SPIREAgentID == "" {
|
||||||
return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for kubernetes-worker")
|
return controllerConfig{}, errors.New("SPIRE_AGENT_ID is required for pod-worker")
|
||||||
}
|
}
|
||||||
if slices.Contains(selection, controller.OpenSandboxWorker) {
|
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 opensandbox-worker")
|
return controllerConfig{}, errors.New("OPENSANDBOX_API is required for vm-worker")
|
||||||
}
|
}
|
||||||
config.OpenSandboxAPIKey, err = read("OPENSANDBOX_API_KEY_FILE")
|
config.OpenSandboxAPIKey, err = read("OPENSANDBOX_API_KEY_FILE")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -34,7 +34,7 @@ func TestLoadControllerConfigUsesFileSecrets(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.KubernetesWorker {
|
if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.PodWorker {
|
||||||
t.Fatalf("components = %#v", config.Components)
|
t.Fatalf("components = %#v", config.Components)
|
||||||
}
|
}
|
||||||
if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSProducerPassword != "producer-password" || config.NATSWorkerPassword != "worker-password" {
|
if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSProducerPassword != "producer-password" || config.NATSWorkerPassword != "worker-password" {
|
||||||
@@ -56,7 +56,6 @@ 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,7 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
|
||||||
"os"
|
"os"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"syscall"
|
"syscall"
|
||||||
@@ -13,9 +12,8 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, nil)))
|
|
||||||
if err := run(); err != nil {
|
if err := run(); err != nil {
|
||||||
slog.Error("runner stopped", "error", err)
|
fmt.Fprintln(os.Stderr, err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -21,16 +21,14 @@ RUN groupadd --gid 2000 runner \
|
|||||||
&& useradd --uid 2000 --gid 2000 --groups docker --create-home --shell /bin/bash runner \
|
&& useradd --uid 2000 --gid 2000 --groups docker --create-home --shell /bin/bash runner \
|
||||||
&& printf 'runner ALL=(ALL) NOPASSWD:ALL\n' >/etc/sudoers.d/runner \
|
&& printf 'runner ALL=(ALL) NOPASSWD:ALL\n' >/etc/sudoers.d/runner \
|
||||||
&& chmod 0440 /etc/sudoers.d/runner \
|
&& chmod 0440 /etc/sudoers.d/runner \
|
||||||
&& install -d -o 2000 -g 2000 /data /workspace
|
&& install -d -o 2000 -g 2000 /data
|
||||||
|
|
||||||
COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
|
COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
|
||||||
COPY --from=controller /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner
|
COPY --from=controller /out/gitea-dynamic-runner /usr/local/bin/gitea-dynamic-runner
|
||||||
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
|
COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
|
||||||
RUN ln -s /opt/spire/bin/spire-agent /usr/local/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
|
||||||
|
|||||||
@@ -2,22 +2,19 @@
|
|||||||
|
|
||||||
## 对 workflow 的接口
|
## 对 workflow 的接口
|
||||||
|
|
||||||
Runner 将执行环境的能力类型与实现 driver 分开声明:
|
Runner 只向 workflow 暴露两个执行环境:
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
runs-on: [self-hosted, container, kubernetes]
|
runs-on: [self-hosted, pod]
|
||||||
```
|
```
|
||||||
|
|
||||||
```yaml
|
```yaml
|
||||||
runs-on: [self-hosted, vm, opensandbox]
|
runs-on: [self-hosted, vm]
|
||||||
```
|
```
|
||||||
|
|
||||||
- `self-hosted` 是固定前缀。
|
- `self-hosted` 是固定前缀。
|
||||||
- `container`、`vm` 是 workload class;`kubernetes`、`opensandbox` 是 placement driver。
|
- `pod` 表示一次性 Kubernetes Pod,承担常规 CI、镜像构建和 kind 等任务。
|
||||||
- 当前支持 `container+kubernetes` 和 `vm+opensandbox`。旧 `pod` 与 `vm` 标签分别是
|
- `vm` 表示一次性 microVM,承担需要独立内核、KVM、systemd 或更强隔离的任务。
|
||||||
两个组合的严格兼容别名,显式指定 driver 后不得因容量或故障回退到另一 driver。
|
|
||||||
- container 承担常规 CI、镜像构建和 kind 等任务;VM 承担需要独立内核、KVM、
|
|
||||||
systemd 或更强隔离的任务。
|
|
||||||
|
|
||||||
执行后端是基础设施选择,不是权限角色。workflow 不需要额外声明由 controller
|
执行后端是基础设施选择,不是权限角色。workflow 不需要额外声明由 controller
|
||||||
维护的 role 或权限 label。
|
维护的 role 或权限 label。
|
||||||
|
|||||||
@@ -40,11 +40,9 @@ 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、Docker 工具和
|
- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI 和 identity gate;
|
||||||
identity gate;Pod 与 VM backend 使用同一个镜像;
|
- runner UID 2000,SPIRE Agent 与 privileged dockerd 使用不同 UID;
|
||||||
- runner UID 2000;SPIRE Agent 独立运行;job-started hook 在第一步 workflow 之前
|
- Docker socket 通过 group 2000 共享,Docker 数据仅存在于 sandbox emptyDir;
|
||||||
启动 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 完成一项任务后退出。
|
||||||
|
|
||||||
|
|||||||
@@ -3,9 +3,9 @@
|
|||||||
## 目标
|
## 目标
|
||||||
|
|
||||||
长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
|
长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
|
||||||
协议客户端注册,并声明 workload class 与 driver labels;单个 registration 内按总
|
协议客户端注册,并声明 `self-hosted`、`pod` 和 `vm` labels;单个 registration 内按总
|
||||||
配置容量启动多个 `FetchTask` goroutine,再将 task 按 `runs-on` 交给 placement 对应的
|
配置容量启动多个 `FetchTask` goroutine,再将 task 按 `runs-on` 交给 Pod 或 VM 的独立
|
||||||
独立容量池,由一次性 Pod 或 microVM 执行。
|
容量池,由一次性 Pod 或 microVM 执行。
|
||||||
|
|
||||||
```text
|
```text
|
||||||
Gitea RunnerService
|
Gitea RunnerService
|
||||||
@@ -13,14 +13,14 @@ Gitea RunnerService
|
|||||||
▼
|
▼
|
||||||
dynamic-runner scheduler
|
dynamic-runner scheduler
|
||||||
│ 已领取的 task + lease
|
│ 已领取的 task + lease
|
||||||
├── container.kubernetes executor
|
├── Pod executor
|
||||||
└── vm.opensandbox executor
|
└── microVM executor
|
||||||
│ logs / state / result
|
│ logs / state / result
|
||||||
└──────────────────────► Gitea
|
└──────────────────────► Gitea
|
||||||
```
|
```
|
||||||
|
|
||||||
controller 使用单一 Go 二进制;默认在同一进程启用 `scheduler`、`kubernetes-worker`
|
controller 使用单一 Go 二进制;默认在同一进程启用 `scheduler`、`pod-worker` 和
|
||||||
和 `opensandbox-worker`,也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界,
|
`vm-worker`,也可通过 `--components` 只启用其中一部分。组件是独立应用服务边界,
|
||||||
共享进程不意味着共享后端状态或把 assignment 降级为内存 channel。
|
共享进程不意味着共享后端状态或把 assignment 降级为内存 channel。
|
||||||
|
|
||||||
首轮集成的 facade pending/claimed registry 与三个组件同进程。虽然二进制保留组件选择
|
首轮集成的 facade pending/claimed registry 与三个组件同进程。虽然二进制保留组件选择
|
||||||
@@ -47,13 +47,10 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
|
|||||||
|
|
||||||
## 设计约束
|
## 设计约束
|
||||||
|
|
||||||
- 规范接口为 `[self-hosted, container, kubernetes]` 和
|
- 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
|
||||||
`[self-hosted, vm, opensandbox]`。旧 `pod`、`vm`、`vm-dev` 标签保留严格映射,不能与
|
|
||||||
冲突 class/driver 混用;显式 driver 不允许自动回退。
|
|
||||||
- scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`,不为并发槽位重复注册;
|
- scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`,不为并发槽位重复注册;
|
||||||
`POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。
|
`POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。
|
||||||
- task 领取并持久化后按 placement 进入独立 durable consumer;v2 subject 为
|
- task 领取并持久化后按 backend 进入独立 durable consumer;对应容量池已满时延迟 NAK,
|
||||||
`<subject-base>.<workload-class>.<driver>`。对应容量池已满时延迟 NAK,
|
|
||||||
assignment 保持 JetStream pending,且不得创建超出配置容量的 workload。
|
assignment 保持 JetStream pending,且不得创建超出配置容量的 workload。
|
||||||
- scheduler Declare 后使用 RunnerService 长轮询;一旦 FetchTask 返回已分配 task,在
|
- scheduler Declare 后使用 RunnerService 长轮询;一旦 FetchTask 返回已分配 task,在
|
||||||
JetStream publish 成功前只重试该 assignment,不领取下一项。
|
JetStream publish 成功前只重试该 assignment,不领取下一项。
|
||||||
@@ -69,7 +66,7 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
|
|||||||
- JetStream 只持久化和投递 assignment,不保存 executor 生命周期状态。Pod labels/annotations
|
- JetStream 只持久化和投递 assignment,不保存 executor 生命周期状态。Pod labels/annotations
|
||||||
与 OpenSandbox metadata 是后端运行状态的权威来源,Gitea 是 task 终态的权威来源。
|
与 OpenSandbox metadata 是后端运行状态的权威来源,Gitea 是 task 终态的权威来源。
|
||||||
- assignment 使用版本化 envelope 保存完整 Gitea protobuf task,并从 workflow `runs-on`
|
- assignment 使用版本化 envelope 保存完整 Gitea protobuf task,并从 workflow `runs-on`
|
||||||
严格选择 workload class 与 driver;消费者解码后重新派生 placement 与身份,拒绝被篡改的冗余字段。
|
严格选择 pod 或 vm subject;消费者解码后重新派生 backend 与身份,拒绝被篡改的冗余字段。
|
||||||
- JetStream 的 message ID 等于稳定 assignment ID `gitea-task-<task-id>`,仅用于发布去重,
|
- JetStream 的 message ID 等于稳定 assignment ID `gitea-task-<task-id>`,仅用于发布去重,
|
||||||
不承担 executor 生命周期记录。
|
不承担 executor 生命周期记录。
|
||||||
- worker 按稳定 assignment ID reconcile 后端资源,进程内只保留并发控制等可丢弃状态;
|
- worker 按稳定 assignment ID reconcile 后端资源,进程内只保留并发控制等可丢弃状态;
|
||||||
@@ -79,7 +76,7 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
|
|||||||
- executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后,facade 在后端
|
- executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后,facade 在后端
|
||||||
metadata 写入持久 terminal marker;backend reconciler 仅在执行环境也进入终态后清理,
|
metadata 写入持久 terminal marker;backend reconciler 仅在执行环境也进入终态后清理,
|
||||||
从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。
|
从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。
|
||||||
- 每个 placement 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment
|
- pod 与 vm 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment
|
||||||
幂等落到后端;executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得
|
幂等落到后端;executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得
|
||||||
Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。
|
Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。
|
||||||
- admission pool 只保存可重建的并发状态:启动时从 Pod labels/annotations 或 OpenSandbox
|
- admission pool 只保存可重建的并发状态:启动时从 Pod labels/annotations 或 OpenSandbox
|
||||||
@@ -89,10 +86,10 @@ facade,并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
|
|||||||
- consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment;无需把完整
|
- consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment;无需把完整
|
||||||
task 写入 Pod annotation、OpenSandbox metadata 或环境变量。
|
task 写入 Pod annotation、OpenSandbox metadata 或环境变量。
|
||||||
- Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。
|
- Pod 与 VM 共享 task/executor 协议,只有环境创建和销毁实现不同。
|
||||||
- scheduler 在 assignment 持久化到 JetStream 后即可继续领取;各 placement 分别由 durable
|
- scheduler 在 assignment 持久化到 JetStream 后即可继续领取;Pod 与 VM 分别由 durable
|
||||||
consumer 的 capacity 限制并发,不共享全局执行槽位。未知后端故障由对应 consumer 的
|
consumer 的 capacity 限制并发,不共享全局执行槽位。未知后端故障由对应 consumer 的
|
||||||
NAK/redelivery 收敛,不能阻塞另一种 backend。
|
NAK/redelivery 收敛,不能阻塞另一种 backend。
|
||||||
- 两种现有 driver 都注入同一份 runner bootstrap 环境;container 仍由 homelab Kubernetes 原生
|
- 两种 backend 都注入同一份 runner bootstrap 环境;Pod 仍由 homelab Kubernetes 原生
|
||||||
创建,只有 VM 经 OpenSandbox 创建,bootstrap 机制不改变 backend 边界。
|
创建,只有 VM 经 OpenSandbox 创建,bootstrap 机制不改变 backend 边界。
|
||||||
|
|
||||||
## 实现顺序
|
## 实现顺序
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ type publishAPI interface {
|
|||||||
PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error)
|
PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Publisher implements the scheduler dispatcher with one subject per placement.
|
// Publisher implements the scheduler dispatcher with one subject per backend.
|
||||||
type Publisher struct {
|
type Publisher struct {
|
||||||
JetStream publishAPI
|
JetStream publishAPI
|
||||||
SubjectBase string
|
SubjectBase string
|
||||||
@@ -38,7 +38,7 @@ func (p Publisher) Dispatch(ctx context.Context, assignment taskassignment.Assig
|
|||||||
return errors.New("assignment subject base is required")
|
return errors.New("assignment subject base is required")
|
||||||
}
|
}
|
||||||
message := &nats.Msg{
|
message := &nats.Msg{
|
||||||
Subject: base + "." + assignment.Placement.Key(),
|
Subject: base + "." + string(assignment.Backend),
|
||||||
Header: nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}},
|
Header: nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}},
|
||||||
Data: body,
|
Data: body,
|
||||||
}
|
}
|
||||||
@@ -78,22 +78,6 @@ 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
|
|
||||||
Placement taskassignment.Placement
|
|
||||||
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, Placement: assignment.Placement, RetryDelay: retryDelay})
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p Processor) Process(ctx context.Context, message Message) error {
|
func (p Processor) Process(ctx context.Context, message Message) error {
|
||||||
@@ -104,7 +88,6 @@ 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"))
|
||||||
}
|
}
|
||||||
@@ -113,10 +96,8 @@ 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)
|
||||||
@@ -124,11 +105,9 @@ 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
|
||||||
@@ -141,21 +120,17 @@ 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)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -178,29 +153,23 @@ type consumerManager interface {
|
|||||||
|
|
||||||
// OpenConsumer creates the durable backend cursor. Capacity is enforced both
|
// OpenConsumer creates the durable backend cursor. Capacity is enforced both
|
||||||
// server-side and by ConsumerComponent's local semaphore.
|
// server-side and by ConsumerComponent's local semaphore.
|
||||||
func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, placement taskassignment.Placement, capacity int) (jetstream.Consumer, error) {
|
func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, backend taskassignment.Backend, capacity int) (jetstream.Consumer, error) {
|
||||||
if manager == nil || stream == "" || strings.TrimSuffix(subjectBase, ".") == "" || capacity < 1 {
|
if manager == nil || stream == "" || strings.TrimSuffix(subjectBase, ".") == "" || capacity < 1 {
|
||||||
return nil, errors.New("JetStream manager, stream, subject base, and positive capacity are required")
|
return nil, errors.New("JetStream manager, stream, subject base, and positive capacity are required")
|
||||||
}
|
}
|
||||||
if err := placement.Validate(); err != nil {
|
if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
|
||||||
return nil, err
|
return nil, fmt.Errorf("unsupported assignment backend %q", backend)
|
||||||
}
|
}
|
||||||
key := placement.Key()
|
|
||||||
// JetStream consumer names may not contain dots even though subjects do.
|
|
||||||
// Keep the placement subject hierarchical while using a stable, legal name
|
|
||||||
// for the durable cursor.
|
|
||||||
consumerName := strings.ReplaceAll(key, ".", "-")
|
|
||||||
consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
|
consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
|
||||||
Name: consumerName,
|
Name: string(backend),
|
||||||
Durable: consumerName,
|
Durable: string(backend),
|
||||||
FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + key,
|
FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + string(backend),
|
||||||
AckPolicy: jetstream.AckExplicitPolicy,
|
AckPolicy: jetstream.AckExplicitPolicy,
|
||||||
AckWait: 5 * time.Minute,
|
AckWait: 5 * time.Minute,
|
||||||
MaxAckPending: capacity,
|
MaxAckPending: capacity,
|
||||||
MaxDeliver: 1000,
|
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("open %s assignment consumer: %w", key, err)
|
return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)
|
||||||
}
|
}
|
||||||
return consumer, nil
|
return consumer, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -38,13 +38,13 @@ func (p *fakePublisher) PublishMsg(_ context.Context, message *nats.Msg, _ ...je
|
|||||||
return &jetstream.PubAck{}, nil
|
return &jetstream.PubAck{}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestPublisherUsesPlacementSubjectAndAssignmentDeduplication(t *testing.T) {
|
func TestPublisherUsesBackendSubjectAndAssignmentDeduplication(t *testing.T) {
|
||||||
api := &fakePublisher{}
|
api := &fakePublisher{}
|
||||||
publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"}
|
publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"}
|
||||||
if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil {
|
if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if api.message.Subject != "ci.assignment.container.kubernetes" {
|
if api.message.Subject != "ci.assignment.pod" {
|
||||||
t.Fatalf("subject = %q", api.message.Subject)
|
t.Fatalf("subject = %q", api.message.Subject)
|
||||||
}
|
}
|
||||||
if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" {
|
if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" {
|
||||||
@@ -126,23 +126,13 @@ 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)}
|
||||||
var events []Event
|
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}}
|
||||||
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].Placement != taskassignment.KubernetesContainer {
|
|
||||||
t.Fatalf("event[%d] = %#v", index, events[index])
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
|
func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
|
||||||
@@ -210,12 +200,12 @@ func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string
|
|||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestOpenConsumerUsesIndependentDurablePerPlacement(t *testing.T) {
|
func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
|
||||||
manager := &fakeConsumerManager{}
|
manager := &fakeConsumerManager{}
|
||||||
if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.KubernetesContainer, 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.Name != "container-kubernetes" || manager.config.Durable != "container-kubernetes" || manager.config.FilterSubject != "ci.assignment.container.kubernetes" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
|
if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 0 {
|
||||||
t.Fatalf("config = %#v", manager.config)
|
t.Fatalf("config = %#v", manager.config)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -14,12 +14,12 @@ import (
|
|||||||
type ComponentName string
|
type ComponentName string
|
||||||
|
|
||||||
const (
|
const (
|
||||||
Scheduler ComponentName = "scheduler"
|
Scheduler ComponentName = "scheduler"
|
||||||
KubernetesWorker ComponentName = "kubernetes-worker"
|
PodWorker ComponentName = "pod-worker"
|
||||||
OpenSandboxWorker ComponentName = "opensandbox-worker"
|
VMWorker ComponentName = "vm-worker"
|
||||||
)
|
)
|
||||||
|
|
||||||
var defaultComponents = []ComponentName{Scheduler, KubernetesWorker, OpenSandboxWorker}
|
var defaultComponents = []ComponentName{Scheduler, PodWorker, VMWorker}
|
||||||
|
|
||||||
// Selection parses --components. An empty value enables all components.
|
// Selection parses --components. An empty value enables all components.
|
||||||
type Selection []ComponentName
|
type Selection []ComponentName
|
||||||
@@ -31,12 +31,6 @@ func ParseSelection(value string) (Selection, error) {
|
|||||||
var selected Selection
|
var selected Selection
|
||||||
for _, raw := range strings.Split(value, ",") {
|
for _, raw := range strings.Split(value, ",") {
|
||||||
name := ComponentName(strings.TrimSpace(raw))
|
name := ComponentName(strings.TrimSpace(raw))
|
||||||
switch name {
|
|
||||||
case "pod-worker":
|
|
||||||
name = KubernetesWorker
|
|
||||||
case "vm-worker":
|
|
||||||
name = OpenSandboxWorker
|
|
||||||
}
|
|
||||||
if !slices.Contains(defaultComponents, name) {
|
if !slices.Contains(defaultComponents, name) {
|
||||||
return nil, fmt.Errorf("unknown controller component %q", name)
|
return nil, fmt.Errorf("unknown controller component %q", name)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,18 +13,18 @@ func TestParseSelectionDefaultsToAll(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(selection) != 3 || selection[0] != Scheduler || selection[1] != KubernetesWorker || selection[2] != OpenSandboxWorker {
|
if len(selection) != 3 || selection[0] != Scheduler || selection[1] != PodWorker || selection[2] != VMWorker {
|
||||||
t.Fatalf("selection = %v", selection)
|
t.Fatalf("selection = %v", selection)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) {
|
func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) {
|
||||||
selection, err := ParseSelection("opensandbox-worker,scheduler,opensandbox-worker")
|
selection, err := ParseSelection("vm-worker,scheduler,vm-worker")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if len(selection) != 2 || selection[0] != OpenSandboxWorker || selection[1] != Scheduler {
|
if len(selection) != 2 || selection[0] != VMWorker || selection[1] != Scheduler {
|
||||||
t.Fatalf("selection = %v", selection)
|
t.Fatalf("selection = %v", selection)
|
||||||
}
|
}
|
||||||
if _, err := ParseSelection("webhook"); err == nil {
|
if _, err := ParseSelection("webhook"); err == nil {
|
||||||
@@ -32,16 +32,6 @@ func TestParseSelectionAllowsOneOrMoreComponents(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestParseSelectionNormalizesLegacyWorkerNames(t *testing.T) {
|
|
||||||
selection, err := ParseSelection("pod-worker,vm-worker")
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if len(selection) != 2 || selection[0] != KubernetesWorker || selection[1] != OpenSandboxWorker {
|
|
||||||
t.Fatalf("selection = %v", selection)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
type componentFunc func(context.Context) error
|
type componentFunc func(context.Context) error
|
||||||
|
|
||||||
func (f componentFunc) Run(ctx context.Context) error { return f(ctx) }
|
func (f componentFunc) Run(ctx context.Context) error { return f(ctx) }
|
||||||
@@ -55,14 +45,14 @@ func TestRunStartsSelectedComponentsAndCancelsPeers(t *testing.T) {
|
|||||||
started <- Scheduler
|
started <- Scheduler
|
||||||
return errors.New("poll failed")
|
return errors.New("poll failed")
|
||||||
}),
|
}),
|
||||||
KubernetesWorker: componentFunc(func(ctx context.Context) error {
|
PodWorker: componentFunc(func(ctx context.Context) error {
|
||||||
started <- KubernetesWorker
|
started <- PodWorker
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
once.Do(func() { close(peerStopped) })
|
once.Do(func() { close(peerStopped) })
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
err := Run(context.Background(), Selection{Scheduler, KubernetesWorker}, registry)
|
err := Run(context.Background(), Selection{Scheduler, PodWorker}, registry)
|
||||||
if err == nil || !errors.Is(err, context.Canceled) && err.Error() != "component scheduler: poll failed" {
|
if err == nil || !errors.Is(err, context.Canceled) && err.Error() != "component scheduler: poll failed" {
|
||||||
t.Fatalf("Run() error = %v", err)
|
t.Fatalf("Run() error = %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -119,10 +119,7 @@ func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
|
result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
|
||||||
Metadata: map[string]string{
|
Metadata: map[string]string{"ci.ddupan.top/backend": "vm"}, PageSize: 100,
|
||||||
"ci.ddupan.top/workload-class": "vm",
|
|
||||||
"ci.ddupan.top/driver": "opensandbox",
|
|
||||||
}, PageSize: 100,
|
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("list recoverable sandboxes: %w", err)
|
return nil, fmt.Errorf("list recoverable sandboxes: %w", err)
|
||||||
@@ -175,8 +172,8 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen
|
|||||||
if err := b.validate(); err != nil {
|
if err := b.validate(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if assignment.Placement != taskassignment.OpenSandboxVM {
|
if assignment.Backend != taskassignment.BackendVM {
|
||||||
return nil, fmt.Errorf("OpenSandbox backend cannot create %q", assignment.Placement.Key())
|
return nil, fmt.Errorf("OpenSandbox backend cannot create %q assignment", assignment.Backend)
|
||||||
}
|
}
|
||||||
environment := clone(b.Config.Env)
|
environment := clone(b.Config.Env)
|
||||||
for key, value := range launch.Environment {
|
for key, value := range launch.Environment {
|
||||||
|
|||||||
@@ -63,7 +63,7 @@ func backend(lifecycle Lifecycle) Backend {
|
|||||||
|
|
||||||
func assignment() taskassignment.Assignment {
|
func assignment() taskassignment.Assignment {
|
||||||
return taskassignment.Assignment{
|
return taskassignment.Assignment{
|
||||||
ID: "gitea-task-42", Placement: taskassignment.OpenSandboxVM,
|
ID: "gitea-task-42", Backend: taskassignment.BackendVM,
|
||||||
Task: &runnerv1.Task{Id: 42},
|
Task: &runnerv1.Task{Id: 42},
|
||||||
Identity: taskidentity.Identity{Repository: "owner/repo", Task: "publish", SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
|
Identity: taskidentity.Identity{Repository: "owner/repo", Task: "publish", SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -120,7 +120,7 @@ func (b Backend) RecoverAssignments(ctx context.Context) ([]taskassignment.Assig
|
|||||||
if err := b.validate(); err != nil {
|
if err := b.validate(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/workload-class=container,ci.ddupan.top/driver=kubernetes")
|
pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/backend=pod")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("list recoverable assignment Pods: %w", err)
|
return nil, fmt.Errorf("list recoverable assignment Pods: %w", err)
|
||||||
}
|
}
|
||||||
@@ -159,8 +159,8 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen
|
|||||||
if err := b.validate(); err != nil {
|
if err := b.validate(); err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if assignment.Placement != taskassignment.KubernetesContainer {
|
if assignment.Backend != taskassignment.BackendPod {
|
||||||
return nil, fmt.Errorf("Kubernetes backend cannot create %q", assignment.Placement.Key())
|
return nil, fmt.Errorf("Pod backend cannot create %q assignment", assignment.Backend)
|
||||||
}
|
}
|
||||||
labels := clone(launch.Metadata.Labels)
|
labels := clone(launch.Metadata.Labels)
|
||||||
labels["app.kubernetes.io/name"] = "gitea-dynamic-runner"
|
labels["app.kubernetes.io/name"] = "gitea-dynamic-runner"
|
||||||
|
|||||||
@@ -59,7 +59,7 @@ func backend(api API) Backend {
|
|||||||
|
|
||||||
func assignment() taskassignment.Assignment {
|
func assignment() taskassignment.Assignment {
|
||||||
return taskassignment.Assignment{
|
return taskassignment.Assignment{
|
||||||
ID: "gitea-task-42", Placement: taskassignment.KubernetesContainer,
|
ID: "gitea-task-42", Backend: taskassignment.BackendPod,
|
||||||
Task: &runnerv1.Task{Id: 42},
|
Task: &runnerv1.Task{Id: 42},
|
||||||
Identity: taskidentity.Identity{
|
Identity: taskidentity.Identity{
|
||||||
Repository: "owner/repo", Task: "publish",
|
Repository: "owner/repo", Task: "publish",
|
||||||
|
|||||||
@@ -8,7 +8,6 @@ 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"
|
||||||
@@ -77,25 +76,16 @@ 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{{
|
||||||
|
Name: "spire-agent-socket",
|
||||||
|
VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
|
||||||
|
Driver: "csi.spiffe.io", ReadOnly: boolPointer(true),
|
||||||
|
}},
|
||||||
}},
|
}},
|
||||||
Volumes: []corev1.Volume{
|
|
||||||
{
|
|
||||||
Name: "spire-agent-socket",
|
|
||||||
VolumeSource: corev1.VolumeSource{CSI: &corev1.CSIVolumeSource{
|
|
||||||
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{})
|
||||||
@@ -179,11 +169,6 @@ 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,12 +37,6 @@ 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)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,13 +13,12 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
EnvAssignmentID = "CI_ASSIGNMENT_ID"
|
EnvAssignmentID = "CI_ASSIGNMENT_ID"
|
||||||
EnvCapability = "CI_RUNNER_CAPABILITY"
|
EnvCapability = "CI_RUNNER_CAPABILITY"
|
||||||
EnvFacadeURL = "CI_RUNNER_FACADE_URL"
|
EnvFacadeURL = "CI_RUNNER_FACADE_URL"
|
||||||
EnvFacadeID = "CI_RUNNER_FACADE_SPIFFE_ID"
|
EnvFacadeID = "CI_RUNNER_FACADE_SPIFFE_ID"
|
||||||
EnvSPIFFEID = "CI_SPIFFE_ID"
|
EnvSPIFFEID = "CI_SPIFFE_ID"
|
||||||
EnvWorkloadClass = "CI_WORKLOAD_CLASS"
|
EnvBackend = "CI_RUNNER_BACKEND"
|
||||||
EnvDriver = "CI_WORKLOAD_DRIVER"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
// Bootstrap emits assignment-scoped launch configuration. FacadeURL is the
|
// Bootstrap emits assignment-scoped launch configuration. FacadeURL is the
|
||||||
@@ -49,8 +48,7 @@ func (b Bootstrap) Environment(assignment taskassignment.Assignment) (map[string
|
|||||||
EnvFacadeURL: b.FacadeURL,
|
EnvFacadeURL: b.FacadeURL,
|
||||||
EnvFacadeID: b.FacadeSPIFFEID,
|
EnvFacadeID: b.FacadeSPIFFEID,
|
||||||
EnvSPIFFEID: assignment.Identity.SPIFFEID,
|
EnvSPIFFEID: assignment.Identity.SPIFFEID,
|
||||||
EnvWorkloadClass: string(assignment.Placement.Class),
|
EnvBackend: string(assignment.Backend),
|
||||||
EnvDriver: string(assignment.Placement.Driver),
|
|
||||||
"SPIFFE_ENDPOINT_SOCKET": b.WorkloadAPIAddr,
|
"SPIFFE_ENDPOINT_SOCKET": b.WorkloadAPIAddr,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
@@ -69,7 +67,7 @@ type Registration struct {
|
|||||||
Ephemeral bool `json:"ephemeral"`
|
Ephemeral bool `json:"ephemeral"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func RegistrationJSON(assignmentID, capability, localProxyURL string, placement taskassignment.Placement) ([]byte, error) {
|
func RegistrationJSON(assignmentID, capability, localProxyURL string, backend taskassignment.Backend) ([]byte, error) {
|
||||||
if assignmentID == "" || capability == "" {
|
if assignmentID == "" || capability == "" {
|
||||||
return nil, errors.New("assignment ID and runner capability are required")
|
return nil, errors.New("assignment ID and runner capability are required")
|
||||||
}
|
}
|
||||||
@@ -77,13 +75,13 @@ func RegistrationJSON(assignmentID, capability, localProxyURL string, placement
|
|||||||
if err != nil || parsed.Scheme != "http" || parsed.Host == "" {
|
if err != nil || parsed.Scheme != "http" || parsed.Host == "" {
|
||||||
return nil, errors.New("local runner proxy URL must be an absolute http URL")
|
return nil, errors.New("local runner proxy URL must be an absolute http URL")
|
||||||
}
|
}
|
||||||
if err := placement.Validate(); err != nil {
|
if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
|
||||||
return nil, err
|
return nil, fmt.Errorf("unsupported runner backend %q", backend)
|
||||||
}
|
}
|
||||||
registration := Registration{
|
registration := Registration{
|
||||||
Warning: "Generated for one preassigned task by gitea-dynamic-runner.",
|
Warning: "Generated for one preassigned task by gitea-dynamic-runner.",
|
||||||
UUID: assignmentID, Name: assignmentID, Token: capability,
|
UUID: assignmentID, Name: assignmentID, Token: capability,
|
||||||
Address: localProxyURL, Labels: []string{"self-hosted", string(placement.Class), string(placement.Driver)}, Ephemeral: true,
|
Address: localProxyURL, Labels: []string{"self-hosted", string(backend)}, Ephemeral: true,
|
||||||
}
|
}
|
||||||
data, err := json.MarshalIndent(registration, "", " ")
|
data, err := json.MarshalIndent(registration, "", " ")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ func testBootstrap(t *testing.T) Bootstrap {
|
|||||||
|
|
||||||
func testAssignment() taskassignment.Assignment {
|
func testAssignment() taskassignment.Assignment {
|
||||||
return taskassignment.Assignment{
|
return taskassignment.Assignment{
|
||||||
ID: "gitea-task-42", Placement: taskassignment.KubernetesContainer,
|
ID: "gitea-task-42", Backend: taskassignment.BackendPod,
|
||||||
Identity: taskidentity.Identity{SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
|
Identity: taskidentity.Identity{SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -46,7 +46,7 @@ func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) {
|
|||||||
if first[EnvAssignmentID] != "gitea-task-42" || first[EnvSPIFFEID] != testAssignment().Identity.SPIFFEID {
|
if first[EnvAssignmentID] != "gitea-task-42" || first[EnvSPIFFEID] != testAssignment().Identity.SPIFFEID {
|
||||||
t.Fatalf("environment = %#v", first)
|
t.Fatalf("environment = %#v", first)
|
||||||
}
|
}
|
||||||
if first[EnvWorkloadClass] != "container" || first[EnvDriver] != "kubernetes" || first[EnvFacadeID] == "" {
|
if first[EnvBackend] != "pod" || first[EnvFacadeID] == "" {
|
||||||
t.Fatalf("environment = %#v", first)
|
t.Fatalf("environment = %#v", first)
|
||||||
}
|
}
|
||||||
if first["SPIFFE_ENDPOINT_SOCKET"] != "unix:///run/spire/agent-sockets/spire-agent.sock" {
|
if first["SPIFFE_ENDPOINT_SOCKET"] != "unix:///run/spire/agent-sockets/spire-agent.sock" {
|
||||||
@@ -55,7 +55,7 @@ func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) {
|
func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) {
|
||||||
data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.OpenSandboxVM)
|
data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.BackendVM)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -66,13 +66,13 @@ func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) {
|
|||||||
if registration.UUID != "gitea-task-42" || registration.Token != "capability" || registration.Address != "http://127.0.0.1:8080" || !registration.Ephemeral {
|
if registration.UUID != "gitea-task-42" || registration.Token != "capability" || registration.Address != "http://127.0.0.1:8080" || !registration.Ephemeral {
|
||||||
t.Fatalf("registration = %#v", registration)
|
t.Fatalf("registration = %#v", registration)
|
||||||
}
|
}
|
||||||
if len(registration.Labels) != 3 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" || registration.Labels[2] != "opensandbox" {
|
if len(registration.Labels) != 2 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" {
|
||||||
t.Fatalf("labels = %#v", registration.Labels)
|
t.Fatalf("labels = %#v", registration.Labels)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRegistrationRejectsNonLocalTLSAddress(t *testing.T) {
|
func TestRegistrationRejectsNonLocalTLSAddress(t *testing.T) {
|
||||||
if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.KubernetesContainer); err == nil {
|
if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.BackendPod); err == nil {
|
||||||
t.Fatal("expected local proxy URL validation error")
|
t.Fatal("expected local proxy URL validation error")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -17,12 +17,11 @@ import (
|
|||||||
type ExecutorConfig struct {
|
type ExecutorConfig struct {
|
||||||
AssignmentID string
|
AssignmentID string
|
||||||
Capability string
|
Capability string
|
||||||
Placement taskassignment.Placement
|
Backend taskassignment.Backend
|
||||||
FacadeURL string
|
FacadeURL string
|
||||||
FacadeSPIFFEID string
|
FacadeSPIFFEID string
|
||||||
WorkloadAPIAddr string
|
WorkloadAPIAddr string
|
||||||
RunnerBinary string
|
RunnerBinary string
|
||||||
RunnerConfig string
|
|
||||||
ListenAddress string
|
ListenAddress string
|
||||||
WorkDir string
|
WorkDir string
|
||||||
Stdout *os.File
|
Stdout *os.File
|
||||||
@@ -36,9 +35,6 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error {
|
|||||||
if config.RunnerBinary == "" {
|
if config.RunnerBinary == "" {
|
||||||
config.RunnerBinary = "gitea-runner"
|
config.RunnerBinary = "gitea-runner"
|
||||||
}
|
}
|
||||||
if config.RunnerConfig == "" {
|
|
||||||
config.RunnerConfig = "/etc/gitea-runner/config.yaml"
|
|
||||||
}
|
|
||||||
if config.ListenAddress == "" {
|
if config.ListenAddress == "" {
|
||||||
config.ListenAddress = "127.0.0.1:0"
|
config.ListenAddress = "127.0.0.1:0"
|
||||||
}
|
}
|
||||||
@@ -71,7 +67,7 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error {
|
|||||||
defer os.RemoveAll(workDir)
|
defer os.RemoveAll(workDir)
|
||||||
}
|
}
|
||||||
registration, err := RegistrationJSON(
|
registration, err := RegistrationJSON(
|
||||||
config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Placement,
|
config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Backend,
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -97,7 +93,7 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error {
|
|||||||
return errors.Join(readyErr, shutdownErr, serverErr)
|
return errors.Join(readyErr, shutdownErr, serverErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
command := exec.CommandContext(ctx, config.RunnerBinary, runnerArguments(config.RunnerConfig)...)
|
command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once")
|
||||||
command.Dir = workDir
|
command.Dir = workDir
|
||||||
command.Stdout = config.Stdout
|
command.Stdout = config.Stdout
|
||||||
command.Stderr = config.Stderr
|
command.Stderr = config.Stderr
|
||||||
@@ -112,10 +108,6 @@ func RunExecutor(ctx context.Context, config ExecutorConfig) error {
|
|||||||
return errors.Join(runnerErr, shutdownErr, serverErr)
|
return errors.Join(runnerErr, shutdownErr, serverErr)
|
||||||
}
|
}
|
||||||
|
|
||||||
func runnerArguments(configFile string) []string {
|
|
||||||
return []string{"daemon", "--config", configFile, "--once"}
|
|
||||||
}
|
|
||||||
|
|
||||||
func waitForFacade(ctx context.Context, endpoint string) error {
|
func waitForFacade(ctx context.Context, endpoint string) error {
|
||||||
client := &http.Client{Timeout: 2 * time.Second}
|
client := &http.Client{Timeout: 2 * time.Second}
|
||||||
ticker := time.NewTicker(250 * time.Millisecond)
|
ticker := time.NewTicker(250 * time.Millisecond)
|
||||||
@@ -144,16 +136,15 @@ func waitForFacade(ctx context.Context, endpoint string) error {
|
|||||||
// the assignment-scoped values injected by the backend. The Workload API
|
// the assignment-scoped values injected by the backend. The Workload API
|
||||||
// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
|
// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here.
|
||||||
func ExecutorConfigFromEnvironment() (ExecutorConfig, error) {
|
func ExecutorConfigFromEnvironment() (ExecutorConfig, error) {
|
||||||
placement := taskassignment.Placement{Class: taskassignment.WorkloadClass(os.Getenv(EnvWorkloadClass)), Driver: taskassignment.Driver(os.Getenv(EnvDriver))}
|
backend := taskassignment.Backend(os.Getenv(EnvBackend))
|
||||||
if err := placement.Validate(); err != nil {
|
if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
|
||||||
return ExecutorConfig{}, err
|
return ExecutorConfig{}, fmt.Errorf("invalid %s %q", EnvBackend, backend)
|
||||||
}
|
}
|
||||||
config := ExecutorConfig{
|
config := ExecutorConfig{
|
||||||
AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability),
|
AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability),
|
||||||
Placement: placement, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID),
|
Backend: backend, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID),
|
||||||
RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), RunnerConfig: os.Getenv("GITEA_RUNNER_CONFIG_FILE"),
|
RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), ListenAddress: "127.0.0.1:0",
|
||||||
ListenAddress: "127.0.0.1:0",
|
Stdout: os.Stdout, Stderr: os.Stderr,
|
||||||
Stdout: os.Stdout, Stderr: os.Stderr,
|
|
||||||
}
|
}
|
||||||
if config.AssignmentID == "" || config.Capability == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
|
if config.AssignmentID == "" || config.Capability == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" {
|
||||||
return ExecutorConfig{}, errors.New("complete runner assignment and facade environment is required")
|
return ExecutorConfig{}, errors.New("complete runner assignment and facade environment is required")
|
||||||
|
|||||||
@@ -4,19 +4,11 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"reflect"
|
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestRunnerArgumentsLoadJobHooksConfig(t *testing.T) {
|
|
||||||
want := []string{"daemon", "--config", "/etc/gitea-runner/config.yaml", "--once"}
|
|
||||||
if got := runnerArguments("/etc/gitea-runner/config.yaml"); !reflect.DeepEqual(got, want) {
|
|
||||||
t.Fatalf("runner arguments = %q, want %q", got, want)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestWaitForFacadeRetriesTransientGatewayFailure(t *testing.T) {
|
func TestWaitForFacadeRetriesTransientGatewayFailure(t *testing.T) {
|
||||||
var requests atomic.Int32
|
var requests atomic.Int32
|
||||||
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
|
server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import (
|
|||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"slices"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
@@ -15,27 +16,34 @@ import (
|
|||||||
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
|
"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
|
||||||
)
|
)
|
||||||
|
|
||||||
const wireVersion = 2
|
const wireVersion = 1
|
||||||
|
|
||||||
|
type Backend string
|
||||||
|
|
||||||
|
const (
|
||||||
|
BackendPod Backend = "pod"
|
||||||
|
BackendVM Backend = "vm"
|
||||||
|
)
|
||||||
|
|
||||||
// Assignment is the only document persisted in the handoff queue.
|
// Assignment is the only document persisted in the handoff queue.
|
||||||
type Assignment struct {
|
type Assignment struct {
|
||||||
ID string
|
ID string
|
||||||
Placement Placement
|
Backend Backend
|
||||||
Task *runnerv1.Task
|
Task *runnerv1.Task
|
||||||
Identity taskidentity.Identity
|
Identity taskidentity.Identity
|
||||||
}
|
}
|
||||||
|
|
||||||
// FromMetadata reconstructs the minimal assignment needed to authorize an
|
// FromMetadata reconstructs the minimal assignment needed to authorize an
|
||||||
// already-running executor after a controller restart. Placement metadata was
|
// already-running executor after a controller restart. Backend metadata was
|
||||||
// originally derived from the trusted Gitea task and is validated again here.
|
// originally derived from the trusted Gitea task and is validated again here.
|
||||||
func FromMetadata(labels, annotations map[string]string, trustDomain string) (Assignment, error) {
|
func FromMetadata(labels, annotations map[string]string, trustDomain string) (Assignment, error) {
|
||||||
taskID, err := strconv.ParseInt(labels["ci.ddupan.top/task-id"], 10, 64)
|
taskID, err := strconv.ParseInt(labels["ci.ddupan.top/task-id"], 10, 64)
|
||||||
if err != nil || taskID < 1 {
|
if err != nil || taskID < 1 {
|
||||||
return Assignment{}, errors.New("backend metadata has invalid task ID")
|
return Assignment{}, errors.New("backend metadata has invalid task ID")
|
||||||
}
|
}
|
||||||
placement := Placement{Class: WorkloadClass(labels["ci.ddupan.top/workload-class"]), Driver: Driver(labels["ci.ddupan.top/driver"])}
|
backend := Backend(labels["ci.ddupan.top/backend"])
|
||||||
if err := placement.Validate(); err != nil {
|
if backend != BackendPod && backend != BackendVM {
|
||||||
return Assignment{}, err
|
return Assignment{}, errors.New("backend metadata has invalid backend")
|
||||||
}
|
}
|
||||||
id := labels["ci.ddupan.top/assignment-id"]
|
id := labels["ci.ddupan.top/assignment-id"]
|
||||||
if id != fmt.Sprintf("gitea-task-%d", taskID) {
|
if id != fmt.Sprintf("gitea-task-%d", taskID) {
|
||||||
@@ -48,15 +56,15 @@ func FromMetadata(labels, annotations map[string]string, trustDomain string) (As
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return Assignment{}, err
|
return Assignment{}, err
|
||||||
}
|
}
|
||||||
return Assignment{ID: id, Placement: placement, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil
|
return Assignment{ID: id, Backend: backend, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
type envelope struct {
|
type envelope struct {
|
||||||
Version int `json:"version"`
|
Version int `json:"version"`
|
||||||
ID string `json:"id"`
|
ID string `json:"id"`
|
||||||
Placement Placement `json:"placement"`
|
Backend Backend `json:"backend"`
|
||||||
Task []byte `json:"task"`
|
Task []byte `json:"task"`
|
||||||
Identity taskidentity.Identity `json:"identity"`
|
Identity taskidentity.Identity `json:"identity"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// New derives all trusted assignment fields from the task fetched from Gitea.
|
// New derives all trusted assignment fields from the task fetched from Gitea.
|
||||||
@@ -68,28 +76,40 @@ func New(task *runnerv1.Task, trustDomain string) (Assignment, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return Assignment{}, err
|
return Assignment{}, err
|
||||||
}
|
}
|
||||||
placement, err := placementFromTask(task)
|
backend, err := backendFromTask(task)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Assignment{}, err
|
return Assignment{}, err
|
||||||
}
|
}
|
||||||
return Assignment{
|
return Assignment{
|
||||||
ID: fmt.Sprintf("gitea-task-%d", task.GetId()),
|
ID: fmt.Sprintf("gitea-task-%d", task.GetId()),
|
||||||
Placement: placement,
|
Backend: backend,
|
||||||
Task: task,
|
Task: task,
|
||||||
Identity: identity,
|
Identity: identity,
|
||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func placementFromTask(task *runnerv1.Task) (Placement, error) {
|
func backendFromTask(task *runnerv1.Task) (Backend, error) {
|
||||||
workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload()))
|
workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload()))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Placement{}, fmt.Errorf("parse task workflow for placement: %w", err)
|
return "", fmt.Errorf("parse task workflow for backend: %w", err)
|
||||||
}
|
}
|
||||||
jobIDs := workflow.GetJobIDs()
|
jobIDs := workflow.GetJobIDs()
|
||||||
if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil {
|
if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil {
|
||||||
return Placement{}, fmt.Errorf("task workflow must contain exactly one non-empty job")
|
return "", fmt.Errorf("task workflow must contain exactly one non-empty job")
|
||||||
}
|
}
|
||||||
return PlacementFromLabels(workflow.GetJob(jobIDs[0]).RunsOnLabels())
|
labels := workflow.GetJob(jobIDs[0]).RunsOnLabels()
|
||||||
|
if !slices.Contains(labels, "self-hosted") {
|
||||||
|
return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
|
||||||
|
}
|
||||||
|
hasPod := slices.Contains(labels, string(BackendPod))
|
||||||
|
hasVM := slices.Contains(labels, string(BackendVM))
|
||||||
|
if hasPod == hasVM {
|
||||||
|
return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels)
|
||||||
|
}
|
||||||
|
if hasPod {
|
||||||
|
return BackendPod, nil
|
||||||
|
}
|
||||||
|
return BackendVM, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task.
|
// Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task.
|
||||||
@@ -97,16 +117,13 @@ func Marshal(assignment Assignment) ([]byte, error) {
|
|||||||
if assignment.Task == nil {
|
if assignment.Task == nil {
|
||||||
return nil, errors.New("assignment task is required")
|
return nil, errors.New("assignment task is required")
|
||||||
}
|
}
|
||||||
if err := assignment.Placement.Validate(); err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
task, err := proto.Marshal(assignment.Task)
|
task, err := proto.Marshal(assignment.Task)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("marshal Gitea task: %w", err)
|
return nil, fmt.Errorf("marshal Gitea task: %w", err)
|
||||||
}
|
}
|
||||||
return json.Marshal(envelope{
|
return json.Marshal(envelope{
|
||||||
Version: wireVersion,
|
Version: wireVersion,
|
||||||
ID: assignment.ID, Placement: assignment.Placement,
|
ID: assignment.ID, Backend: assignment.Backend,
|
||||||
Task: task, Identity: assignment.Identity,
|
Task: task, Identity: assignment.Identity,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -128,7 +145,7 @@ func Unmarshal(data []byte, trustDomain string) (Assignment, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return Assignment{}, err
|
return Assignment{}, err
|
||||||
}
|
}
|
||||||
if wire.ID != canonical.ID || wire.Placement != canonical.Placement || wire.Identity != canonical.Identity {
|
if wire.ID != canonical.ID || wire.Backend != canonical.Backend || wire.Identity != canonical.Identity {
|
||||||
return Assignment{}, errors.New("assignment metadata does not match its Gitea task")
|
return Assignment{}, errors.New("assignment metadata does not match its Gitea task")
|
||||||
}
|
}
|
||||||
return canonical, nil
|
return canonical, nil
|
||||||
|
|||||||
@@ -21,37 +21,28 @@ func task(t *testing.T, labels string) *runnerv1.Task {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestNewSelectsPlacementFromRunsOn(t *testing.T) {
|
func TestNewSelectsBackendFromRunsOn(t *testing.T) {
|
||||||
for _, test := range []struct {
|
for _, test := range []struct {
|
||||||
labels string
|
labels string
|
||||||
placement Placement
|
backend Backend
|
||||||
}{
|
}{
|
||||||
{"[self-hosted, pod]", KubernetesContainer},
|
{"[self-hosted, pod]", BackendPod},
|
||||||
{"[self-hosted, container]", KubernetesContainer},
|
{"[self-hosted, vm]", BackendVM},
|
||||||
{"[self-hosted, container, kubernetes]", KubernetesContainer},
|
|
||||||
{"[self-hosted, vm]", OpenSandboxVM},
|
|
||||||
{"[self-hosted, vm-dev]", OpenSandboxVM},
|
|
||||||
{"[self-hosted, vm, opensandbox]", OpenSandboxVM},
|
|
||||||
} {
|
} {
|
||||||
assignment, err := New(task(t, test.labels), "ddupan.top")
|
assignment, err := New(task(t, test.labels), "ddupan.top")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if assignment.Placement != test.placement || assignment.ID != "gitea-task-42" {
|
if assignment.Backend != test.backend || assignment.ID != "gitea-task-42" {
|
||||||
t.Fatalf("assignment = %#v", assignment)
|
t.Fatalf("assignment = %#v", assignment)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestNewRejectsInvalidPlacement(t *testing.T) {
|
func TestNewRejectsAmbiguousBackend(t *testing.T) {
|
||||||
for _, labels := range []string{
|
for _, labels := range []string{
|
||||||
"[self-hosted]",
|
"[self-hosted]",
|
||||||
"[self-hosted, pod, vm]",
|
"[self-hosted, pod, vm]",
|
||||||
"[self-hosted, pod, container]",
|
|
||||||
"[self-hosted, pod, kubernetes]",
|
|
||||||
"[self-hosted, container, opensandbox]",
|
|
||||||
"[self-hosted, vm, kubernetes]",
|
|
||||||
"[self-hosted, container, kubernetes, opensandbox]",
|
|
||||||
"[pod]",
|
"[pod]",
|
||||||
} {
|
} {
|
||||||
if _, err := New(task(t, labels), "ddupan.top"); err == nil {
|
if _, err := New(task(t, labels), "ddupan.top"); err == nil {
|
||||||
@@ -73,28 +64,27 @@ func TestAssignmentWireRoundTripAndValidation(t *testing.T) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if got.ID != want.ID || got.Placement != want.Placement || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) {
|
if got.ID != want.ID || got.Backend != want.Backend || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) {
|
||||||
t.Fatalf("round trip = %#v, want %#v", got, want)
|
t.Fatalf("round trip = %#v, want %#v", got, want)
|
||||||
}
|
}
|
||||||
|
|
||||||
tampered := bytes.Replace(data, []byte(`"driver":"kubernetes"`), []byte(`"driver":"opensandbox"`), 1)
|
tampered := bytes.Replace(data, []byte(`"backend":"pod"`), []byte(`"backend":"vm"`), 1)
|
||||||
if _, err := Unmarshal(tampered, "ddupan.top"); err == nil {
|
if _, err := Unmarshal(tampered, "ddupan.top"); err == nil {
|
||||||
t.Fatal("expected tampered placement to fail")
|
t.Fatal("expected tampered backend to fail")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestFromMetadataRecoversMinimalAssignment(t *testing.T) {
|
func TestFromMetadataRecoversMinimalAssignment(t *testing.T) {
|
||||||
assignment, err := FromMetadata(map[string]string{
|
assignment, err := FromMetadata(map[string]string{
|
||||||
"ci.ddupan.top/assignment-id": "gitea-task-42",
|
"ci.ddupan.top/assignment-id": "gitea-task-42",
|
||||||
"ci.ddupan.top/task-id": "42",
|
"ci.ddupan.top/task-id": "42",
|
||||||
"ci.ddupan.top/workload-class": "vm",
|
"ci.ddupan.top/backend": "vm",
|
||||||
"ci.ddupan.top/driver": "opensandbox",
|
|
||||||
}, map[string]string{
|
}, map[string]string{
|
||||||
"ci.ddupan.top/repository": "owner/repo",
|
"ci.ddupan.top/repository": "owner/repo",
|
||||||
"ci.ddupan.top/job-key": "publish",
|
"ci.ddupan.top/job-key": "publish",
|
||||||
"ci.ddupan.top/spiffe-id": "spiffe://ddupan.top/ci/owner/repo/publish",
|
"ci.ddupan.top/spiffe-id": "spiffe://ddupan.top/ci/owner/repo/publish",
|
||||||
}, "ddupan.top")
|
}, "ddupan.top")
|
||||||
if err != nil || assignment.Task.GetId() != 42 || assignment.Placement != OpenSandboxVM {
|
if err != nil || assignment.Task.GetId() != 42 || assignment.Backend != BackendVM {
|
||||||
t.Fatalf("assignment=%#v err=%v", assignment, err)
|
t.Fatalf("assignment=%#v err=%v", assignment, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,76 +0,0 @@
|
|||||||
package taskassignment
|
|
||||||
|
|
||||||
import (
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"slices"
|
|
||||||
)
|
|
||||||
|
|
||||||
type WorkloadClass string
|
|
||||||
|
|
||||||
const (
|
|
||||||
WorkloadContainer WorkloadClass = "container"
|
|
||||||
WorkloadVM WorkloadClass = "vm"
|
|
||||||
)
|
|
||||||
|
|
||||||
type Driver string
|
|
||||||
|
|
||||||
const (
|
|
||||||
DriverKubernetes Driver = "kubernetes"
|
|
||||||
DriverOpenSandbox Driver = "opensandbox"
|
|
||||||
)
|
|
||||||
|
|
||||||
type Placement struct {
|
|
||||||
Class WorkloadClass `json:"workload_class"`
|
|
||||||
Driver Driver `json:"driver"`
|
|
||||||
}
|
|
||||||
|
|
||||||
var (
|
|
||||||
KubernetesContainer = Placement{Class: WorkloadContainer, Driver: DriverKubernetes}
|
|
||||||
OpenSandboxVM = Placement{Class: WorkloadVM, Driver: DriverOpenSandbox}
|
|
||||||
)
|
|
||||||
|
|
||||||
func (p Placement) Validate() error {
|
|
||||||
switch p {
|
|
||||||
case KubernetesContainer, OpenSandboxVM:
|
|
||||||
return nil
|
|
||||||
default:
|
|
||||||
return fmt.Errorf("unsupported workload placement %s/%s", p.Class, p.Driver)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (p Placement) Key() string { return string(p.Class) + "." + string(p.Driver) }
|
|
||||||
|
|
||||||
func PlacementFromLabels(labels []string) (Placement, error) {
|
|
||||||
if !slices.Contains(labels, "self-hosted") {
|
|
||||||
return Placement{}, fmt.Errorf("task runs-on labels must include self-hosted: %v", labels)
|
|
||||||
}
|
|
||||||
legacyPod := slices.Contains(labels, "pod")
|
|
||||||
container := slices.Contains(labels, string(WorkloadContainer))
|
|
||||||
vm := slices.Contains(labels, string(WorkloadVM)) || slices.Contains(labels, "vm-dev")
|
|
||||||
kubernetes := slices.Contains(labels, string(DriverKubernetes))
|
|
||||||
opensandbox := slices.Contains(labels, string(DriverOpenSandbox))
|
|
||||||
|
|
||||||
if legacyPod {
|
|
||||||
if container || vm || kubernetes || opensandbox {
|
|
||||||
return Placement{}, errors.New("legacy pod label cannot be combined with workload or VM driver labels")
|
|
||||||
}
|
|
||||||
return KubernetesContainer, nil
|
|
||||||
}
|
|
||||||
if container == vm {
|
|
||||||
return Placement{}, fmt.Errorf("task runs-on labels must select exactly one workload class: %v", labels)
|
|
||||||
}
|
|
||||||
if kubernetes && opensandbox {
|
|
||||||
return Placement{}, fmt.Errorf("task runs-on labels select multiple drivers: %v", labels)
|
|
||||||
}
|
|
||||||
if container {
|
|
||||||
if opensandbox {
|
|
||||||
return Placement{}, fmt.Errorf("opensandbox does not support container workloads")
|
|
||||||
}
|
|
||||||
return KubernetesContainer, nil
|
|
||||||
}
|
|
||||||
if kubernetes {
|
|
||||||
return Placement{}, fmt.Errorf("kubernetes does not support VM workloads")
|
|
||||||
}
|
|
||||||
return OpenSandboxVM, nil
|
|
||||||
}
|
|
||||||
@@ -71,28 +71,6 @@ 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
|
||||||
@@ -110,7 +88,6 @@ 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
|
||||||
@@ -119,18 +96,13 @@ 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
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -206,11 +178,10 @@ 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/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/workload-class": string(assignment.Placement.Class),
|
"ci.ddupan.top/backend": string(assignment.Backend),
|
||||||
"ci.ddupan.top/driver": string(assignment.Placement.Driver),
|
|
||||||
},
|
},
|
||||||
Annotations: map[string]string{
|
Annotations: map[string]string{
|
||||||
"ci.ddupan.top/repository": assignment.Identity.Repository,
|
"ci.ddupan.top/repository": assignment.Identity.Repository,
|
||||||
|
|||||||
@@ -53,9 +53,9 @@ func (t *fakeTasks) Report(_ context.Context, _ int64, phase Phase) error {
|
|||||||
|
|
||||||
func assignment() taskassignment.Assignment {
|
func assignment() taskassignment.Assignment {
|
||||||
return taskassignment.Assignment{
|
return taskassignment.Assignment{
|
||||||
ID: "gitea-task-42",
|
ID: "gitea-task-42",
|
||||||
Placement: taskassignment.KubernetesContainer,
|
Backend: taskassignment.BackendPod,
|
||||||
Task: &runnerv1.Task{Id: 42},
|
Task: &runnerv1.Task{Id: 42},
|
||||||
Identity: taskidentity.Identity{
|
Identity: taskidentity.Identity{
|
||||||
Repository: "owner/repo",
|
Repository: "owner/repo",
|
||||||
Task: "publish",
|
Task: "publish",
|
||||||
@@ -79,8 +79,7 @@ func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
|
|||||||
|
|
||||||
func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
||||||
backend := &fakeBackend{}
|
backend := &fakeBackend{}
|
||||||
var events []Event
|
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
|
||||||
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 {
|
||||||
@@ -89,18 +88,6 @@ 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,7 +10,6 @@ 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
|
||||||
|
|||||||
@@ -1,68 +0,0 @@
|
|||||||
#!/usr/bin/env bash
|
|
||||||
set -euo pipefail
|
|
||||||
set +x
|
|
||||||
|
|
||||||
: "${GITHUB_SHA:?GITHUB_SHA is required}"
|
|
||||||
: "${IMAGE_NAME:?IMAGE_NAME is required}"
|
|
||||||
: "${IMAGE_REPOSITORY:?IMAGE_REPOSITORY is required}"
|
|
||||||
: "${IMAGE_DOCKERFILE:?IMAGE_DOCKERFILE is required}"
|
|
||||||
: "${PUSH_REGISTRY:?PUSH_REGISTRY is required}"
|
|
||||||
: "${PULL_REGISTRY:?PULL_REGISTRY is required}"
|
|
||||||
: "${SPIRE_AGENT_SOCKET:?SPIRE_AGENT_SOCKET is required}"
|
|
||||||
|
|
||||||
# Bootstrap the image that first introduces automatic Docker setup. Once that
|
|
||||||
# runner is deployed, the job-started hook makes this an idempotent no-op.
|
|
||||||
sudo scripts/setup-job-docker
|
|
||||||
|
|
||||||
image_tag="sha-${GITHUB_SHA}"
|
|
||||||
metadata="${IMAGE_NAME}-metadata.json"
|
|
||||||
docker_config=$(mktemp -d)
|
|
||||||
jwt_file=$(mktemp)
|
|
||||||
cleanup() {
|
|
||||||
docker buildx rm ci-builder >/dev/null 2>&1 || true
|
|
||||||
rm -rf -- "$docker_config" "$jwt_file"
|
|
||||||
}
|
|
||||||
trap cleanup EXIT
|
|
||||||
export DOCKER_CONFIG="$docker_config"
|
|
||||||
|
|
||||||
/opt/spire/bin/spire-agent api fetch jwt \
|
|
||||||
-audience zot \
|
|
||||||
-socketPath "$SPIRE_AGENT_SOCKET" \
|
|
||||||
-output json >"$jwt_file"
|
|
||||||
# shellcheck disable=SC2016
|
|
||||||
jq -er '.[0].svids[0].svid' "$jwt_file" | \
|
|
||||||
docker login "$PUSH_REGISTRY" --username zot --password-stdin
|
|
||||||
|
|
||||||
docker buildx create \
|
|
||||||
--name ci-builder \
|
|
||||||
--driver docker-container \
|
|
||||||
--use
|
|
||||||
|
|
||||||
docker buildx build \
|
|
||||||
--builder ci-builder \
|
|
||||||
--platform linux/amd64 \
|
|
||||||
--file "$IMAGE_DOCKERFILE" \
|
|
||||||
--tag "${PUSH_REGISTRY}/${IMAGE_REPOSITORY}:${image_tag}" \
|
|
||||||
--tag "${PUSH_REGISTRY}/${IMAGE_REPOSITORY}:main" \
|
|
||||||
--provenance=mode=max \
|
|
||||||
--sbom=true \
|
|
||||||
--metadata-file "$metadata" \
|
|
||||||
--push \
|
|
||||||
.
|
|
||||||
|
|
||||||
image_digest=$(
|
|
||||||
# shellcheck disable=SC2016
|
|
||||||
jq -er '."containerimage.digest"' "$metadata"
|
|
||||||
)
|
|
||||||
image_ref="${PULL_REGISTRY}/${IMAGE_REPOSITORY}@${image_digest}"
|
|
||||||
|
|
||||||
printf '%s=%s\n' "$IMAGE_NAME" "$image_ref"
|
|
||||||
if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then
|
|
||||||
{
|
|
||||||
printf '## Published image\n\n'
|
|
||||||
# shellcheck disable=SC2016
|
|
||||||
printf -- '- %s: `%s`\n' "$IMAGE_NAME" "$image_ref"
|
|
||||||
# shellcheck disable=SC2016
|
|
||||||
printf -- '- Source: `%s`\n' "$GITHUB_SHA"
|
|
||||||
} >>"$GITHUB_STEP_SUMMARY"
|
|
||||||
fi
|
|
||||||
@@ -1,60 +0,0 @@
|
|||||||
#!/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 'nohup dockerd "$@" </dev/null >/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