Skip to content

Commit eac6cad

Browse files
authored
Merge pull request #21 from appwrite-labs/fix/fail-invalid-config
fix(orchestrator): fail the run when a job can never start instead of leaving it queued
2 parents adb3033 + 4bd372f commit eac6cad

11 files changed

Lines changed: 304 additions & 38 deletions

File tree

‎SPEC.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -191,7 +191,7 @@ Multiple layers prevent orphaned VMs:
191191
- **Minimal IAM**: Cloud Function service account has only `compute.instanceAdmin.v1` on the runner subnet, `iam.serviceAccountUser` on the runner SA. Runner SA has only GCS access (cache bucket) and `compute.instances.delete` on itself.
192192
- **Ephemeral runners**: Every VM handles exactly one job. No shared state between jobs. No persistent disk reuse.
193193
- **No ingress**: Runner VMs have no public IP by default (configurable). All communication is outbound to GitHub and GCS.
194-
- **GitHub App scoping**: The GitHub App only needs `actions:write` (to re-run preempted jobs) and `administration:write` (for runner registration). No code access.
194+
- **GitHub App scoping**: The GitHub App only needs `actions:write` (to re-run preempted jobs and cancel runs that can never start), `checks:write` (to report why) and `administration:write` (for runner registration). No code access.
195195
- **Secrets in Secret Manager**: GitHub App private key and webhook secret stored in Secret Manager, not in Terraform state.
196196

197197
---

‎orchestrator/cmd/setup/main.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ func main() {
5050
DefaultPermissions: map[string]string{
5151
"actions": "write",
5252
"administration": "write",
53+
"checks": "write",
5354
"contents": "read",
5455
},
5556
DefaultEvents: []string{"workflow_job"},

‎orchestrator/configuration_test.go‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
package function
2+
3+
import (
4+
"context"
5+
"errors"
6+
"fmt"
7+
"net/http"
8+
"net/http/httptest"
9+
"strings"
10+
"testing"
11+
12+
computepb "cloud.google.com/go/compute/apiv1/computepb"
13+
"google.golang.org/api/option"
14+
)
15+
16+
// engine is what Compute Engine answers: zones from discovery (nil when it
17+
// fails), each zone's catalogue (absent when unreadable), each insert's error.
18+
type engine struct {
19+
zones []string
20+
catalogue map[string][]*MachineTypeInfo
21+
insert map[string]error
22+
}
23+
24+
var fourVCPUs = []*MachineTypeInfo{{Name: "c3d-highcpu-4", Family: "c3d", VCPUs: 4, MemoryMB: 8192}}
25+
26+
// queuedJob delivers the queued task for job "manifest" of run 7 asking for
27+
// preset "deploy", with the repository's config file and Compute as given,
28+
// and reports the status handed back.
29+
func queuedJob(t *testing.T, gh *github, config string, gce engine) int {
30+
t.Helper()
31+
t.Setenv("GCRUNNER_ZONES", "")
32+
t.Setenv("GCE_REGION", "europe-west3")
33+
t.Setenv("GCP_PROJECT", "p")
34+
useGitHub(t, gh)
35+
originalFetch, originalZones, originalOptions := fetchRepositoryFile, listZones, computeOptions
36+
t.Cleanup(func() {
37+
fetchRepositoryFile, listZones, computeOptions = originalFetch, originalZones, originalOptions
38+
configCache.configs = map[string]configCacheEntry{}
39+
zoneCache.zones = map[string]zoneCacheEntry{}
40+
machineTypeCache.types = map[string]machineTypeCacheEntry{}
41+
})
42+
configCache.configs = map[string]configCacheEntry{}
43+
zoneCache.zones = map[string]zoneCacheEntry{}
44+
machineTypeCache.types = map[string]machineTypeCacheEntry{}
45+
zoneOffset.Store(0)
46+
empty := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { fmt.Fprint(w, `{"items": {}}`) }))
47+
t.Cleanup(empty.Close)
48+
computeOptions = []option.ClientOption{option.WithEndpoint(empty.URL), option.WithoutAuthentication()}
49+
fetchRepositoryFile = func(context.Context, string, string, string, string) ([]byte, error) { return []byte(config), nil }
50+
listZones = func(context.Context, string, string) ([]string, error) {
51+
if gce.zones == nil {
52+
return nil, errors.New("compute unavailable")
53+
}
54+
return gce.zones, nil
55+
}
56+
withMachineTypes(t, func(_ context.Context, _, zone string) ([]*MachineTypeInfo, error) {
57+
if types, ok := gce.catalogue[zone]; ok {
58+
return types, nil
59+
}
60+
return nil, errors.New("compute unavailable in " + zone)
61+
})
62+
withInsert(t, func(_ context.Context, zone string, _ *computepb.Instance) error { return gce.insert[zone] })
63+
return deliverTask(t, "/task/queued", WorkflowJobEvent{
64+
Action: "queued",
65+
WorkflowJob: WorkflowJob{ID: 42, Name: "manifest", RunID: 7, HeadSHA: "6d8b388", Labels: []string{"gcrunner=7/runner=deploy"}},
66+
Repository: cloudRepository,
67+
})
68+
}
69+
70+
const (
71+
twoVCPUsConfig = "runners:\n deploy:\n family: c3d\n cpu: 2\n ram: 8\n"
72+
exactNameConfig = "runners:\n deploy:\n machine: n2d-standard-999\n"
73+
)
74+
75+
func TestAJobThatCanNeverStartFailsItsRunOnce(t *testing.T) {
76+
missing := errors.New(missingMachineTypeError)
77+
two := engine{zones: []string{"europe-west3-a", "europe-west3-b"}, catalogue: map[string][]*MachineTypeInfo{"europe-west3-a": fourVCPUs, "europe-west3-b": fourVCPUs}}
78+
for name, c := range map[string]struct {
79+
config string
80+
gce engine
81+
reason string
82+
}{
83+
"preset not defined": {"runners:\n build:\n cpu: 4\n", two, `runner "deploy" is not defined`},
84+
"no zone offers the size": {twoVCPUsConfig, two, "no machine type matching"},
85+
"no zone offers the exact name": {exactNameConfig, engine{zones: two.zones, insert: map[string]error{"europe-west3-a": missing, "europe-west3-b": missing}}, "does not exist"},
86+
} {
87+
gh := &github{checkStatus: http.StatusCreated}
88+
if code := queuedJob(t, gh, c.config, c.gce); code != http.StatusOK {
89+
t.Errorf("%s: status %d, want 200 so Cloud Tasks does not retry", name, code)
90+
}
91+
if len(gh.checks) != 1 || gh.cancels != 1 {
92+
t.Fatalf("%s: %d check runs and %d cancels, want one of each", name, len(gh.checks), gh.cancels)
93+
}
94+
check := gh.checks[0]
95+
output, _ := check["output"].(map[string]any)
96+
summary, _ := output["summary"].(string)
97+
if check["head_sha"] != "6d8b388" || check["conclusion"] != "failure" || output["title"] != "manifest cannot start" || !strings.Contains(summary, c.reason) {
98+
t.Errorf("%s: check run %v does not carry the reason %q", name, check, c.reason)
99+
}
100+
}
101+
}
102+
103+
// Only Compute answering in every zone proves the workflow wrong. A zone
104+
// that cannot be read, a zone out of quota, guessed zones after discovery
105+
// failed, and GitHub itself being down all leave the task to be retried.
106+
func TestATransientFailureIsRetriedWithoutFailingTheRun(t *testing.T) {
107+
missing := errors.New(missingMachineTypeError)
108+
zones := []string{"europe-west3-a", "europe-west3-b"}
109+
for name, c := range map[string]struct {
110+
config string
111+
gce engine
112+
check int
113+
}{
114+
"one catalogue unreadable": {twoVCPUsConfig, engine{zones: zones, catalogue: map[string][]*MachineTypeInfo{"europe-west3-a": fourVCPUs}}, http.StatusCreated},
115+
"one zone out of quota": {exactNameConfig, engine{zones: zones, insert: map[string]error{"europe-west3-a": missing, "europe-west3-b": errors.New("QUOTA_EXCEEDED")}}, http.StatusCreated},
116+
"zones guessed": {twoVCPUsConfig, engine{catalogue: map[string][]*MachineTypeInfo{"europe-west3-a": fourVCPUs, "europe-west3-b": fourVCPUs, "europe-west3-c": fourVCPUs}}, http.StatusCreated},
117+
"GitHub down for the check": {twoVCPUsConfig, engine{zones: zones, catalogue: map[string][]*MachineTypeInfo{"europe-west3-a": fourVCPUs, "europe-west3-b": fourVCPUs}}, http.StatusBadGateway},
118+
} {
119+
gh := &github{checkStatus: c.check}
120+
if code := queuedJob(t, gh, c.config, c.gce); code != http.StatusInternalServerError {
121+
t.Errorf("%s: status %d, want 500 so Cloud Tasks retries", name, code)
122+
}
123+
if gh.cancels != 0 {
124+
t.Errorf("%s: cancelled the run", name)
125+
}
126+
}
127+
}
128+
129+
// Without checks: write the report cannot be made, but retrying would not
130+
// grant it, so the task is still acknowledged.
131+
func TestARefusedCheckRunIsNotRetried(t *testing.T) {
132+
gh := &github{checkStatus: http.StatusForbidden}
133+
if code := queuedJob(t, gh, "runners: {}\n", engine{}); code != http.StatusOK {
134+
t.Errorf("status %d, want 200", code)
135+
}
136+
if gh.cancels != 0 {
137+
t.Errorf("cancelled the run without having explained why")
138+
}
139+
}
140+
141+
func withMachineTypes(t *testing.T, fn func(ctx context.Context, project, zone string) ([]*MachineTypeInfo, error)) {
142+
t.Helper()
143+
original := listMachineTypes
144+
listMachineTypes = fn
145+
t.Cleanup(func() { listMachineTypes = original })
146+
}

‎orchestrator/github.go‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package function
22

33
import (
4+
"bytes"
45
"context"
56
"crypto/rsa"
67
"crypto/x509"
@@ -433,3 +434,59 @@ func githubRequest(ctx context.Context, method, endpoint, installationToken stri
433434
req.Header.Set("Accept", "application/vnd.github+json")
434435
return githubClient.Do(req)
435436
}
437+
438+
// failRun reports a job that can never start: a failed check on its commit
439+
// carrying the reason, then a cancel of the run so it does not wait a day.
440+
func failRun(ctx context.Context, owner, repo string, job WorkflowJob, reason error) error {
441+
installationToken, err := getInstallationToken(ctx, owner)
442+
if err != nil {
443+
return fmt.Errorf("get installation token: %w", err)
444+
}
445+
check, err := json.Marshal(map[string]any{
446+
"name": "gcrunner",
447+
"head_sha": job.HeadSHA,
448+
"status": "completed",
449+
"conclusion": "failure",
450+
"output": map[string]string{
451+
"title": job.Name + " cannot start",
452+
"summary": reason.Error() + ". Fix " + configPath + " or the runs-on label and re-run the workflow.",
453+
},
454+
})
455+
if err != nil {
456+
return err
457+
}
458+
endpoint := fmt.Sprintf("https://api.github.com/repos/%s/%s/check-runs", owner, repo)
459+
req, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(check))
460+
if err != nil {
461+
return err
462+
}
463+
req.Header.Set("Authorization", "Bearer "+installationToken)
464+
req.Header.Set("Accept", "application/vnd.github+json")
465+
req.Header.Set("Content-Type", "application/json")
466+
resp, err := githubClient.Do(req)
467+
if err := answered(resp, err, http.StatusCreated); err != nil {
468+
return fmt.Errorf("create check run: %w", err)
469+
}
470+
endpoint = fmt.Sprintf("https://api.github.com/repos/%s/%s/actions/runs/%d/cancel", owner, repo, job.RunID)
471+
resp, err = githubRequest(ctx, "POST", endpoint, installationToken)
472+
if err := answered(resp, err, http.StatusAccepted, http.StatusConflict); err != nil {
473+
return fmt.Errorf("cancel run %d: %w", job.RunID, err)
474+
}
475+
return nil
476+
}
477+
478+
// answered closes the response and returns a githubError unless its status is
479+
// one of those given.
480+
func answered(resp *http.Response, err error, statuses ...int) error {
481+
if err != nil {
482+
return err
483+
}
484+
defer resp.Body.Close()
485+
for _, status := range statuses {
486+
if resp.StatusCode == status {
487+
return nil
488+
}
489+
}
490+
respBody, _ := io.ReadAll(resp.Body)
491+
return &githubError{Status: resp.StatusCode, Body: string(respBody)}
492+
}

‎orchestrator/machines.go‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ package function
22

33
import (
44
"context"
5+
"errors"
56
"fmt"
67
"strconv"
78
"strings"
@@ -35,6 +36,11 @@ type machineTypeCacheEntry struct {
3536
fetchedAt time.Time
3637
}
3738

39+
var errNoMachineType = errors.New("no machine type matching constraints")
40+
41+
// Indirected so tests can supply a zone's catalogue without Compute.
42+
var listMachineTypes = fetchMachineTypes
43+
3844
var machineTypeCache = &MachineTypeCache{
3945
types: make(map[string]machineTypeCacheEntry),
4046
ttl: 1 * time.Hour,
@@ -57,7 +63,7 @@ func (c *MachineTypeCache) list(ctx context.Context, project, zone string) ([]*M
5763
return entry.types, nil
5864
}
5965

60-
types, err := fetchMachineTypes(ctx, project, zone)
66+
types, err := listMachineTypes(ctx, project, zone)
6167
if err != nil {
6268
return nil, err
6369
}
@@ -157,8 +163,7 @@ func ResolveMachineType(ctx context.Context, project, zone string, labels *Runne
157163
}
158164

159165
if best == nil {
160-
return "", fmt.Errorf("no machine type matching constraints (families=%v, cpu=%s, ram=%s) in zone %s",
161-
families, labels.CPU, labels.RAM, zone)
166+
return "", fmt.Errorf("%w (families=%v, cpu=%s, ram=%s) in zone %s", errNoMachineType, families, labels.CPU, labels.RAM, zone)
162167
}
163168

164169
return best.Name, nil

‎orchestrator/preemption_test.go‎

Lines changed: 52 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,15 @@ var (
2424
)
2525

2626
// github stands in for api.github.com: it answers the rerun request with
27-
// rerunStatus, lists the run's latest jobs from latest (name to attempts),
28-
// and counts the reruns it received.
27+
// rerunStatus and the check run with checkStatus, lists the run's latest
28+
// jobs from latest (name to attempts), and records what it was asked to do.
2929
type github struct {
3030
rerunStatus int
31+
checkStatus int
3132
latest map[string][]int
3233
reruns int
34+
cancels int
35+
checks []map[string]any
3336
}
3437

3538
func (g *github) RoundTrip(req *http.Request) (*http.Response, error) {
@@ -44,6 +47,20 @@ func (g *github) RoundTrip(req *http.Request) (*http.Response, error) {
4447
case req.URL.Path == "/repos/appwrite-labs/cloud/actions/jobs/42/rerun":
4548
g.reruns++
4649
return answer(g.rerunStatus, `{"message": "This workflow is already running"}`)
50+
case strings.HasSuffix(req.URL.Path, "/actions/runners/generate-jitconfig"):
51+
return answer(http.StatusCreated, `{"encoded_jit_config": "jit"}`)
52+
case strings.HasSuffix(req.URL.Path, "/actions/runners"):
53+
return answer(http.StatusOK, `{"runners": []}`)
54+
case req.URL.Path == "/repos/appwrite-labs/cloud/check-runs":
55+
var check map[string]any
56+
if err := json.NewDecoder(req.Body).Decode(&check); err != nil {
57+
return answer(http.StatusBadRequest, `{}`)
58+
}
59+
g.checks = append(g.checks, check)
60+
return answer(g.checkStatus, `{}`)
61+
case req.URL.Path == "/repos/appwrite-labs/cloud/actions/runs/7/cancel":
62+
g.cancels++
63+
return answer(http.StatusAccepted, ``)
4764
case req.URL.Path == "/repos/appwrite-labs/cloud/actions/runs/7/jobs" && req.URL.Query().Get("filter") == "latest":
4865
var jobs []string
4966
for name, attempts := range g.latest {
@@ -81,46 +98,60 @@ func operations(t *testing.T, preemptedAt ...time.Time) *httptest.Server {
8198
return server
8299
}
83100

84-
// failedJob delivers the completed task for a failed job "build" of run 7
85-
// that ran on gcrunner-7-42 and reports the status handed to Cloud Tasks.
86-
func failedJob(t *testing.T, attempt int, gh *github, preemptedAt ...time.Time) int {
101+
// useGitHub puts gh behind githubClient with App credentials that sign.
102+
func useGitHub(t *testing.T, gh *github) {
87103
t.Helper()
88-
t.Setenv("GCP_PROJECT", "p")
89104
key, err := rsa.GenerateKey(rand.Reader, 2048)
90105
if err != nil {
91106
t.Fatal(err)
92107
}
93108
keyPEM := string(pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)}))
94-
95-
originalSecret, originalTransport, originalOptions, originalDelete := loadSecret, githubClient.Transport, computeOptions, deleteVM
96-
t.Cleanup(func() {
97-
loadSecret, githubClient.Transport, computeOptions, deleteVM = originalSecret, originalTransport, originalOptions, originalDelete
98-
})
109+
originalSecret, originalTransport := loadSecret, githubClient.Transport
110+
t.Cleanup(func() { loadSecret, githubClient.Transport = originalSecret, originalTransport })
99111
loadSecret = func(_ context.Context, name string) (string, error) {
100112
return map[string]string{"gcrunner-app-id": "1", "gcrunner-private-key": keyPEM}[name], nil
101113
}
102114
githubClient.Transport = gh
115+
}
116+
117+
// deliverTask posts event to the task path as Cloud Tasks does and reports
118+
// the status handed back.
119+
func deliverTask(t *testing.T, path string, event WorkflowJobEvent) int {
120+
t.Helper()
121+
body, err := json.Marshal(event)
122+
if err != nil {
123+
t.Fatal(err)
124+
}
125+
request := httptest.NewRequest(http.MethodPost, path, strings.NewReader(string(body)))
126+
request.Header.Set("X-CloudTasks-TaskName", "task-42")
127+
response := httptest.NewRecorder()
128+
HandleTask(response, request)
129+
return response.Code
130+
}
131+
132+
// failedJob delivers the completed task for a failed job "build" of run 7
133+
// that ran on gcrunner-7-42 and reports the status handed to Cloud Tasks.
134+
func failedJob(t *testing.T, attempt int, gh *github, preemptedAt ...time.Time) int {
135+
t.Helper()
136+
t.Setenv("GCP_PROJECT", "p")
137+
useGitHub(t, gh)
138+
originalOptions, originalDelete := computeOptions, deleteVM
139+
t.Cleanup(func() { computeOptions, deleteVM = originalOptions, originalDelete })
103140
computeOptions = []option.ClientOption{option.WithEndpoint(operations(t, preemptedAt...).URL), option.WithoutAuthentication()}
104141
deleteVM = func(context.Context, string) error { return nil }
105142

106-
body, err := json.Marshal(WorkflowJobEvent{
143+
return deliverTask(t, "/task/completed", WorkflowJobEvent{
107144
Action: "completed",
108145
WorkflowJob: WorkflowJob{
109146
ID: 42, Name: "build", RunID: 7, RunAttempt: attempt, RunnerName: "gcrunner-7-42", Labels: gcrunnerLabels,
110147
Conclusion: "failure", StartedAt: jobStart, CompletedAt: jobEnd,
111148
},
112-
Repository: Repository{FullName: "appwrite-labs/cloud", Name: "cloud", Owner: RepositoryOwner{Login: "appwrite-labs"}},
149+
Repository: cloudRepository,
113150
})
114-
if err != nil {
115-
t.Fatal(err)
116-
}
117-
request := httptest.NewRequest(http.MethodPost, "/task/completed", strings.NewReader(string(body)))
118-
request.Header.Set("X-CloudTasks-TaskName", "task-42")
119-
response := httptest.NewRecorder()
120-
HandleTask(response, request)
121-
return response.Code
122151
}
123152

153+
var cloudRepository = Repository{FullName: "appwrite-labs/cloud", Name: "cloud", Owner: RepositoryOwner{Login: "appwrite-labs"}}
154+
124155
func TestOnlyAJobPreemptedWhileRunningIsRerun(t *testing.T) {
125156
during := jobStart.Add(5 * time.Minute)
126157
for name, c := range map[string]struct {

0 commit comments

Comments
 (0)