vole
Clone
Read only, no credentials:
https://rodent.software/vole.git
Write over SSH with a registered key:
git clone ssh://git@rodent.software:2222/vole.git
A free and open-source forge that lives in your Git repo.
|
c5902451
Keenan Nemetz
16d ago
|
1 | package ssh |
|
b44cbfaf
Keenan Nemetz
17d ago
|
2 | |
| 3 | import ( | |
| 4 | "encoding/json" | |
|
78cd3437
Keenan Nemetz
13d ago
|
5 | "errors" |
|
213f4d6d
Keenan Nemetz
17d ago
|
6 | "fmt" |
|
78cd3437
Keenan Nemetz
13d ago
|
7 | "io" |
|
b44cbfaf
Keenan Nemetz
17d ago
|
8 | "time" |
| 9 | ||
|
c5902451
Keenan Nemetz
16d ago
|
10 | xssh "golang.org/x/crypto/ssh" |
|
d37d301f
Keenan Nemetz
19h ago
|
11 | "rodent.software/vole/ci/runs" |
| 12 | "rodent.software/vole/core/id" | |
| 13 | "rodent.software/vole/core/lfs" | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
14 | ) |
| 15 | ||
| 16 | // The runner protocol. | |
| 17 | // | |
| 18 | // A runner connects over the same SSH transport as everything else, so it | |
| 19 | // authenticates with a key and needs no token: the account it connects as decides | |
| 20 | // which repositories it may run jobs for. One JSON object per line, one reply per | |
| 21 | // request, on a single channel. | |
| 22 | // | |
|
90fdaf39
Keenan Nemetz
17d ago
|
23 | // {"op":"lease"} -> {"run":{...}} or {} |
| 24 | // {"op":"heartbeat","run":"<id>"} -> {"ok":true|false} | |
|
b312e166
Keenan Nemetz
11d ago
|
25 | // {"op":"finish","run":"<id>","record":{...}} -> {"ok":true|false} |
|
90fdaf39
Keenan Nemetz
17d ago
|
26 | // |
|
b312e166
Keenan Nemetz
11d ago
|
27 | // A finish carries the attempt's record: the runner uploads its log and artifacts, |
| 28 | // then sends the record on this channel. The server checks that it names the | |
| 29 | // account that holds the lease, that it describes the job the queue handed out, and | |
| 30 | // that every log and artifact pointer exists in the object store. It then stores the | |
| 31 | // attempt and the run's result in one transaction. The record is not a git commit; | |
| 32 | // CI history is instance-local. | |
| 33 | // | |
| 34 | // A tagged build that publishes release artifacts still pushes the release pointers | |
| 35 | // to the metadata branch after the record is stored. Releases are product metadata, | |
| 36 | // not CI history. | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
37 | // |
| 38 | // "ok":false means the lease is gone — another runner has the job — and the runner | |
|
b312e166
Keenan Nemetz
11d ago
|
39 | // must stop rather than report a result nobody will record. A retry of the same |
| 40 | // record after a lost reply is answered "ok":true, because the bytes are already | |
| 41 | // stored and the attempt is append-only. | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
42 | |
|
34afc569
Keenan Nemetz
17d ago
|
43 | const ( |
| 44 | // eligibleFor is how long a session reuses its repository list. Access rules live | |
| 45 | // in the instance database and only change when an admin edits them, so a runner | |
| 46 | // re-derives its list on the roster's five-second beat instead of on every poll. | |
| 47 | eligibleFor = 5 * time.Second | |
| 48 | // maxLeaseWait caps what a runner may ask for. The wait is a hint — the lease | |
| 49 | // remains the database's decision — so all a long wait holds is one SSH channel. | |
| 50 | maxLeaseWait = 60 * time.Second | |
|
b312e166
Keenan Nemetz
11d ago
|
51 | // maxRunnerRequestBytes bounds one request line. A lease is small, but a finish |
| 52 | // carries the attempt's steps and artifact names, so the bound has room for a | |
| 53 | // run at the artifact cap. It is per line, so a long-lived runner is not limited | |
| 54 | // in total. | |
| 55 | maxRunnerRequestBytes = 4 << 20 | |
|
34afc569
Keenan Nemetz
17d ago
|
56 | ) |
| 57 | ||
|
78cd3437
Keenan Nemetz
13d ago
|
58 | // errRequestTooLarge ends a runner session whose request line outgrew the bound. |
| 59 | var errRequestTooLarge = errors.New("runner request line is too large") | |
| 60 | ||
| 61 | // lineLimitReader fails once a single newline-terminated line passes limit bytes. | |
| 62 | // It counts bytes as they cross the reader, so json.Decoder's internal buffering | |
| 63 | // cannot hide a large request, and it resets at every newline so one oversized | |
| 64 | // line is the only thing it refuses. | |
| 65 | type lineLimitReader struct { | |
| 66 | r io.Reader | |
| 67 | limit int64 | |
| 68 | line int64 | |
| 69 | } | |
| 70 | ||
| 71 | func (l *lineLimitReader) Read(p []byte) (int, error) { | |
| 72 | n, err := l.r.Read(p) | |
| 73 | for _, b := range p[:n] { | |
| 74 | if b == '\n' { | |
| 75 | l.line = 0 | |
| 76 | continue | |
| 77 | } | |
| 78 | l.line++ | |
| 79 | if l.line > l.limit { | |
| 80 | return 0, errRequestTooLarge | |
| 81 | } | |
| 82 | } | |
| 83 | return n, err | |
| 84 | } | |
| 85 | ||
|
b44cbfaf
Keenan Nemetz
17d ago
|
86 | type runnerRequest struct { |
|
90fdaf39
Keenan Nemetz
17d ago
|
87 | Op string `json:"op"` |
| 88 | Run string `json:"run"` | |
|
b238ac24
Keenan Nemetz
17d ago
|
89 | // Repos narrows this session's leases to the repositories the runner was started |
| 90 | // for (--repos). It is applied as an intersection with what the account's runner | |
| 91 | // allowlists grant, so it can only subtract: a runner cannot ask its way into a | |
| 92 | // repository no allowlist names. Empty means "no narrowing", which keeps the list | |
| 93 | // following allowlist edits instead of freezing it at connect time. | |
| 94 | Repos []string `json:"repos,omitempty"` | |
|
6c03e461
Keenan Nemetz
16d ago
|
95 | // Labels are what this machine is (--label), and a job is only leased to a runner |
| 96 | // carrying every label the job asked for with runs-on. They describe a machine, not | |
| 97 | // a permission: a runner that claims a label it does not have only hurts itself, | |
| 98 | // which is why this is a client claim where --repos is server-enforced. | |
| 99 | Labels []string `json:"labels,omitempty"` | |
|
d7958024
Keenan Nemetz
7d ago
|
100 | // Network is the network this machine can give, derived from --container-network. |
| 101 | // A job that declares `network: bridge` is only leased to a runner that says it can | |
| 102 | // provide one, so the job waits for a capable machine instead of running without | |
| 103 | // the network it asked for. A runner that claims more than it can give only fails | |
| 104 | // its own jobs, which is the same trade as a label. | |
| 105 | Network string `json:"network,omitempty"` | |
|
34afc569
Keenan Nemetz
17d ago
|
106 | // Wait is how many seconds an idle lease may park before answering "nothing to |
| 107 | // do". Zero answers at once, which is what a one-shot runner wants: it is here to | |
| 108 | // run the job it was started for, not to wait for the next one. | |
| 109 | Wait int `json:"wait,omitempty"` | |
|
b312e166
Keenan Nemetz
11d ago
|
110 | // Record is the attempt's result, sent with finish. The runner no longer |
| 111 | // publishes it to Git: this channel is where the server receives it, checks it | |
| 112 | // against the lease, and stores it. | |
| 113 | Record *runs.Record `json:"record,omitempty"` | |
|
5ece70a5
Keenan Nemetz
10d ago
|
114 | // Log is one chunk of the attempt's log, sent while the job runs. |
| 115 | Log *runnerLogChunk `json:"log,omitempty"` | |
| 116 | } | |
| 117 | ||
| 118 | // runnerLogChunk is one piece of an attempt's log. Offset is where it starts in | |
| 119 | // the attempt's file, so the server can refuse a gap or a duplicate. | |
| 120 | type runnerLogChunk struct { | |
| 121 | Attempt int `json:"attempt"` | |
| 122 | Offset int64 `json:"offset"` | |
| 123 | Data []byte `json:"data"` | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
124 | } |
| 125 | ||
|
6e1fca51
Keenan Nemetz
6d ago
|
126 | // runnerCredential is the private half of a run's key. It is delivered once, written |
| 127 | // into the job's run directory, and never stored by the instance. | |
| 128 | type runnerCredential struct { | |
| 129 | PrivateKey string `json:"private_key"` | |
| 130 | Fingerprint string `json:"fingerprint"` | |
| 131 | } | |
| 132 | ||
|
4ac42680
Keenan Nemetz
7d ago
|
133 | // runnerSecret is one delivered value and the version it came from, so the attempt |
| 134 | // record can say which value ran without carrying it. | |
| 135 | type runnerSecret struct { | |
| 136 | Version int `json:"version"` | |
| 137 | Value string `json:"value"` | |
| 138 | } | |
| 139 | ||
| 140 | // runnerSecretIssue is a declared secret the repository could not supply. Reason is | |
| 141 | // empty when the reason is simply that nobody has stored that name. | |
| 142 | type runnerSecretIssue struct { | |
| 143 | Name string `json:"name"` | |
| 144 | Reason string `json:"reason,omitempty"` | |
| 145 | } | |
| 146 | ||
|
b44cbfaf
Keenan Nemetz
17d ago
|
147 | type runnerStep struct { |
| 148 | Name string `json:"name"` | |
| 149 | Run string `json:"run"` | |
| 150 | Timeout string `json:"timeout"` | |
| 151 | } | |
| 152 | ||
|
c1929d2f
Keenan Nemetz
5d ago
|
153 | type runnerService struct { |
| 154 | ID string `json:"id"` | |
| 155 | Image string `json:"image"` | |
| 156 | Env map[string]string `json:"env,omitempty"` | |
| 157 | } | |
| 158 | ||
|
b44cbfaf
Keenan Nemetz
17d ago
|
159 | type runnerJob struct { |
|
22528ac3
Keenan Nemetz
17d ago
|
160 | ID string `json:"id"` |
| 161 | Repo string `json:"repo"` | |
| 162 | SHA string `json:"sha"` | |
|
90fdaf39
Keenan Nemetz
17d ago
|
163 | // Attempt is which try at this job this is. The record lives under it, so the |
| 164 | // runner has to know the number it is reporting. | |
| 165 | Attempt int `json:"attempt"` | |
|
22528ac3
Keenan Nemetz
17d ago
|
166 | // Ref, WorkflowSHA and StepsSHA are what the runner signs when the attempt ends: |
| 167 | // the exact revision, the exact workflow file, and the digest of the steps that | |
|
90fdaf39
Keenan Nemetz
17d ago
|
168 | // were pinned when the job was queued. Without them the record could only say |
|
22528ac3
Keenan Nemetz
17d ago
|
169 | // "some job", which is not a statement anybody can check later. |
| 170 | Ref string `json:"ref"` | |
| 171 | WorkflowSHA string `json:"workflow_sha"` | |
| 172 | StepsSHA string `json:"steps_sha"` | |
| 173 | Branch string `json:"branch"` | |
| 174 | Workflow string `json:"workflow"` | |
| 175 | Job string `json:"job"` | |
| 176 | Steps []runnerStep `json:"steps"` | |
|
eb07ff56
Keenan Nemetz
8d ago
|
177 | // Trigger and EventPayload are the pinned event that requested this run. The |
| 178 | // runner writes the payload into a file and sets VOLE_EVENT, VOLE_EVENT_ACTION | |
| 179 | // and VOLE_EVENT_PATH; it is data, never an interpolated command. | |
| 180 | Trigger string `json:"trigger,omitempty"` | |
| 181 | TriggerAction string `json:"trigger_action,omitempty"` | |
| 182 | EventPayload string `json:"event_payload,omitempty"` | |
|
d7958024
Keenan Nemetz
7d ago
|
183 | // Network is the isolation the job asked for with `network:`, or empty when the |
| 184 | // workflow did not say. The runner applies it to the container it starts, so a job | |
| 185 | // that asks for none gets none even on a machine that could provide a bridge. | |
| 186 | Network string `json:"network,omitempty"` | |
|
6e1fca51
Keenan Nemetz
6d ago
|
187 | // Bot is the account this job acts as, and Credential is the key it may do it with. |
| 188 | // The private half is here and nowhere else: the queue holds the public half and | |
| 189 | // the account name, so a credential can be refused the moment the run ends. | |
| 190 | Bot string `json:"bot,omitempty"` | |
| 191 | Credential *runnerCredential `json:"credential,omitempty"` | |
| 192 | // CredentialError says the job asked to act as an account it cannot have. The | |
| 193 | // runner fails the job with this line rather than running it without the key, | |
| 194 | // because a step that cannot write is a broken job, not a degraded one. | |
| 195 | CredentialError string `json:"credential_error,omitempty"` | |
| 196 | // HostKey is where to find the instance when the job pushes, paired with the | |
| 197 | // credential above. | |
| 198 | HostKey string `json:"host_key,omitempty"` | |
|
e7aa0785
Keenan Nemetz
6d ago
|
199 | // Timeout is the job's budget as a duration string, or empty when it has none. |
| 200 | Timeout string `json:"timeout,omitempty"` | |
|
70b70264
Keenan Nemetz
15d ago
|
201 | // Image is a registry reference or "file:<name>". For the latter, ImageDefinition |
| 202 | // is the pinned recipe and ImageSHA is its digest; the runner builds it locally. | |
| 203 | Image string `json:"image"` | |
| 204 | ImageFile string `json:"image_file,omitempty"` | |
| 205 | ImageDefinition string `json:"image_definition,omitempty"` | |
| 206 | ImageSHA string `json:"image_sha,omitempty"` | |
|
3b39a84c
Keenan Nemetz
14d ago
|
207 | // Env is the job's declared environment, pinned when the run was queued. |
| 208 | Env map[string]string `json:"env,omitempty"` | |
|
c1929d2f
Keenan Nemetz
5d ago
|
209 | // Services are the containers the job asked to run beside it, pinned when the |
| 210 | // run was queued. The runner gives the job and these one network for the attempt. | |
| 211 | Services []runnerService `json:"services,omitempty"` | |
|
4ac42680
Keenan Nemetz
7d ago
|
212 | // Secrets are the values the job declared that this repository has, and Missing |
| 213 | // are the names it declared that it does not. A missing name is not a failure: | |
| 214 | // the variable is left unset and the runner says so at the top of the log. The | |
| 215 | // values travel here and nowhere else — the queue row holds the names only. | |
| 216 | Secrets map[string]runnerSecret `json:"secrets,omitempty"` | |
| 217 | Missing []runnerSecretIssue `json:"missing_secrets,omitempty"` | |
|
213f4d6d
Keenan Nemetz
17d ago
|
218 | // Artifacts are the declared patterns the runner expands and uploads after the |
| 219 | // steps succeed. | |
| 220 | Artifacts []string `json:"artifacts,omitempty"` | |
|
0f442b1c
Keenan Nemetz
17d ago
|
221 | // Tag is set for a tag-triggered run: the release being built. TagAnnotated says |
| 222 | // whether that tag carries notes, which is what makes it a release. | |
| 223 | Tag string `json:"tag,omitempty"` | |
| 224 | TagAnnotated bool `json:"tag_annotated,omitempty"` | |
| 225 | // ReleaseArtifacts are the patterns that also publish to the tag's release. | |
| 226 | ReleaseArtifacts []string `json:"release_artifacts,omitempty"` | |
|
88e980b0
Keenan Nemetz
17d ago
|
227 | // Pages is the site deploy this run performs, if any. The branch comes from |
| 228 | // instance configuration, which is why it is in the payload rather than in the | |
|
c5902451
Keenan Nemetz
16d ago
|
229 | // ci. |
|
88e980b0
Keenan Nemetz
17d ago
|
230 | Pages *runnerPages `json:"pages,omitempty"` |
| 231 | } | |
| 232 | ||
| 233 | type runnerPages struct { | |
| 234 | Path string `json:"path"` | |
| 235 | Ref string `json:"ref"` | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
236 | } |
| 237 | ||
| 238 | type runnerReply struct { | |
| 239 | Op string `json:"op"` | |
| 240 | Error string `json:"error,omitempty"` | |
| 241 | Run *runnerJob `json:"run,omitempty"` | |
| 242 | OK bool `json:"ok"` | |
| 243 | Lease int `json:"lease,omitempty"` | |
|
62576fa8
Keenan Nemetz
14d ago
|
244 | // Who am I: the account name this channel authenticated as. A runner answers |
| 245 | // "ping" with this so it can say whose results it will be reporting, and so a | |
| 246 | // --repos filter can be checked against what the account may run. | |
| 247 | Name string `json:"name,omitempty"` | |
|
b238ac24
Keenan Nemetz
17d ago
|
248 | // Repos are the repositories this account may run jobs for, sent with "ping" so |
| 249 | // a runner started with --repos can find out at startup that its filter matches | |
| 250 | // nothing — a typo it would otherwise discover as a permanently idle runner — | |
| 251 | // instead of after the first build that never arrived. | |
| 252 | Repos []string `json:"repos,omitempty"` | |
|
5ece70a5
Keenan Nemetz
10d ago
|
253 | // LogSize is how much of the log the server has after a log chunk. |
| 254 | LogSize int64 `json:"log_size,omitempty"` | |
|
0c14f078
Keenan Nemetz
6d ago
|
255 | // Cancel names the account that asked this running job to stop. It rides the |
| 256 | // heartbeat reply, so stopping a run needs no new traffic and a runner too old to | |
| 257 | // know the field ignores it. Empty means nobody has asked. | |
| 258 | Cancel string `json:"cancel,omitempty"` | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
259 | } |
| 260 | ||
| 261 | // serveRunner answers one runner session until it disconnects. | |
|
62576fa8
Keenan Nemetz
14d ago
|
262 | func (s *Server) serveRunner(ch xssh.Channel, user string) { |
|
b44cbfaf
Keenan Nemetz
17d ago
|
263 | defer ch.Close() |
| 264 | cfg := s.cfg | |
| 265 | if cfg.Runs == nil { | |
| 266 | _ = json.NewEncoder(ch).Encode(runnerReply{Op: "error", Error: "the run queue is not configured on this server"}) | |
| 267 | return | |
| 268 | } | |
|
78cd3437
Keenan Nemetz
13d ago
|
269 | dec := json.NewDecoder(&lineLimitReader{r: ch, limit: maxRunnerRequestBytes}) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
270 | enc := json.NewEncoder(ch) |
|
62576fa8
Keenan Nemetz
14d ago
|
271 | owner := user |
|
ac433531
Keenan Nemetz
17d ago
|
272 | // Presence is one row per channel: a pool of three runners on one account is |
| 273 | // three rows, and the row goes away when the channel does. Nothing else records | |
| 274 | // that a runner exists, so this is what the runners page shows. | |
| 275 | sessionID, err := id.New(time.Now().UTC()) | |
| 276 | if err != nil { | |
| 277 | _ = enc.Encode(runnerReply{Op: "error", Error: err.Error()}) | |
| 278 | return | |
| 279 | } | |
|
62576fa8
Keenan Nemetz
14d ago
|
280 | if err := cfg.Runs.TouchSession(sessionID, user, nil, time.Now().UTC()); err != nil { |
|
ac433531
Keenan Nemetz
17d ago
|
281 | cfg.Log.Warn("runner: cannot record the session", "user", user, "err", err) |
| 282 | } | |
| 283 | defer cfg.Runs.DropSession(sessionID) | |
|
34afc569
Keenan Nemetz
17d ago
|
284 | // repos is this session's eligibility list, reused for one poll window. Reading |
| 285 | // every repository's rules on every lease request is O(repos) per idle runner, and | |
| 286 | // the answer changes only when an admin edits an allowlist — so it is refreshed on | |
| 287 | // the same five-second beat the roster is, and not once per request. | |
| 288 | var repos []string | |
| 289 | var reposAt time.Time | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
290 | cfg.Log.Info("runner connected", "user", user) |
| 291 | ||
| 292 | for { | |
| 293 | var req runnerRequest | |
| 294 | if err := dec.Decode(&req); err != nil { | |
|
78cd3437
Keenan Nemetz
13d ago
|
295 | if errors.Is(err, errRequestTooLarge) { |
| 296 | _ = enc.Encode(runnerReply{Op: "error", Error: fmt.Sprintf( | |
| 297 | "a runner request must fit in %d bytes", maxRunnerRequestBytes)}) | |
| 298 | cfg.Log.Warn("runner: refused an oversized request", "user", user) | |
| 299 | } | |
| 300 | return // disconnected, or refused | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
301 | } |
| 302 | switch req.Op { | |
| 303 | case "lease": | |
|
34afc569
Keenan Nemetz
17d ago
|
304 | now := time.Now().UTC() |
| 305 | // A runner may only run jobs for repositories whose runner allowlist it is | |
| 306 | // on: running a job executes that repository's code, which is a bigger | |
| 307 | // power than writing to it. | |
| 308 | if reposAt.IsZero() || now.Sub(reposAt) > eligibleFor { | |
|
62576fa8
Keenan Nemetz
14d ago
|
309 | list, err := s.runnableRepos(user) |
|
34afc569
Keenan Nemetz
17d ago
|
310 | if err != nil { |
| 311 | _ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()}) | |
| 312 | continue | |
| 313 | } | |
| 314 | repos, reposAt = list, now | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
315 | } |
|
b238ac24
Keenan Nemetz
17d ago
|
316 | // --repos narrows the grant. The intersection is computed here rather than |
| 317 | // trusted from the runner, so the worst a hostile client can do is ask for | |
| 318 | // nothing — never for a repository no allowlist gave it. | |
| 319 | want := narrow(repos, req.Repos) | |
|
34afc569
Keenan Nemetz
17d ago
|
320 | // Take the wake-up before asking, so a run queued between the lease and the |
| 321 | // wait closes a channel somebody is holding. Asking first and waiting second | |
| 322 | // loses exactly that race. | |
| 323 | wake := cfg.Runs.Work() | |
|
6c03e461
Keenan Nemetz
16d ago
|
324 | lease := runs.LeaseRequest{ |
|
d7958024
Keenan Nemetz
7d ago
|
325 | Owner: owner, Repos: want, Labels: req.Labels, Network: req.Network, |
|
6c03e461
Keenan Nemetz
16d ago
|
326 | TTL: cfg.Lease, MaxPerRepo: cfg.MaxJobsPerRepo, |
| 327 | } | |
|
b238ac24
Keenan Nemetz
17d ago
|
328 | run, err := cfg.Runs.Lease(lease, now) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
329 | if err != nil { |
| 330 | _ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()}) | |
| 331 | continue | |
| 332 | } | |
|
34afc569
Keenan Nemetz
17d ago
|
333 | if run == nil && req.Wait > 0 { |
| 334 | wait := time.Duration(req.Wait) * time.Second | |
| 335 | if wait > maxLeaseWait { | |
| 336 | wait = maxLeaseWait | |
| 337 | } | |
| 338 | select { | |
| 339 | case <-wake: | |
| 340 | case <-time.After(wait): | |
|
786d7ce1
Keenan Nemetz
2d ago
|
341 | case <-s.done: |
|
34afc569
Keenan Nemetz
17d ago
|
342 | } |
| 343 | // Ask again rather than assume: the wake-up may belong to a job another | |
| 344 | // runner took first, and a lease is the only thing that hands work out. | |
|
b238ac24
Keenan Nemetz
17d ago
|
345 | run, err = cfg.Runs.Lease(lease, time.Now().UTC()) |
|
34afc569
Keenan Nemetz
17d ago
|
346 | if err != nil { |
| 347 | _ = enc.Encode(runnerReply{Op: "lease", Error: err.Error()}) | |
| 348 | continue | |
| 349 | } | |
| 350 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
351 | if run == nil { |
|
ac433531
Keenan Nemetz
17d ago
|
352 | // An idle runner is still connected: refresh its presence so the page can |
|
6c03e461
Keenan Nemetz
16d ago
|
353 | // say "waiting for work" instead of showing a runner that vanished. The |
| 354 | // labels ride along, because a runner waiting for the wrong kind of work | |
| 355 | // is exactly what this row is for. | |
|
62576fa8
Keenan Nemetz
14d ago
|
356 | _ = cfg.Runs.TouchSession(sessionID, user, req.Labels, time.Now().UTC()) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
357 | _ = enc.Encode(runnerReply{Op: "lease", OK: true}) |
| 358 | continue | |
| 359 | } | |
|
4ac42680
Keenan Nemetz
7d ago
|
360 | // The values are resolved here, at the moment the job is handed out, so |
| 361 | // the queue never holds one and a rotation between queueing and leasing | |
| 362 | // delivers the value the job actually ran with. | |
| 363 | secrets, missing := s.leaseSecrets(run, user) | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
364 | job := &runnerJob{ |
|
90fdaf39
Keenan Nemetz
17d ago
|
365 | ID: run.ID, Repo: run.Repo, SHA: run.SHA, Attempt: run.Attempts, Branch: run.Branch, |
|
22528ac3
Keenan Nemetz
17d ago
|
366 | Ref: run.Ref, Workflow: run.Workflow, WorkflowSHA: run.WorkflowSHA, |
| 367 | StepsSHA: run.StepsSHA, Job: run.Job, Artifacts: run.Artifacts, | |
|
eb07ff56
Keenan Nemetz
8d ago
|
368 | Trigger: run.Trigger, TriggerAction: run.TriggerAction, EventPayload: run.EventPayload, |
|
70b70264
Keenan Nemetz
15d ago
|
369 | Image: run.Image, ImageFile: run.ImageFile, ImageDefinition: run.ImageDefinition, |
| 370 | ImageSHA: run.ImageSHA, | |
|
d7958024
Keenan Nemetz
7d ago
|
371 | Network: run.Network, |
|
e7aa0785
Keenan Nemetz
6d ago
|
372 | Timeout: run.Timeout, |
|
3b39a84c
Keenan Nemetz
14d ago
|
373 | Env: run.Env, |
|
4ac42680
Keenan Nemetz
7d ago
|
374 | Secrets: secrets, Missing: missing, |
| 375 | Tag: run.Tag, TagAnnotated: run.TagAnnotated, ReleaseArtifacts: run.ReleaseArtifacts, | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
376 | } |
|
88e980b0
Keenan Nemetz
17d ago
|
377 | if run.PagesPath != "" { |
| 378 | ref := cfg.PagesRef | |
| 379 | if ref == "" { | |
| 380 | ref = "pages" | |
| 381 | } | |
| 382 | job.Pages = &runnerPages{Path: run.PagesPath, Ref: ref} | |
| 383 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
384 | for _, st := range run.Steps { |
| 385 | job.Steps = append(job.Steps, runnerStep{Name: st.Name, Run: st.Run, Timeout: st.Timeout}) | |
| 386 | } | |
|
c1929d2f
Keenan Nemetz
5d ago
|
387 | for _, svc := range run.Services { |
| 388 | job.Services = append(job.Services, runnerService{ID: svc.ID, Image: svc.Image, Env: svc.Env}) | |
| 389 | } | |
|
6e1fca51
Keenan Nemetz
6d ago
|
390 | // A job that acts as a bot gets a key for this run only. The account and |
| 391 | // the repository are pinned in the row, so the hook can refuse a write that | |
| 392 | // the account's role would not allow and the key stops working when the run | |
| 393 | // ends. A bot that cannot be used fails the job here rather than at its | |
| 394 | // first push, which is a much better place to learn it. | |
| 395 | if run.Bot != "" { | |
| 396 | cred, hostKey, reason := s.mintCredential(run, now) | |
| 397 | if reason != "" { | |
| 398 | job.CredentialError = reason | |
| 399 | } else { | |
| 400 | job.Bot, job.Credential, job.HostKey = run.Bot, cred, hostKey | |
| 401 | } | |
| 402 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
403 | cfg.Log.Info("runner leased a job", "user", user, "repo", run.Repo, |
| 404 | "job", run.Job, "sha", run.SHA[:8], "run", run.ID, "attempt", run.Attempts) | |
|
ac433531
Keenan Nemetz
17d ago
|
405 | _ = cfg.Runs.SessionWorking(sessionID, run.Repo, run.ID, time.Now().UTC()) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
406 | _ = enc.Encode(runnerReply{Op: "lease", OK: true, Run: job, Lease: int(cfg.Lease.Seconds())}) |
| 407 | ||
|
5ece70a5
Keenan Nemetz
10d ago
|
408 | case "log": |
| 409 | if req.Log == nil { | |
| 410 | _ = enc.Encode(runnerReply{Op: "log", Error: "a log request carries no chunk"}) | |
| 411 | continue | |
| 412 | } | |
| 413 | run, err := cfg.Runs.Get(req.Run) | |
| 414 | if err != nil { | |
| 415 | _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()}) | |
| 416 | continue | |
| 417 | } | |
| 418 | if run == nil { | |
| 419 | _ = enc.Encode(runnerReply{Op: "log", Error: fmt.Sprintf("no run %s", req.Run)}) | |
| 420 | continue | |
| 421 | } | |
| 422 | if req.Log.Attempt != run.Attempts { | |
| 423 | _ = enc.Encode(runnerReply{Op: "log", Error: fmt.Sprintf( | |
| 424 | "%s: the log is attempt %d but this runner holds attempt %d", | |
| 425 | req.Run, req.Log.Attempt, run.Attempts)}) | |
| 426 | continue | |
| 427 | } | |
| 428 | held, err := cfg.Runs.Held(run.ID, owner) | |
| 429 | if err != nil { | |
| 430 | _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()}) | |
| 431 | continue | |
| 432 | } | |
| 433 | if !held { | |
| 434 | _ = enc.Encode(runnerReply{Op: "log", OK: false}) | |
| 435 | continue | |
| 436 | } | |
| 437 | if cfg.Logs == nil { | |
| 438 | _ = enc.Encode(runnerReply{Op: "log", Error: "this instance has no log directory"}) | |
| 439 | continue | |
| 440 | } | |
| 441 | size, err := cfg.Logs.Append(run.Repo, run.ID, req.Log.Attempt, req.Log.Offset, req.Log.Data) | |
| 442 | if err != nil { | |
| 443 | _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()}) | |
| 444 | continue | |
| 445 | } | |
| 446 | if ok, err := cfg.Runs.RecordLogSize(run.ID, owner, req.Log.Attempt, size, time.Now().UTC()); err != nil { | |
| 447 | _ = enc.Encode(runnerReply{Op: "log", Error: err.Error()}) | |
| 448 | continue | |
| 449 | } else if !ok { | |
| 450 | _ = enc.Encode(runnerReply{Op: "log", OK: false}) | |
| 451 | continue | |
| 452 | } | |
| 453 | _ = enc.Encode(runnerReply{Op: "log", OK: true, LogSize: size}) | |
| 454 | ||
|
b44cbfaf
Keenan Nemetz
17d ago
|
455 | case "heartbeat": |
|
ac433531
Keenan Nemetz
17d ago
|
456 | // Liveness only: the lease is what a heartbeat buys. Presence is refreshed |
|
0c14f078
Keenan Nemetz
6d ago
|
457 | // here too, so a long build does not make its runner look gone. A stop request |
| 458 | // rides the same reply rather than a new message: the runner already asks this | |
| 459 | // question every few seconds, and a runner that does not know the field keeps | |
| 460 | // working exactly as before. | |
|
62576fa8
Keenan Nemetz
14d ago
|
461 | _ = cfg.Runs.TouchSession(sessionID, user, nil, time.Now().UTC()) |
|
0c14f078
Keenan Nemetz
6d ago
|
462 | held, cancelBy, err := cfg.Runs.Heartbeat(req.Run, owner, cfg.Lease, time.Now().UTC()) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
463 | if err != nil { |
| 464 | _ = enc.Encode(runnerReply{Op: "heartbeat", Error: err.Error()}) | |
| 465 | continue | |
| 466 | } | |
|
0c14f078
Keenan Nemetz
6d ago
|
467 | _ = enc.Encode(runnerReply{Op: "heartbeat", OK: held, Cancel: cancelBy}) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
468 | |
| 469 | case "finish": | |
|
b312e166
Keenan Nemetz
11d ago
|
470 | // The result arrives here, not on the metadata branch. The record is |
| 471 | // checked against the lease and the job it claims to describe, then | |
| 472 | // stored with the run's status in one transaction. | |
|
ac433531
Keenan Nemetz
17d ago
|
473 | _ = cfg.Runs.SessionIdle(sessionID, time.Now().UTC()) |
|
90fdaf39
Keenan Nemetz
17d ago
|
474 | run, err := cfg.Runs.Get(req.Run) |
|
22528ac3
Keenan Nemetz
17d ago
|
475 | if err != nil { |
| 476 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 477 | continue | |
| 478 | } | |
|
90fdaf39
Keenan Nemetz
17d ago
|
479 | if run == nil { |
| 480 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf("no run %s", req.Run)}) | |
| 481 | continue | |
|
22528ac3
Keenan Nemetz
17d ago
|
482 | } |
|
b312e166
Keenan Nemetz
11d ago
|
483 | if req.Record == nil { |
|
90fdaf39
Keenan Nemetz
17d ago
|
484 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( |
|
b312e166
Keenan Nemetz
11d ago
|
485 | "%s: this runner did not send a record; upgrade the runner and the server together", req.Run)}) |
|
90fdaf39
Keenan Nemetz
17d ago
|
486 | continue |
|
b312e166
Keenan Nemetz
11d ago
|
487 | } |
| 488 | if req.Record.Attempt != run.Attempts { | |
|
90fdaf39
Keenan Nemetz
17d ago
|
489 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( |
|
b312e166
Keenan Nemetz
11d ago
|
490 | "%s: the record is attempt %d but this runner holds attempt %d", |
| 491 | req.Run, req.Record.Attempt, run.Attempts)}) | |
|
90fdaf39
Keenan Nemetz
17d ago
|
492 | continue |
| 493 | } | |
|
b312e166
Keenan Nemetz
11d ago
|
494 | if req.Record.Signer != owner { |
| 495 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( | |
| 496 | "%s: the record names %s as the reporter, but %s reported it", | |
| 497 | req.Run, req.Record.Signer, owner)}) | |
|
379dad1e
Keenan Nemetz
17d ago
|
498 | continue |
| 499 | } | |
|
b312e166
Keenan Nemetz
11d ago
|
500 | if err := checkRecordPointers(cfg.LFS, run.Repo, *req.Record); err != nil { |
|
86fe5bb4
Keenan Nemetz
17d ago
|
501 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) |
| 502 | continue | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
503 | } |
|
5ece70a5
Keenan Nemetz
10d ago
|
504 | if req.Record.LogSize == 0 { |
| 505 | // A failed stream leaves a partial file, and an attempt that produced | |
| 506 | // no output leaves none. Either way the result names no log. | |
| 507 | if cfg.Logs != nil { | |
| 508 | if err := cfg.Logs.Remove(run.Repo, run.ID, req.Record.Attempt); err != nil { | |
| 509 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 510 | continue | |
| 511 | } | |
| 512 | } | |
| 513 | } else { | |
| 514 | if cfg.Logs == nil { | |
| 515 | _ = enc.Encode(runnerReply{Op: "finish", Error: "this instance has no log directory"}) | |
| 516 | continue | |
| 517 | } | |
| 518 | size, err := cfg.Logs.Size(run.Repo, run.ID, req.Record.Attempt) | |
| 519 | if err != nil { | |
| 520 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 521 | continue | |
| 522 | } | |
| 523 | if size != req.Record.LogSize { | |
| 524 | _ = enc.Encode(runnerReply{Op: "finish", Error: fmt.Sprintf( | |
| 525 | "%s: the log is %d bytes but the record says %d", | |
| 526 | req.Run, size, req.Record.LogSize)}) | |
| 527 | continue | |
| 528 | } | |
| 529 | if err := cfg.Logs.Sync(run.Repo, run.ID, req.Record.Attempt); err != nil { | |
| 530 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 531 | continue | |
| 532 | } | |
| 533 | } | |
|
b312e166
Keenan Nemetz
11d ago
|
534 | ok, err := cfg.Runs.RecordAttempt(run.ID, owner, *req.Record, time.Now().UTC()) |
|
22528ac3
Keenan Nemetz
17d ago
|
535 | if err != nil { |
| 536 | _ = enc.Encode(runnerReply{Op: "finish", Error: err.Error()}) | |
| 537 | continue | |
| 538 | } | |
|
6e1fca51
Keenan Nemetz
6d ago
|
539 | // The run is over, so the key it could act with is over too. The expiry would |
| 540 | // catch it eventually; this is the part that does not wait. | |
| 541 | if err := cfg.Runs.DropRunCredentials(run.ID); err != nil { | |
| 542 | cfg.Log.Warn("runner: cannot drop the run credential", "run", run.ID, "err", err) | |
| 543 | } | |
|
b312e166
Keenan Nemetz
11d ago
|
544 | cfg.Log.Info("run finished", "user", user, "run", run.ID, "exit", req.Record.Exit, |
| 545 | "signer", req.Record.Signer, "recorded", ok) | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
546 | _ = enc.Encode(runnerReply{Op: "finish", OK: ok}) |
| 547 | ||
| 548 | case "ping": | |
|
62576fa8
Keenan Nemetz
14d ago
|
549 | reply := runnerReply{Op: "ping", OK: true, Name: user} |
|
b238ac24
Keenan Nemetz
17d ago
|
550 | // What this account may run, so a runner started with --repos can say at |
| 551 | // startup that its filter matches nothing rather than idling in silence. | |
|
62576fa8
Keenan Nemetz
14d ago
|
552 | if granted, gerr := s.runnableRepos(user); gerr == nil { |
|
b238ac24
Keenan Nemetz
17d ago
|
553 | reply.Repos = granted |
| 554 | } else { | |
| 555 | cfg.Log.Warn("runner: cannot list runnable repositories", "user", user, "err", gerr) | |
| 556 | } | |
|
22528ac3
Keenan Nemetz
17d ago
|
557 | _ = enc.Encode(reply) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
558 | |
| 559 | default: | |
| 560 | _ = enc.Encode(runnerReply{Op: req.Op, Error: "unknown op"}) | |
| 561 | } | |
| 562 | } | |
| 563 | } | |
| 564 | ||
|
b312e166
Keenan Nemetz
11d ago
|
565 | // checkRecordPointers refuses a record whose log or artifacts are not actually in |
| 566 | // the object store. A push used to be checked for pointer hygiene at the boundary; | |
| 567 | // the finish channel is the boundary now. | |
| 568 | func checkRecordPointers(objects *lfs.Store, repoName string, rec runs.Record) error { | |
| 569 | for _, a := range rec.Artifacts { | |
| 570 | if err := checkPointer(objects, repoName, lfs.Pointer{OID: a.OID, Size: a.Size}); err != nil { | |
| 571 | return fmt.Errorf("artifact %s: %w", a.Name, err) | |
| 572 | } | |
| 573 | } | |
| 574 | return nil | |
| 575 | } | |
| 576 | ||
| 577 | func checkPointer(objects *lfs.Store, repoName string, p lfs.Pointer) error { | |
| 578 | if objects == nil { | |
| 579 | return fmt.Errorf("this instance has no object store for run outputs") | |
| 580 | } | |
| 581 | size, err := objects.Stat(repoName, p.OID) | |
| 582 | if err != nil { | |
| 583 | return fmt.Errorf("object %s is missing: %w", p.OID, err) | |
| 584 | } | |
| 585 | if size != p.Size { | |
| 586 | return fmt.Errorf("object %s is %d bytes, pointer says %d", p.OID, size, p.Size) | |
| 587 | } | |
| 588 | return nil | |
| 589 | } | |
| 590 | ||
|
b44cbfaf
Keenan Nemetz
17d ago
|
591 | // runnableRepos lists the repositories this account may execute jobs for. |
|
62576fa8
Keenan Nemetz
14d ago
|
592 | func (s *Server) runnableRepos(name string) ([]string, error) { |
|
b44cbfaf
Keenan Nemetz
17d ago
|
593 | repos, err := s.cfg.Repos.List() |
| 594 | if err != nil { | |
| 595 | return nil, err | |
| 596 | } | |
| 597 | var out []string | |
| 598 | for _, m := range repos { | |
| 599 | a, err := m.ACL() | |
| 600 | if err != nil { | |
| 601 | continue // a repository without readable rules grants nothing | |
| 602 | } | |
|
852c067d
Keenan Nemetz
17d ago
|
603 | // Whose results count is the same list as whose job may be leased: running a |
| 604 | // job executes that repository's code, and a result from an account the | |
| 605 | // repository will not count is only worth refusing here rather than after | |
| 606 | // the build. | |
|
62576fa8
Keenan Nemetz
14d ago
|
607 | if a.CanReport(name) { |
|
b44cbfaf
Keenan Nemetz
17d ago
|
608 | out = append(out, m.Name) |
| 609 | } | |
| 610 | } | |
| 611 | return out, nil | |
| 612 | } | |
|
b238ac24
Keenan Nemetz
17d ago
|
613 | |
| 614 | // narrow keeps only the repositories a runner asked for, and is the whole of what | |
| 615 | // --repos is allowed to do. The granted list is what the account's runner allowlists | |
| 616 | // produce; the asked list comes from the client. Intersecting them means the worst a | |
| 617 | // client can do is run less than it was allowed to — the alternative, honouring the | |
| 618 | // client's list, would make an SSH flag an access-control decision. | |
| 619 | // | |
| 620 | // An empty asked list means "no narrowing" rather than "nothing": that is what a runner | |
| 621 | // without --repos sends, and it keeps such a runner following allowlist edits instead | |
| 622 | // of freezing its list at connect time. Asked names that were never granted simply | |
| 623 | // disappear, which the runner reports on startup because it can see both lists. | |
| 624 | func narrow(granted, asked []string) []string { | |
| 625 | if len(asked) == 0 { | |
| 626 | return granted | |
| 627 | } | |
| 628 | want := make(map[string]bool, len(asked)) | |
| 629 | for _, r := range asked { | |
| 630 | want[r] = true | |
| 631 | } | |
| 632 | out := make([]string, 0, len(granted)) | |
| 633 | for _, r := range granted { | |
| 634 | if want[r] { | |
| 635 | out = append(out, r) | |
| 636 | } | |
| 637 | } | |
| 638 | return out | |
| 639 | } |