Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fef7e5a214
|
||
|
|
51b940468e
|
||
|
|
b614ef4c2f
|
||
|
|
8fa8e46320
|
||
|
|
cc4405f788
|
||
|
|
4a428c4384 | ||
|
|
9a28a1573e
|
||
|
|
01995bf084 | ||
|
|
3d787b2dd3
|
||
|
|
54661411e3 | ||
|
|
53a080b310
|
||
|
|
38e8d59541
|
||
|
|
661b5e9218 |
@@ -17,7 +17,7 @@ jobs:
|
||||
- run: go vet ./...
|
||||
|
||||
python:
|
||||
runs-on: self-hosted
|
||||
runs-on: [self-hosted, pod]
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: actions/setup-python@v5
|
||||
@@ -28,7 +28,7 @@ jobs:
|
||||
- run: python -m compileall -q src
|
||||
|
||||
shell:
|
||||
runs-on: self-hosted
|
||||
runs-on: [self-hosted, pod]
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- run: |
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
---
|
||||
name: VM kind smoke
|
||||
|
||||
# yamllint disable-line rule:truthy
|
||||
on:
|
||||
workflow_dispatch:
|
||||
|
||||
@@ -7,25 +9,133 @@ jobs:
|
||||
kind:
|
||||
runs-on: [self-hosted, vm-dev]
|
||||
steps:
|
||||
- name: Start job-local Docker
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
sudo install -d /var/lib/docker
|
||||
sudo truncate -s 20G /tmp/docker-data.img
|
||||
sudo mkfs.ext4 -F /tmp/docker-data.img
|
||||
sudo mount -o loop /tmp/docker-data.img /var/lib/docker
|
||||
|
||||
# A nested systemd needs a domain cgroup namespace. This is the
|
||||
# cgroup v2 nesting initialization performed by the official DinD
|
||||
# entrypoint, kept here because the shared runner image deliberately
|
||||
# does not carry a second DinD-specific entrypoint.
|
||||
if [[ -f /sys/fs/cgroup/cgroup.controllers ]]; then
|
||||
sudo sh -c '
|
||||
mkdir -p /sys/fs/cgroup/init
|
||||
while read -r pid; do
|
||||
printf "%s\n" "$pid" \
|
||||
>/sys/fs/cgroup/init/cgroup.procs 2>/dev/null || true
|
||||
done </sys/fs/cgroup/cgroup.procs
|
||||
sed -e "s/ / +/g" -e "s/^/+/" \
|
||||
/sys/fs/cgroup/cgroup.controllers \
|
||||
>/sys/fs/cgroup/cgroup.subtree_control
|
||||
'
|
||||
fi
|
||||
|
||||
sudo nohup dockerd \
|
||||
--host=unix:///var/run/docker.sock \
|
||||
--storage-driver=overlay2 \
|
||||
>/tmp/dockerd.log 2>&1 &
|
||||
for _ in {1..60}; do
|
||||
if docker info >/dev/null 2>&1; then
|
||||
exit 0
|
||||
fi
|
||||
sleep 1
|
||||
done
|
||||
cat /tmp/dockerd.log
|
||||
exit 1
|
||||
|
||||
- name: Verify Docker
|
||||
run: docker info
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
docker info
|
||||
echo '### runner cgroup'
|
||||
cat /proc/self/cgroup
|
||||
cat /sys/fs/cgroup/cgroup.type
|
||||
cat /sys/fs/cgroup/cgroup.controllers
|
||||
cat /sys/fs/cgroup/cgroup.subtree_control
|
||||
echo '### nested private cgroup namespace'
|
||||
docker run --rm --privileged --cgroupns=private alpine:3.22 \
|
||||
sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
|
||||
echo '### nested host cgroup namespace'
|
||||
docker run --rm --privileged --cgroupns=host alpine:3.22 \
|
||||
sh -c 'cat /proc/self/cgroup; cat /sys/fs/cgroup/cgroup.type'
|
||||
|
||||
- name: Install kind
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
version=v0.33.0
|
||||
base_url="https://kind.sigs.k8s.io/dl/${version}"
|
||||
curl --fail --location --silent --show-error \
|
||||
--output /tmp/kind "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64"
|
||||
--output /tmp/kind "${base_url}/kind-linux-amd64"
|
||||
curl --fail --location --silent --show-error \
|
||||
--output /tmp/kind.sha256sum "https://kind.sigs.k8s.io/dl/${version}/kind-linux-amd64.sha256sum"
|
||||
printf '%s %s\n' "$(cut -d ' ' -f1 /tmp/kind.sha256sum)" /tmp/kind | sha256sum --check
|
||||
--output /tmp/kind.sha256sum \
|
||||
"${base_url}/kind-linux-amd64.sha256sum"
|
||||
checksum=$(cut -d ' ' -f1 /tmp/kind.sha256sum)
|
||||
printf '%s %s\n' "$checksum" /tmp/kind | sha256sum --check
|
||||
chmod 0755 /tmp/kind
|
||||
|
||||
- name: Create and delete kind cluster
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
trap '/tmp/kind delete cluster --name smoke' EXIT
|
||||
/tmp/kind create cluster --name smoke --wait 180s
|
||||
diagnose_and_cleanup() {
|
||||
status=$?
|
||||
node=smoke-control-plane
|
||||
if (( status != 0 )) && docker inspect "$node" >/dev/null 2>&1; then
|
||||
echo '::group::kind node inspect'
|
||||
docker inspect "$node"
|
||||
echo '::endgroup::'
|
||||
echo '::group::kind node logs'
|
||||
docker logs "$node" 2>&1 || true
|
||||
echo '::endgroup::'
|
||||
echo '::group::kind node guest state'
|
||||
docker exec "$node" bash -c '
|
||||
set +e
|
||||
echo "### pid 1"
|
||||
ps -p 1 -o pid,ppid,user,stat,comm,args
|
||||
cat /proc/1/status
|
||||
echo "### cgroup"
|
||||
cat /proc/1/cgroup
|
||||
findmnt -R /sys/fs/cgroup
|
||||
stat -fc "%T %a" /sys/fs/cgroup
|
||||
echo "### systemd"
|
||||
systemctl --no-pager --failed
|
||||
systemctl --no-pager status \
|
||||
multi-user.target containerd.service kubelet.service
|
||||
echo "### CRI containers"
|
||||
endpoint=unix:///run/containerd/containerd.sock
|
||||
crictl --runtime-endpoint "$endpoint" ps --all
|
||||
for id in $(
|
||||
crictl --runtime-endpoint "$endpoint" ps --all --quiet
|
||||
); do
|
||||
echo "### CRI container $id"
|
||||
crictl --runtime-endpoint "$endpoint" inspect "$id"
|
||||
crictl --runtime-endpoint "$endpoint" logs "$id"
|
||||
done
|
||||
echo "### containerd metadata"
|
||||
timeout 10 ctr --namespace k8s.io containers list
|
||||
timeout 10 ctr --namespace k8s.io snapshots list
|
||||
echo "### runtime process stacks"
|
||||
ps -e -o pid,ppid,stat,wchan:32,comm,args
|
||||
for pid in $(pidof containerd containerd-shim-runc-v2); do
|
||||
echo "### kernel stack $pid"
|
||||
cat "/proc/$pid/stack"
|
||||
done
|
||||
kill -USR1 "$(pidof containerd)"
|
||||
sleep 2
|
||||
journalctl --no-pager -b -n 500
|
||||
' 2>&1 || true
|
||||
echo '::endgroup::'
|
||||
fi
|
||||
/tmp/kind delete cluster --name smoke || true
|
||||
exit "$status"
|
||||
}
|
||||
trap diagnose_and_cleanup EXIT
|
||||
/tmp/kind create cluster --name smoke --wait 180s --retain
|
||||
/tmp/kind get clusters | grep -Fx smoke
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
---
|
||||
name: VM runtime smoke
|
||||
|
||||
# yamllint disable-line rule:truthy
|
||||
on:
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
runtime:
|
||||
runs-on: [self-hosted, vm-dev]
|
||||
steps:
|
||||
- name: Verify workload identity socket
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
: "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}"
|
||||
socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://}
|
||||
test -S "$socket_path"
|
||||
|
||||
- name: Verify Docker daemon
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
format='{{json .ServerVersion}} {{json .Driver}}'
|
||||
format="$format {{json .CgroupVersion}}"
|
||||
docker info --format "$format"
|
||||
|
||||
- name: Run and clean nested container
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
name=vm-runtime-smoke
|
||||
trap 'docker rm --force "$name" >/dev/null 2>&1 || true' EXIT
|
||||
output=$(docker run --name "$name" alpine:3.22 /bin/sh -c \
|
||||
'test "$(uname -m)" = x86_64 && printf vm-runtime-ok')
|
||||
test "$output" = vm-runtime-ok
|
||||
@@ -4,7 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"slices"
|
||||
@@ -139,7 +139,7 @@ func runController(ctx context.Context) error {
|
||||
JetStream: producerJS, SubjectBase: config.SubjectBase,
|
||||
}},
|
||||
Config: taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels, Capacity: config.PodCapacity + config.VMCapacity},
|
||||
OnError: func(err error) { log.Printf("scheduler: %v", err) },
|
||||
OnError: func(err error) { slog.Error("scheduler error", "component", "scheduler", "error", err) },
|
||||
}
|
||||
kubernetesConfig, err := rest.InClusterConfig()
|
||||
if err != nil {
|
||||
@@ -190,11 +190,13 @@ func runController(ctx context.Context) error {
|
||||
return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
|
||||
}
|
||||
}
|
||||
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, podPool)
|
||||
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendPod)}, registry, podPool)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) { log.Printf("pod lifecycle: %v", err) }}
|
||||
lifecycle := podbackend.Lifecycle{Backend: backend, OnError: func(err error) {
|
||||
slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendPod, "error", err)
|
||||
}}
|
||||
components[controller.PodWorker] = runComponent(func(ctx context.Context) error {
|
||||
group, groupContext := errgroup.WithContext(ctx)
|
||||
group.Go(func() error { return component.Run(groupContext) })
|
||||
@@ -220,11 +222,13 @@ func runController(ctx context.Context) error {
|
||||
return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
|
||||
}
|
||||
}
|
||||
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry, vmPool)
|
||||
component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap, OnEvent: workerEventLogger(taskassignment.BackendVM)}, registry, vmPool)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) { log.Printf("VM lifecycle: %v", err) }}
|
||||
lifecycleReconciler := opensandboxbackend.LifecycleReconciler{Backend: backend, OnError: func(err error) {
|
||||
slog.Error("backend lifecycle error", "component", "lifecycle", "backend", taskassignment.BackendVM, "error", err)
|
||||
}}
|
||||
components[controller.VMWorker] = runComponent(func(ctx context.Context) error {
|
||||
group, groupContext := errgroup.WithContext(ctx)
|
||||
group.Go(func() error { return component.Run(groupContext) })
|
||||
@@ -294,11 +298,21 @@ func workerComponent(ctx context.Context, js jetstream.JetStream, config control
|
||||
}
|
||||
return assignmentqueue.ConsumerComponent{
|
||||
Consumer: consumer, Capacity: capacity,
|
||||
Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission},
|
||||
OnError: func(err error) { log.Printf("%s worker: %v", backend, err) },
|
||||
Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission, OnEvent: func(event assignmentqueue.Event) {
|
||||
slog.Info("assignment transition", "component", "worker", "event", event.Name, "backend", event.Backend, "assignment", event.AssignmentID, "retry_delay", event.RetryDelay)
|
||||
}},
|
||||
OnError: func(err error) {
|
||||
slog.Error("assignment processing error", "component", "worker", "backend", backend, "error", err)
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func workerEventLogger(backend taskassignment.Backend) func(taskworker.Event) {
|
||||
return func(event taskworker.Event) {
|
||||
slog.Info("executor transition", "component", "worker", "event", event.Name, "backend", backend, "assignment", event.AssignmentID, "executor", event.Executor, "phase", event.Phase)
|
||||
}
|
||||
}
|
||||
|
||||
func loadControllerConfig() (controllerConfig, error) {
|
||||
selection, err := controller.ParseSelection(os.Getenv("COMPONENTS"))
|
||||
if err != nil {
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
@@ -12,8 +13,9 @@ import (
|
||||
)
|
||||
|
||||
func main() {
|
||||
slog.SetDefault(slog.New(slog.NewJSONHandler(os.Stderr, nil)))
|
||||
if err := run(); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
slog.Error("runner stopped", "error", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -40,9 +40,11 @@ UID attestation 的临时 Agent 失去父级。
|
||||
|
||||
`ci-vm` 使用 `kata-clh-runtime-rs`;`ci-pod` 使用默认 runc。两者都要求:
|
||||
|
||||
- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI 和 identity gate;
|
||||
- runner UID 2000,SPIRE Agent 与 privileged dockerd 使用不同 UID;
|
||||
- Docker socket 通过 group 2000 共享,Docker 数据仅存在于 sandbox emptyDir;
|
||||
- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI、Docker 工具和
|
||||
identity gate;Pod 与 VM backend 使用同一个镜像;
|
||||
- runner UID 2000;SPIRE Agent 独立运行,需要 Docker 的 workflow 通过 sudo 在
|
||||
privileged executor 内启动 job-local daemon;
|
||||
- Kata VM 中 Docker 数据使用 guest 内的 loop-backed ext4,并随 sandbox 一起删除;
|
||||
- `self-hosted` 必须是所有 runner labels 的前缀;
|
||||
- ephemeral/once runner 完成一项任务后退出。
|
||||
|
||||
|
||||
@@ -78,6 +78,22 @@ type Processor struct {
|
||||
Admission Admission
|
||||
RetryDelay time.Duration
|
||||
ClaimTimeout time.Duration
|
||||
OnEvent func(Event)
|
||||
}
|
||||
|
||||
// Event describes a non-sensitive assignment handoff transition. It never
|
||||
// contains task payloads, credentials, capabilities, or workload identities.
|
||||
type Event struct {
|
||||
Name string
|
||||
AssignmentID string
|
||||
Backend taskassignment.Backend
|
||||
RetryDelay time.Duration
|
||||
}
|
||||
|
||||
func (p Processor) event(name string, assignment taskassignment.Assignment, retryDelay time.Duration) {
|
||||
if p.OnEvent != nil {
|
||||
p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Backend: assignment.Backend, RetryDelay: retryDelay})
|
||||
}
|
||||
}
|
||||
|
||||
func (p Processor) Process(ctx context.Context, message Message) error {
|
||||
@@ -88,6 +104,7 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
||||
if err != nil {
|
||||
return errors.Join(err, message.TermWithReason("invalid assignment"))
|
||||
}
|
||||
p.event("received", assignment, 0)
|
||||
if _, err := p.Claims.Offer(assignment); err != nil {
|
||||
return errors.Join(err, message.TermWithReason("conflicting assignment"))
|
||||
}
|
||||
@@ -96,8 +113,10 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
||||
if delay <= 0 {
|
||||
delay = 2 * time.Second
|
||||
}
|
||||
p.event("capacity_wait", assignment, delay)
|
||||
return message.NakWithDelay(delay)
|
||||
}
|
||||
p.event("capacity_acquired", assignment, 0)
|
||||
accepted, err := p.Accepter.Accept(ctx, assignment)
|
||||
if err != nil {
|
||||
p.Admission.Release(assignment.ID)
|
||||
@@ -105,9 +124,11 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
||||
if delay <= 0 {
|
||||
delay = 15 * time.Second
|
||||
}
|
||||
p.event("backend_retry", assignment, delay)
|
||||
return errors.Join(err, message.NakWithDelay(delay))
|
||||
}
|
||||
if accepted {
|
||||
p.event("backend_ready", assignment, 0)
|
||||
timeout := p.ClaimTimeout
|
||||
if timeout <= 0 {
|
||||
timeout = 4 * time.Minute
|
||||
@@ -120,17 +141,21 @@ func (p Processor) Process(ctx context.Context, message Message) error {
|
||||
if delay <= 0 {
|
||||
delay = 2 * time.Second
|
||||
}
|
||||
p.event("claim_timeout", assignment, delay)
|
||||
return errors.Join(err, message.NakWithDelay(delay))
|
||||
}
|
||||
p.event("runner_claimed", assignment, 0)
|
||||
if err := message.DoubleAck(ctx); err != nil {
|
||||
return fmt.Errorf("ack assignment %s: %w", assignment.ID, err)
|
||||
}
|
||||
p.event("acked", assignment, 0)
|
||||
return nil
|
||||
}
|
||||
delay := p.RetryDelay
|
||||
if delay <= 0 {
|
||||
delay = 2 * time.Second
|
||||
}
|
||||
p.event("backend_pending", assignment, delay)
|
||||
return message.NakWithDelay(delay)
|
||||
}
|
||||
|
||||
|
||||
@@ -126,13 +126,23 @@ func encodedAssignment(t *testing.T) []byte {
|
||||
|
||||
func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
|
||||
message := &fakeMessage{data: encodedAssignment(t)}
|
||||
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}}
|
||||
var events []Event
|
||||
processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}, OnEvent: func(event Event) { events = append(events, event) }}
|
||||
if err := processor.Process(context.Background(), message); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if message.acked != 1 || message.nacked != 0 {
|
||||
t.Fatalf("message = %#v", message)
|
||||
}
|
||||
want := []string{"received", "capacity_acquired", "backend_ready", "runner_claimed", "acked"}
|
||||
if len(events) != len(want) {
|
||||
t.Fatalf("events = %#v", events)
|
||||
}
|
||||
for index := range want {
|
||||
if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Backend != taskassignment.BackendPod {
|
||||
t.Fatalf("event[%d] = %#v", index, events[index])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
|
||||
|
||||
@@ -71,6 +71,28 @@ type Worker struct {
|
||||
Backend Backend
|
||||
Tasks TaskState
|
||||
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
|
||||
@@ -88,6 +110,7 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment
|
||||
return false, err
|
||||
}
|
||||
if executor == nil {
|
||||
w.event("executor_absent", assignment, nil)
|
||||
launch, launchErr := w.launchSpec(assignment)
|
||||
if launchErr != nil {
|
||||
return false, launchErr
|
||||
@@ -96,13 +119,18 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
w.event("executor_created", assignment, executor)
|
||||
} else {
|
||||
w.event("executor_found", assignment, executor)
|
||||
}
|
||||
if executor.IdentityTarget == "" {
|
||||
w.event("identity_target_pending", assignment, executor)
|
||||
return false, nil
|
||||
}
|
||||
if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
|
||||
return false, err
|
||||
}
|
||||
w.event("identity_bound", assignment, executor)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -79,7 +79,8 @@ func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
|
||||
|
||||
func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
||||
backend := &fakeBackend{}
|
||||
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}}
|
||||
var events []Event
|
||||
worker := Worker{Backend: backend, Bootstrap: fakeBootstrap{}, OnEvent: func(event Event) { events = append(events, event) }}
|
||||
|
||||
accepted, err := worker.Accept(context.Background(), assignment())
|
||||
if err != nil || !accepted {
|
||||
@@ -88,6 +89,18 @@ func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) {
|
||||
if backend.created != 1 || backend.bound != 1 || backend.deleted != 0 {
|
||||
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) {
|
||||
|
||||
Reference in New Issue
Block a user