Compare commits

...

10 Commits

Author SHA1 Message Date
bircni a0c4de79f7 feat: add log.job.dir (#1165)
`log.job.dir` makes the runner write a copy of every task's log to that directory, as `<start time>-task-<id>.log`: the rows exactly as Gitea received them, with the same masking and the job's result on the last line. Off by default, and what Gitea shows does not change.

`log.job.retention` (default `168h`) and `log.job.max_size` (default `1GB`) bound the directory. Documented in the README.

---------

Co-authored-by: silverwind <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1165
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
2026-08-21 07:25:11 +00:00
silverwind dfe979e1d0 fix: stop a failed step disabling the toolkit patch (#1177)
A step that failed, for any reason, made the runner restore the action's stock bundle and mark it never to be patched again. Every later `actions/upload-artifact` run then failed with `GHESNotSupportedError`, and nothing in the log said why.

The edit is now made as the action is copied into the job container, under the lock that guards the copy, and nothing reverts it. That also closes the race where another job's checkout reset the bundle mid-job.

`cache.v2` no longer decides whether the edit is made, it only withdraws the v2 advertisement, so artifacts work whatever the cache is set to.

Also added a new `runner.patch_actions` option to turn the edit off if it ever breaks an action.

Fixes https://gitea.com/gitea/runner/issues/1176

Reviewed-on: https://gitea.com/gitea/runner/pulls/1177
Reviewed-by: bircni <bircni@icloud.com>
Co-authored-by: silverwind <me@silverwind.io>
2026-08-20 15:46:33 +00:00
bircni 11ac12efa4 enhance: add runner.default_image for jobs matching no label (#1164)
A job whose `runs-on` matches none of the runner's labels, which includes every job that sets no `runs-on` at all, runs in `runner.default_image`. It defaults to `docker.gitea.com/runner-images:ubuntu-latest` as before, so a mirror can be pointed at instead.

A runner with no reachable docker daemon now runs such a job on the host, rather than failing on an image it cannot pull. Runners that use docker are unaffected and never probe for one.

This matters most to a host-mode runner, one whose labels are all `host`. Such a runner has no daemon to pull an image with, so a job matching none of its labels used to fail at container start. It now runs on the host, where that runner runs everything else anyway, and it takes no configuration to get there. A host-mode runner that does have a daemon within reach keeps using the image, unchanged.

Supersedes https://gitea.com/gitea/runner/pulls/642

---------

Co-authored-by: silverwind <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1164
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
2026-08-19 18:36:00 +00:00
silverwind bc8161c673 fix: bound blocking calls and stop failing silently (#1174)
Jobs occasionally go silent ([example](https://gitea.com/gitea/runner/actions/runs/805045/jobs/1055123)) mid-run and Gitea reaped them after `ZOMBIE_TASK_TIMEOUT`, with no error in the log. This contains a number of related fixes, all with full test coverage:

1. Bound every RPC to Gitea with a timeout, a stalled report otherwise parked logs and heartbeats for the whole job.
2. Cap `runner.fetch_timeout` at that ceiling.
3. Let only the daemon loop close its own channel, the race panicked the process.
4. Stop the job on any terminal server result, not just `RESULT_CANCELLED`.
5. Report that result instead of relabelling it as cancelled.
6. Log reporting failures once at each end of an outage instead of discarding them.
7. Clamp the acknowledged log index, a too-large ack panicked on a slice bound.
8. Stop reading server health from a `FetchTask` deadline, it marked the runner healthy and reset the error backoff on a timeout.
9. Return an error from the Docker version probe instead of a `logrus` panic.
10. Pass the context to go-git's fetch and pull.
11. Fail the clone when a refresh dies on a cancelled context.
12. Set `terminationGracePeriodSeconds` in the Kubernetes examples.

Also contains a deprecation fix for goreleaser.

Reviewed-on: https://gitea.com/gitea/runner/pulls/1174
Reviewed-by: bircni <bircni@icloud.com>
Co-authored-by: silverwind <me@silverwind.io>
2026-08-19 11:13:21 +00:00
silverwind 3f70822458 enhance: add runner.tool_cache_mode and default it to none (#1171)
The current shared tools cache is not concurrency-safe, e.g. multiple jobs can write and corrupt it, for example `setup-go` with explicit go version under concurrency reliably corrupts the tool cache and fails all jobs.

This adds a new `runner.tool_cache_mode` (and `--tool-cache-mode` exec option) option which defaults to unshared tools cache:

- `none` mounts nothing, so a job uses what its image ships there and discards what it installs
- `shared` keeps the single volume every job reuses, and warns when `runner.capacity` is above 1

Under `none` effective tool cache can only come from the image or host, which is the same as it is on GitHub Actions which ships many preinstalled tools in its fat VM images.

Co-authored-by: bircni <bircni@icloud.com>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1171
Reviewed-by: bircni <bircni@icloud.com>
Co-authored-by: silverwind <me@silverwind.io>
2026-08-18 20:16:11 +00:00
bircni be90c01468 feat: add size-based cache eviction (#1170)
The cache server retired entries 30 days after creation regardless of use, so a job that ran often enough to keep its cache warm still lost it on a fixed schedule. Nothing bounded the disk either.

Retention now counts from last access alone, and a repository over its limit sheds least recently accessed entries until it fits, enforced on commit as well as on the periodic sweep.

```yaml
cache:
  retention: 168h        # remove entries not accessed for seven days
  repo_size_limit: 10GB  # cap each repository
  size_limit: 0          # cap the whole cache, off by default
  sweep_interval: 1h     # minimum time between sweeps
```

Sizes accept `10GB`, `512mb`, `1TiB` or a plain byte count, binary either way. Leave a key out for its default; `0` turns a limit off, and `0s` does the same for `retention`. Whatever these allow, the cache also sheds entries to keep free space above `health_check.min_free_disk_space_mb` when health checks are enabled, so it cannot grow past the point where the runner stops accepting work.

Supporting fixes: serving an entry stamps its access time, so a find cannot hand a job a download URL for an entry the next eviction is about to remove; an entry larger than the limit is dropped on its own account rather than emptying its repository to make room; and a blob that cannot be unlinked keeps its row, so the next sweep retries instead of orphaning bytes no limit can account for.

Closes https://gitea.com/gitea/runner/issues/1168

---------

Co-authored-by: silverwind <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1170
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
2026-08-18 15:32:53 +00:00
bircni 6c6a878403 enhance: show workflow_dispatch inputs in the "Set up job" section (#1167)
Trigger-time inputs used to be visible in the job log but are no longer shown after the "Set up job" section was reshaped. This restores them as an "Inputs" group listing each provided input and its value, rendered only when the event payload carries inputs.

For a run dispatched with `required=required input`, `with_default=default` and `boolean=true`, the "Set up job" log now shows:

​```
gitea-com-gitea-0003(version:v3.0.2)
▸ Runner Information
    Task: 268506
    Job: test
    Repository: gitea/runner
    Triggered by event: workflow_dispatch
▸ Inputs
    boolean: true
    required: required input
    with_default: default
▸ Operating System
    Ubuntu 24.04.4 LTS
    linux/amd64
​```

Closes https://gitea.com/gitea/runner/issues/1166

Reviewed-on: https://gitea.com/gitea/runner/pulls/1167
Reviewed-by: Lunny Xiao <xiaolunwen@gmail.com>
2026-08-15 19:57:30 +00:00
Lunny Xiao f97680a68d chore: drop AWS S3 release upload, keep Cloudflare R2 only (#1169)
Release artifacts were uploaded to both AWS S3 and Cloudflare R2 during the migration period. Drop the goreleaser `blobs:` S3 pipe and the AWS_*/S3_* secrets from the release workflows, so artifacts are published to Cloudflare R2 only.

Assisted-by: Codet:GPT-5.1-Codex
Reviewed-on: https://gitea.com/gitea/runner/pulls/1169
Reviewed-by: bircni <bircni@icloud.com>
Co-authored-by: Lunny Xiao <xiaolunwen@gmail.com>
2026-08-15 19:41:04 +00:00
Renovate Bot dbd9a892f8 chore(deps): update dependencies (#1160)
This PR contains the following updates:

| Package | Type | Update | Change | Pending | [Age](https://docs.renovatebot.com/merge-confidence/) | [Confidence](https://docs.renovatebot.com/merge-confidence/) |
|---|---|---|---|---|---|---|
| docker | stage | minor | `29.6.2-dind-rootless` → `29.7.1-dind-rootless` |  | ![age](https://developer.mend.io/api/mc/badges/age/docker/docker/29.7.1?slim=true) | ![confidence](https://developer.mend.io/api/mc/badges/confidence/docker/docker/29.6.2/29.7.1?slim=true) |
| docker | stage | minor | `29.6.2-dind` → `29.7.1-dind` |  | ![age](https://developer.mend.io/api/mc/badges/age/docker/docker/29.7.1?slim=true) | ![confidence](https://developer.mend.io/api/mc/badges/confidence/docker/docker/29.6.2/29.7.1?slim=true) |
| [github.com/docker/cli](https://github.com/docker/cli) | require | minor | `v29.6.2+incompatible` → `v29.7.1+incompatible` | `v29.7.2+incompatible` | ![age](https://developer.mend.io/api/mc/badges/age/go/github.com%2fdocker%2fcli/v29.7.1+incompatible?slim=true) | ![confidence](https://developer.mend.io/api/mc/badges/confidence/go/github.com%2fdocker%2fcli/v29.6.2+incompatible/v29.7.1+incompatible?slim=true) |
| [github.com/moby/go-archive](https://github.com/moby/go-archive) | require | minor | `v0.2.1` → `v0.3.2` | `v0.3.3` | ![age](https://developer.mend.io/api/mc/badges/age/go/github.com%2fmoby%2fgo-archive/v0.3.2?slim=true) | ![confidence](https://developer.mend.io/api/mc/badges/confidence/go/github.com%2fmoby%2fgo-archive/v0.2.1/v0.3.2?slim=true) |

---

### Release Notes

<details>
<summary>docker/cli (github.com/docker/cli)</summary>

### [`v29.7.1+incompatible`](https://github.com/docker/cli/compare/v29.7.0...v29.7.1)

[Compare Source](https://github.com/docker/cli/compare/v29.7.0...v29.7.1)

### [`v29.7.0+incompatible`](https://github.com/docker/cli/compare/v29.6.2...v29.7.0)

[Compare Source](https://github.com/docker/cli/compare/v29.6.2...v29.7.0)

</details>

<details>
<summary>moby/go-archive (github.com/moby/go-archive)</summary>

### [`v0.3.2`](https://github.com/moby/go-archive/releases/tag/v0.3.2)

[Compare Source](https://github.com/moby/go-archive/compare/v0.3.1...v0.3.2)

#### What's Changed

Fix a regression introduced in v0.3.0 that caused archive extraction to fail when paths traversed absolute symlinks inside the destination root, such as `var/run -> /run`. Absolute symlink targets are now resolved relative to the extraction root while relative symlink escapes remain rejected. [#&#8203;93](https://github.com/moby/go-archive/pull/93)

**Full Changelog**: <https://github.com/moby/go-archive/compare/v0.3.1...v0.3.2>

### [`v0.3.1`](https://github.com/moby/go-archive/releases/tag/v0.3.1)

[Compare Source](https://github.com/moby/go-archive/compare/v0.3.0...v0.3.1)

#### Fixes

This patch release fixes a regression introduced in v0.2.1 where archive extraction could fail when an archive omitted explicit entries for parent directories. For example, extracting `etc/dnf/` without a preceding `etc/` entry could return `mkdirat etc/dnf: no such file or directory`.

This prevented affected images from being extracted. Archive extraction now creates implied parent directories for both file and directory entries.

#### What's Changed

- archive: create implied parents for directory entries [#&#8203;92](https://github.com/moby/go-archive/pull/92)
- archive: Tarballer.Go: suppress io.ErrClosedPipe logs on close [#&#8203;94](https://github.com/moby/go-archive/pull/94)

**Full Changelog**: <https://github.com/moby/go-archive/compare/v0.3.0...v0.3.1>

### [`v0.3.0`](https://github.com/moby/go-archive/releases/tag/v0.3.0)

[Compare Source](https://github.com/moby/go-archive/compare/v0.2.1...v0.3.0)

#### Security

This release fixes **CVE-2026-17106** / **[GHSA-hfg8-hc9c-6c3h](https://github.com/moby/go-archive/security/advisories/GHSA-hfg8-hc9c-6c3h)**, where a crafted tar archive could use links to cause extraction operations to create or overwrite files outside the intended destination directory.

The issue affected `Unpack`, `UnpackLayer`, `Untar`, `UntarUncompressed`, and the `ApplyLayer` helpers. Users should upgrade and avoid extracting untrusted archives with earlier versions.

#### What's Changed

- archive: harden tar extraction against path traversal [#&#8203;45](https://github.com/moby/go-archive/pull/45)
- archive: do not follow reparse points in chtimes [#&#8203;90](https://github.com/moby/go-archive/pull/90)
- archive: fix creation time updates on Windows [#&#8203;79](https://github.com/moby/go-archive/pull/79)
- archive: minor cleanups and godoc touch-up [#&#8203;87](https://github.com/moby/go-archive/pull/87)
- archive: RebaseArchiveEntries: fix archive path rebasing [#&#8203;43](https://github.com/moby/go-archive/pull/43)

#### Test and CI changes

- ci: enable dependabot for actions [#&#8203;81](https://github.com/moby/go-archive/pull/81)
- archive: make breakoutErr unwrap its cause [#&#8203;91](https://github.com/moby/go-archive/pull/91)
- archive: use filepath for filesystem paths in tests [#&#8203;80](https://github.com/moby/go-archive/pull/80)
- archive: use filepath for filesystem paths in tests [#&#8203;80](https://github.com/moby/go-archive/pull/80)

**Full Changelog**: <https://github.com/moby/go-archive/compare/v0.2.1...v0.3.0>

</details>

---

### Configuration

📅 **Schedule**: (UTC)

- Branch creation
  - Between 12:00 AM and 03:59 AM, only on Monday (`* 0-3 * * 1`)
- Automerge
  - At any time (no schedule defined)

🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied.

♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox.

👻 **Immortal**: This PR will be recreated if closed unmerged. Get [config help](https://github.com/renovatebot/renovate/discussions) if that's undesired.

---

 - [ ] <!-- rebase-check -->If you want to rebase/retry this PR, check this box

---

This PR has been generated by [Mend Renovate](https://github.com/renovatebot/renovate).
<!--renovate-debug:eyJjcmVhdGVkSW5WZXIiOiI0My4xOTEuMiIsInVwZGF0ZWRJblZlciI6IjQzLjE5MS4yIiwidGFyZ2V0QnJhbmNoIjoibWFpbiIsImxhYmVscyI6W119-->

Reviewed-on: https://gitea.com/gitea/runner/pulls/1160
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
Co-authored-by: Renovate Bot <renovate-bot@gitea.com>
2026-08-10 20:46:33 +00:00
Max P. e178c03adc fix(cache): build job URLs on the address its runner registered (#1153)
The cache server built every URL it hands a job from its own listen address, so jobs whose runner reaches it through a reverse proxy were sent to the internal one. This covered the v1 `archiveLocation`, the v2 signed cache URLs and `ACTIONS_RESULTS_URL`.

Runners now register the address their jobs reach the server at, next to the instance URL they already send. The cache-server needs no configuration of its own, and runners that reach it differently each get their own correct address.

Closes https://gitea.com/gitea/runner/issues/1152

---------

Co-authored-by: silverwind <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1153
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
Reviewed-by: bircni <bircni@icloud.com>
Co-authored-by: Max P. <mail@0xMax42.io>
2026-08-08 07:05:38 +00:00
59 changed files with 1962 additions and 673 deletions
+3 -9
View File
@@ -20,11 +20,10 @@ jobs:
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with: with:
fetch-depth: 0 fetch-depth: 0
# Custom publishers (the R2 mirror below) run as the very last # Custom publishers (the R2 upload below) run as the very last
# step of goreleaser's publish pipeline, after the Gitea release # step of goreleaser's publish pipeline, after the Gitea release
# has already been created and every artifact already uploaded # has already been created. Fail here instead, before anything
# to S3. Fail here instead, before anything is built or # is built or published, if the R2 secrets are missing.
# published, if the R2 secrets are missing.
- name: check R2 configuration - name: check R2 configuration
run: sh scripts/upload-r2.sh --check-config run: sh scripts/upload-r2.sh --check-config
env: env:
@@ -43,11 +42,6 @@ jobs:
args: release --nightly args: release --nightly
env: env:
GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }} GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
AWS_REGION: ${{ secrets.AWS_REGION }}
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
S3_REGION: ${{ secrets.AWS_REGION }}
S3_BUCKET: ${{ secrets.AWS_BUCKET }}
R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }} R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
R2_BUCKET: ${{ secrets.R2_BUCKET }} R2_BUCKET: ${{ secrets.R2_BUCKET }}
R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }} R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+3 -9
View File
@@ -12,11 +12,10 @@ jobs:
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7 - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
with: with:
fetch-depth: 0 # all history for all branches and tags fetch-depth: 0 # all history for all branches and tags
# Custom publishers (the R2 mirror below) run as the very last # Custom publishers (the R2 upload below) run as the very last
# step of goreleaser's publish pipeline, after the Gitea release # step of goreleaser's publish pipeline, after the Gitea release
# has already been created and every artifact already uploaded # has already been created. Fail here instead, before anything
# to S3. Fail here instead, before anything is built or # is built or published, if the R2 secrets are missing.
# published, if the R2 secrets are missing.
- name: check R2 configuration - name: check R2 configuration
run: sh scripts/upload-r2.sh --check-config run: sh scripts/upload-r2.sh --check-config
env: env:
@@ -42,11 +41,6 @@ jobs:
args: release args: release
env: env:
GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }} GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
AWS_REGION: ${{ secrets.AWS_REGION }}
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_KEY_ID }}
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
S3_REGION: ${{ secrets.AWS_REGION }}
S3_BUCKET: ${{ secrets.AWS_BUCKET }}
R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }} R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
R2_BUCKET: ${{ secrets.R2_BUCKET }} R2_BUCKET: ${{ secrets.R2_BUCKET }}
R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }} R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+7 -19
View File
@@ -83,24 +83,12 @@ builds:
- cmd: sh .goreleaser.checksum.sh {{ .Path }} - cmd: sh .goreleaser.checksum.sh {{ .Path }}
- cmd: sh .goreleaser.checksum.sh {{ .Path }}.xz - cmd: sh .goreleaser.checksum.sh {{ .Path }}.xz
blobs: # Uploads every release artifact to Cloudflare R2. The `blobs:` pipe
- # isn't usable here since it authenticates from the global AWS_* env
provider: s3 # with no per-entry credentials; `publishers:` supports per-entry
bucket: "{{ .Env.S3_BUCKET }}" # `env:` instead, so it's used to invoke scripts/upload-r2.sh once per
region: "{{ .Env.S3_REGION }}" # artifact. Custom publishers inherit almost nothing from the
directory: "gitea-runner/{{.Version}}" # environment, hence the explicit R2_* forwarding below.
extra_files:
- glob: ./**.xz
- glob: ./**.sha256
# Mirrors the S3 `blobs:` upload above into Cloudflare R2 during the
# parallel S3+R2 period (S3 will be removed once migration completes).
# A second `blobs:` entry is impossible here since the blob pipe
# authenticates from the global AWS_* env with no per-entry
# credentials; `publishers:` supports per-entry `env:` instead, so
# it's used to invoke scripts/upload-r2.sh once per artifact. Custom
# publishers inherit almost nothing from the environment, hence the
# explicit R2_* forwarding below.
# #
# This publisher fires 109 times for 73 distinct keys because # This publisher fires 109 times for 73 distinct keys because
# goreleaser's release pipe already registers `release.extra_files` # goreleaser's release pipe already registers `release.extra_files`
@@ -125,7 +113,7 @@ publishers:
- R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }} - R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }}
archives: archives:
- format: binary - formats: [binary]
name_template: "{{ .Binary }}" name_template: "{{ .Binary }}"
allow_different_binary_count: true allow_different_binary_count: true
+2 -2
View File
@@ -17,7 +17,7 @@ RUN make clean && make build
### DIND VARIANT ### DIND VARIANT
# #
# #
FROM docker:29.6.2-dind AS dind FROM docker:29.7.1-dind AS dind
ARG VERSION=dev ARG VERSION=dev
@@ -37,7 +37,7 @@ ENTRYPOINT ["s6-svscan","/etc/s6"]
### DIND-ROOTLESS VARIANT ### DIND-ROOTLESS VARIANT
# #
# #
FROM docker:29.6.2-dind-rootless AS dind-rootless FROM docker:29.7.1-dind-rootless AS dind-rootless
ARG VERSION=dev ARG VERSION=dev
+42 -2
View File
@@ -158,6 +158,32 @@ An edit keeps the comments and the key order of the file. Indentation becomes tw
`config get`, `set`, `add` and `remove` use `config.yaml` (or `config.yml`) from the working directory, then from the directory of the binary, and print their choice to stderr. `config init` writes `config.yaml` in the working directory, and refuses to overwrite an existing config without `--force`. Pass `-c` for another path. `config get`, `set`, `add` and `remove` use `config.yaml` (or `config.yml`) from the working directory, then from the directory of the binary, and print their choice to stderr. `config init` writes `config.yaml` in the working directory, and refuses to overwrite an existing config without `--force`. Pass `-c` for another path.
#### Tool cache
Setup actions like `setup-go` install tools into `RUNNER_TOOL_CACHE`, which is `/opt/hostedtoolcache` inside a job. `runner.tool_cache_mode` selects what backs it:
| Mode | Tool cache | Trade-off |
| --- | --- | --- |
| `none` (default) | Per job, provided by the job image | A version the image lacks is downloaded in every job |
| `shared` | One volume reused by every job | Two jobs writing the same tool version at once corrupt it, so use it only with `runner.capacity: 1` |
With `none`, tools must come from the job image. Install them into `/opt/hostedtoolcache/<tool>/<version>/<arch>`, with an empty `<arch>.complete` file next to the directory:
```dockerfile
RUN GO=$(curl -fsSL 'https://go.dev/dl/?mode=json' | grep -oP '"version": "\Kgo1\.26\.[0-9]*' | head -1); \
DIR="/opt/hostedtoolcache/go/${GO#go}/x64" && \
mkdir -p "$(dirname "$DIR")" && \
curl -fsSL "https://dl.google.com/go/${GO}.linux-amd64.tar.gz" | tar -xz -C /tmp && \
mv /tmp/go "$DIR" && \
touch "${DIR}.complete"
```
A workflow requesting a minor version, `go-version: "1.26"`, resolves to the newest matching version in the cache, so a patch update in the image still hits it.
Of the [runner images](https://gitea.com/gitea/runner-images), the `-full` flavour is the one that ships tools in this layout.
`gitea-runner exec` reads no config file and takes `--tool-cache-mode` instead, defaulting to `none`.
#### Environment variables #### Environment variables
Earlier releases let a few environment variables (`GITEA_DEBUG`, `GITEA_TRACE`, `GITEA_RUNNER_CAPACITY`, `GITEA_RUNNER_FILE`, `GITEA_RUNNER_ENVIRON`, `GITEA_RUNNER_ENV_FILE`) override parts of the config. They are gone, use the YAML file for all settings. The Docker images still read their own variables, such as `RUNNER_STATE_FILE`, see [scripts/run.sh](scripts/run.sh) and the container documentation below. Earlier releases let a few environment variables (`GITEA_DEBUG`, `GITEA_TRACE`, `GITEA_RUNNER_CAPACITY`, `GITEA_RUNNER_FILE`, `GITEA_RUNNER_ENVIRON`, `GITEA_RUNNER_ENV_FILE`) override parts of the config. They are gone, use the YAML file for all settings. The Docker images still read their own variables, such as `RUNNER_STATE_FILE`, see [scripts/run.sh](scripts/run.sh) and the container documentation below.
@@ -202,7 +228,7 @@ a workflow with `runs-on: ubuntu-latest` is executed in the `runner-images:ubunt
Names may themselves contain a colon (for example `pool:e57e18d4-10d4-406f-93bf-60f127221bdd`); only `host` and `docker` are treated as schemas. Names may themselves contain a colon (for example `pool:e57e18d4-10d4-406f-93bf-60f127221bdd`); only `host` and `docker` are treated as schemas.
If a job's `runs-on` matches none of the runner's labels, the job still runs, in the default `docker.gitea.com/runner-images:ubuntu-latest` image. Images maintained for this purpose are listed at [gitea/runner-images](https://gitea.com/gitea/runner-images). If a job's `runs-on` matches none of the runner's labels, or sets no `runs-on` at all, it still runs: in `runner.default_image` where docker is available, on the host where it is not. Images maintained for this purpose are listed at [gitea/runner-images](https://gitea.com/gitea/runner-images).
Labels are chosen at registration time (`--labels`, or the interactive prompt) and can be changed afterwards by editing `runner.labels` in the config file, or in the Gitea UI under the runner's settings. Labels are chosen at registration time (`--labels`, or the interactive prompt) and can be changed afterwards by editing `runner.labels` in the config file, or in the Gitea UI under the runner's settings.
@@ -273,6 +299,12 @@ A password in a proxy URL is hidden in job logs. Any step can still read it, bec
Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default. Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default.
**Eviction**
An entry nothing has read or written for `retention` is removed, and a repository past `repo_size_limit` loses its least recently accessed entries until it fits; `size_limit` caps the whole cache the same way. Age alone never retires an entry still in use, and whatever these allow, the cache keeps free space above `health_check.min_free_disk_space_mb` when health checks are enabled.
These apply where the cache server runs, so on a shared server they belong in *its* config, not the runners'. See `retention`, `repo_size_limit`, `size_limit` and `sweep_interval` in [config.example.yaml](internal/pkg/config/config.example.yaml) for units and defaults.
**Cache service v2** **Cache service v2**
`actions/cache@v4.2` and later can use the *cache service v2* API. The runner serves it from the same store as v1, on by default, and it works with `external_server`. Turn it off with: `actions/cache@v4.2` and later can use the *cache service v2* API. The runner serves it from the same store as v1, on by default, and it works with `external_server`. Turn it off with:
@@ -282,7 +314,7 @@ cache:
v2: false v2: false
``` ```
Those actions refuse any host they do not take for GitHub. Rather than misreport the server URL, the runner edits that check out of the action's own bundle and keeps the untouched copy beside it; a bundle it does not recognise is left alone and keeps to v1. The same edit lets the stock `actions/upload-artifact` and `actions/download-artifact` work from `v4.4.0` on, without the `gitea-upload-artifact` fork. Those actions refuse any host they do not take for GitHub. Rather than misreport the server URL, the runner edits that check out of the action's own bundle on its way into the job, undone whenever the action is downloaded again. A bundle it does not recognise is left alone and keeps to v1. The same edit lets the stock `actions/upload-artifact` and `actions/download-artifact` work from `v4.4.0` on, without the `gitea-upload-artifact` fork, so it is made whatever `v2` says: that setting only governs the API the runner advertises. Set `runner.patch_actions: false` to leave every bundle exactly as shipped, an escape hatch for an action the edit breaks. The artifact actions then refuse again and the cache client keeps to v1.
**Shared cache across multiple runners** **Shared cache across multiple runners**
@@ -313,6 +345,8 @@ Run one dedicated `gitea-runner cache-server` that all runners point at.
# external_secret_file: /path/to/secret # secret can also be passed via a file # external_secret_file: /path/to/secret # secret can also be passed via a file
``` ```
Jobs reach the cache server at `external_server`, so when a reverse proxy fronts the server, point `external_server` at the proxy. The cache server itself needs no extra configuration.
Alternatively, mount the same NFS/CIFS share on every runner and point `cache.dir` at it — simpler, but with weaker isolation between repositories. Alternatively, mount the same NFS/CIFS share on every runner and point `cache.dir` at it — simpler, but with weaker isolation between repositories.
**S3 / MinIO** — mount object storage as a FUSE filesystem (e.g. [s3fs](https://github.com/s3fs-fuse/s3fs-fuse) or [goofys](https://github.com/kahing/goofys)) and set `cache.dir` to the mount point. **S3 / MinIO** — mount object storage as a FUSE filesystem (e.g. [s3fs](https://github.com/s3fs-fuse/s3fs-fuse) or [goofys](https://github.com/kahing/goofys)) and set `cache.dir` to the mount point.
@@ -352,6 +386,12 @@ Both hooks are synchronous and block the job while they run. Either one exiting
See **[docs/job-hooks.md](docs/job-hooks.md)** for the execution order, environment, and platform notes. See **[docs/job-hooks.md](docs/job-hooks.md)** for the execution order, environment, and platform notes.
#### Local job logs (`log.job.dir`)
Set `log.job.dir` to a path and the runner writes a copy of every task's log there as `<start time>-task-<id>.log`: the rows exactly as Gitea received them, with the same secrets masked and the job's result on the last line. Off by default, and what Gitea shows does not change.
`log.job.retention` (default `168h`) is how long a log is kept, expired ones being deleted as new tasks start, and `log.job.max_size` (default `1GB`) caps one log. Keep `retention` above `runner.timeout` so a long job cannot outlive its own log, and prefer local disk, the file is written while the job runs. Only the runner's own user can read it.
### Example Deployments ### Example Deployments
Check out the [examples](examples) directory for sample deployment types. Check out the [examples](examples) directory for sample deployment types.
+287 -99
View File
@@ -5,6 +5,7 @@
package artifactcache package artifactcache
import ( import (
"cmp"
"context" "context"
"crypto/hmac" "crypto/hmac"
"crypto/rand" "crypto/rand"
@@ -20,6 +21,7 @@ import (
"os" "os"
"path/filepath" "path/filepath"
"regexp" "regexp"
"slices"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
@@ -27,6 +29,7 @@ import (
"time" "time"
"gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/internal/pkg/disk"
"github.com/julienschmidt/httprouter" "github.com/julienschmidt/httprouter"
"github.com/sirupsen/logrus" "github.com/sirupsen/logrus"
@@ -59,6 +62,9 @@ type JobCredential struct {
// remote runner registers with. // remote runner registers with.
Results string `json:"results"` Results string `json:"results"`
InsecureTLS bool `json:"insecure_tls"` InsecureTLS bool `json:"insecure_tls"`
// PublicURL is this server as a reverse proxy makes the job reach it, not the listen address.
PublicURL string `json:"public_url"`
} }
// credEntry holds a registered job's credential along with an active // credEntry holds a registered job's credential along with an active
@@ -100,19 +106,38 @@ type Handler struct {
credMu sync.RWMutex credMu sync.RWMutex
creds map[string]*credEntry creds map[string]*credEntry
policy Policy
// freeDisk is a field so tests can drive evictForFreeSpace without a full volume.
freeDisk func(string) (uint64, error)
}
// Options configures a cache server started by StartHandler; the zero value is usable.
type Options struct {
Dir string
OutboundIP string
Port uint16
// InternalSecret, when non-empty, enables a control-plane API at
// /_internal/{register,revoke} that lets a remote runner pre-register the
// per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. The
// embedded in-process handler leaves it empty and registers tokens via the
// in-process RegisterJob method directly.
InternalSecret string
Policy Policy
Logger logrus.FieldLogger
} }
// StartHandler opens the on-disk cache store and starts the HTTP server. // StartHandler opens the on-disk cache store and starts the HTTP server.
// func StartHandler(opts Options) (*Handler, error) {
// internalSecret, when non-empty, enables a control-plane API at dir, logger := opts.Dir, opts.Logger
// /_internal/{register,revoke} that lets a remote runner pre-register the
// per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. The
// embedded in-process handler leaves it empty and registers tokens via the
// in-process RegisterJob method directly.
func StartHandler(dir, outboundIP string, port uint16, internalSecret string, logger logrus.FieldLogger) (*Handler, error) {
h := &Handler{ h := &Handler{
creds: make(map[string]*credEntry), creds: make(map[string]*credEntry),
internalSecret: internalSecret, internalSecret: opts.InternalSecret,
policy: opts.Policy.withDefaults(),
freeDisk: disk.FreeBytes,
} }
if logger == nil { if logger == nil {
@@ -142,8 +167,8 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
} }
h.storage = storage h.storage = storage
if outboundIP != "" { if opts.OutboundIP != "" {
h.outboundIP = outboundIP h.outboundIP = opts.OutboundIP
} else if ip := common.GetOutboundIP(); ip == nil { } else if ip := common.GetOutboundIP(); ip == nil {
return nil, errors.New("unable to determine outbound IP address") return nil, errors.New("unable to determine outbound IP address")
} else { } else {
@@ -182,7 +207,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
// can break Docker Desktop variants where the host's outbound IP is not // can break Docker Desktop variants where the host's outbound IP is not
// routable from inside the container network. Authentication is enforced // routable from inside the container network. Authentication is enforced
// by the bearer middleware and per-repo scoping, not by reachability. // by the bearer middleware and per-repo scoping, not by reachability.
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port)) listener, err := net.Listen("tcp", fmt.Sprintf(":%d", opts.Port))
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -212,6 +237,13 @@ func (h *Handler) ExternalURL() string {
return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port) return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
} }
func (h *Handler) baseURL(cred JobCredential) string {
if base := strings.TrimRight(cred.PublicURL, "/"); base != "" {
return base
}
return h.ExternalURL()
}
// RegisterJob makes token a valid bearer credential for cache requests from // RegisterJob makes token a valid bearer credential for cache requests from
// the given repository and returns a function that removes it. The runner // the given repository and returns a function that removes it. The runner
// calls this at job start and defers the returned func so that the credential // calls this at job start and defers the returned func so that the credential
@@ -359,7 +391,7 @@ func (h *Handler) find(w http.ResponseWriter, r *http.Request, _ httprouter.Para
} }
h.responseJSON(w, r, 200, map[string]any{ h.responseJSON(w, r, 200, map[string]any{
"result": "hit", "result": "hit",
"archiveLocation": h.signedArtifactURL(cache.ID, time.Now().Add(artifactURLTTL)), "archiveLocation": h.signedArtifactURL(cred, cache.ID, time.Now().Add(artifactURLTTL)),
"cacheKey": cache.Key, "cacheKey": cache.Key,
}) })
} }
@@ -380,6 +412,9 @@ func (h *Handler) lookupCache(db *bolthold.Store, repo string, keys []string, ve
_ = db.Delete(cache.ID, cache) _ = db.Delete(cache.ID, cache)
return nil, nil //nolint:nilnil // absence is not an error here return nil, nil //nolint:nilnil // absence is not an error here
} }
// Handing out a download URL counts as access, or eviction could drop the entry between
// this call and the GET that follows it.
h.touch(db, cache)
return cache, nil return cache, nil
} }
@@ -517,13 +552,22 @@ func (h *Handler) commitCache(cache *Cache) error {
// write real size back to cache, it may be different from the current value when the request doesn't specify it. // write real size back to cache, it may be different from the current value when the request doesn't specify it.
cache.Size = written cache.Size = written
cache.Complete = true cache.Complete = true
cache.UsedAt = time.Now().Unix() // a just-written entry counts as accessed, so it cannot be its own eviction victim
db, err := h.openDB() db, err := h.openDB()
if err != nil { if err != nil {
return err return err
} }
defer db.Close() defer db.Close()
return db.Update(cache.ID, cache) if err := db.Update(cache.ID, cache); err != nil {
return err
}
// A commit is the only thing that grows the store, so the only thing that can push the
// volume under the floor.
h.evictRepo(db, cache.Repo)
h.evictTotal(db)
h.evictForFreeSpace(db)
return nil
} }
// GET /_apis/artifactcache/artifacts/:id // GET /_apis/artifactcache/artifacts/:id
@@ -641,7 +685,7 @@ func (h *Handler) ResultsURL(cred JobCredential) string {
if h == nil || cred.Results == "" { if h == nil || cred.Results == "" {
return "" return ""
} }
return h.ExternalURL() return h.baseURL(cred)
} }
func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ httprouter.Params) { func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
@@ -700,16 +744,16 @@ func (h *Handler) computeSignature(purpose string, cacheID, exp int64) string {
} }
// signedURL builds a URL under path that signedAuth accepts for the same purpose. // signedURL builds a URL under path that signedAuth accepts for the same purpose.
func (h *Handler) signedURL(path, purpose string, cacheID uint64, exp time.Time) string { func (h *Handler) signedURL(cred JobCredential, path, purpose string, cacheID uint64, exp time.Time) string {
expUnix := exp.Unix() expUnix := exp.Unix()
q := url.Values{} q := url.Values{}
q.Set("exp", strconv.FormatInt(expUnix, 10)) q.Set("exp", strconv.FormatInt(expUnix, 10))
q.Set("sig", h.computeSignature(purpose, int64(cacheID), expUnix)) q.Set("sig", h.computeSignature(purpose, int64(cacheID), expUnix))
return fmt.Sprintf("%s%s/%d?%s", h.ExternalURL(), path, cacheID, q.Encode()) return fmt.Sprintf("%s%s/%d?%s", h.baseURL(cred), path, cacheID, q.Encode())
} }
func (h *Handler) signedArtifactURL(cacheID uint64, exp time.Time) string { func (h *Handler) signedArtifactURL(cred JobCredential, cacheID uint64, exp time.Time) string {
return h.signedURL(apiPath+"/artifacts", "", cacheID, exp) return h.signedURL(cred, apiPath+"/artifacts", "", cacheID, exp)
} }
// if not found, return (nil, nil) instead of an error. // if not found, return (nil, nil) instead of an error.
@@ -811,12 +855,43 @@ func (h *Handler) touchCache(id uint64, requireIncomplete bool) error {
} }
const ( const (
keepUsed = 30 * 24 * time.Hour miB = 1024 * 1024
keepUnused = 7 * 24 * time.Hour
keepTemp = 5 * time.Minute defaultSweepInterval = time.Hour
keepOld = 5 * time.Minute
// inUseGrace matches artifactURLTTL so an entry outlives every signed URL still usable
// for it, and no sweep cuts off a download in progress.
inUseGrace = artifactURLTTL
// uploadStallTimeout is how long a reservation may sit without a chunk before it counts
// as abandoned. Widening it also widens the window for findExactCache to hand a finalize
// a stale reservation.
uploadStallTimeout = 5 * time.Minute
defaultMinFreeDisk = 1024 * miB
) )
// Policy bounds what the cache server keeps: a retention window counted from last access,
// and size limits that evict least recently accessed first. A zero limit is no limit.
type Policy struct {
Retention time.Duration // Retention removes entries nothing has read or written within this window. Zero keeps them regardless of age.
RepoSizeLimit int64 // RepoSizeLimit caps one repository's completed entries in bytes, evicting least recently accessed first.
SizeLimit int64 // SizeLimit caps every repository's completed entries together, in bytes.
SweepInterval time.Duration // SweepInterval is the minimum time between two eviction sweeps.
MinFreeDisk int64 // MinFreeDisk is volume headroom the cache will not eat into. Tracks the runner's health-check floor rather than taking a key of its own.
}
func (p Policy) withDefaults() Policy {
// The limits default in config.LoadDefault, so a written 0 means off.
if p.MinFreeDisk <= 0 {
p.MinFreeDisk = defaultMinFreeDisk
}
if p.SweepInterval <= 0 {
p.SweepInterval = defaultSweepInterval
}
return p
}
func (h *Handler) gcCache() { func (h *Handler) gcCache() {
if h.gcing.Load() { if h.gcing.Load() {
return return
@@ -826,7 +901,7 @@ func (h *Handler) gcCache() {
} }
defer h.gcing.Store(false) defer h.gcing.Store(false)
if time.Since(h.gcAt) < time.Hour { if time.Since(h.gcAt) < h.policy.SweepInterval {
h.logger.Debugf("skip gc: %v", h.gcAt.String()) h.logger.Debugf("skip gc: %v", h.gcAt.String())
return return
} }
@@ -839,95 +914,208 @@ func (h *Handler) gcCache() {
} }
defer db.Close() defer db.Close()
// Remove the caches which are not completed for a while, they are most likely to be broken. h.evictIncomplete(db)
var caches []*Cache h.evictExpired(db)
if err := db.Find(&caches, bolthold. h.evictSuperseded(db)
Where("UsedAt").Lt(time.Now().Add(-keepTemp).Unix()). h.evictOversized(db)
And("Complete").Eq(false), h.evictForFreeSpace(db)
); err != nil { }
h.logger.Warnf("find caches: %v", err)
} else { // evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never
for _, cache := range caches { // accounted for.
h.storage.Remove(cache.ID) func (h *Handler) evictForFreeSpace(db *bolthold.Store) {
if err := db.Delete(cache.ID, cache); err != nil { free, err := h.freeDisk(h.dir)
h.logger.Warnf("delete cache: %v", err) if err != nil {
continue h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full
} return
h.logger.Infof("deleted cache: %+v", cache) }
} if free >= uint64(h.policy.MinFreeDisk) {
return
} }
// Remove the old caches which have not been used recently. caches := h.completedByUse(db)
caches = caches[:0] total, shortfall := totalSize(caches), h.policy.MinFreeDisk-int64(free)
if err := db.Find(&caches, bolthold. if total <= shortfall {
Where("UsedAt").Lt(time.Now().Add(-keepUnused).Unix()), // Say so, or shedding everything and still being short reads as the backstop working.
); err != nil { h.logger.Warnf("cache volume is %d MiB short of the free space floor with only %d MiB of cache on it; something else is filling it", shortfall/miB, total/miB)
h.logger.Warnf("find caches: %v", err)
} else {
for _, cache := range caches {
h.storage.Remove(cache.ID)
if err := db.Delete(cache.ID, cache); err != nil {
h.logger.Warnf("delete cache: %v", err)
continue
}
h.logger.Infof("deleted cache: %+v", cache)
}
} }
h.evictTo(db, caches, total-shortfall, "the cache volume")
}
// Remove the old caches which are too old. // evictIncomplete removes uploads that stopped part way, which are most likely broken.
caches = caches[:0] func (h *Handler) evictIncomplete(db *bolthold.Store) {
if err := db.Find(&caches, bolthold. h.sweep(db, bolthold.
Where("CreatedAt").Lt(time.Now().Add(-keepUsed).Unix()), Where("UsedAt").Lt(time.Now().Add(-uploadStallTimeout).Unix()).
); err != nil { And("Complete").Eq(false).
h.logger.Warnf("find caches: %v", err) Index("UsedAt"))
} else { }
for _, cache := range caches {
h.storage.Remove(cache.ID) func (h *Handler) evictExpired(db *bolthold.Store) {
if err := db.Delete(cache.ID, cache); err != nil { if h.policy.Retention <= 0 {
h.logger.Warnf("delete cache: %v", err) return
continue
}
h.logger.Infof("deleted cache: %+v", cache)
}
} }
// Never below inUseGrace, or a short retention would outrun a signed URL already issued.
window := max(h.policy.Retention, inUseGrace)
h.sweep(db, bolthold.Where("UsedAt").Lt(time.Now().Add(-window).Unix()).Index("UsedAt"))
}
// Remove the old caches with the same key and version within the same // evictSuperseded removes entries a newer one with the same key and version replaced. The
// repository, keep the latest one. Aggregation must include Repo so two // aggregation includes Repo so two repos sharing a (key, version) do not evict each other.
// repos that happen to share a (key, version) do not evict each other — func (h *Handler) evictSuperseded(db *bolthold.Store) {
// otherwise per-repo scoping holds for reads but one repo can age results, err := db.FindAggregate(&Cache{}, bolthold.Where("Complete").Eq(true).Index("Complete"), "Repo", "Key", "Version")
// another out after keepOld. if err != nil {
// Also keep the olds which have been used recently for a while in case of the cache is still in use.
if results, err := db.FindAggregate(
&Cache{},
bolthold.Where("Complete").Eq(true),
"Repo", "Key", "Version",
); err != nil {
h.logger.Warnf("find aggregate caches: %v", err) h.logger.Warnf("find aggregate caches: %v", err)
} else { return
for _, result := range results { }
if result.Count() <= 1 { var caches []*Cache
for _, result := range results {
if result.Count() <= 1 {
continue
}
result.Sort("CreatedAt")
caches = caches[:0]
result.Reduction(&caches)
for _, cache := range caches[:len(caches)-1] {
if inUse(cache) {
continue continue
} }
result.Sort("CreatedAt") h.deleteCache(db, cache)
caches = caches[:0]
result.Reduction(&caches)
for _, cache := range caches[:len(caches)-1] {
if time.Since(time.Unix(cache.UsedAt, 0)) < keepOld {
// Keep it since it has been used recently, even if it's old.
// Or it could break downloading in process.
continue
}
h.storage.Remove(cache.ID)
if err := db.Delete(cache.ID, cache); err != nil {
h.logger.Warnf("delete cache: %v", err)
continue
}
h.logger.Infof("deleted cache: %+v", cache)
}
} }
} }
} }
// evictOversized applies the per-repository limit, then the whole-store one. Only completed
// entries count, since only those carry a size measured at commit rather than claimed.
func (h *Handler) evictOversized(db *bolthold.Store) {
if h.policy.RepoSizeLimit > 0 {
byRepo := make(map[string][]*Cache)
for _, cache := range h.completedByUse(db) {
byRepo[cache.Repo] = append(byRepo[cache.Repo], cache)
}
for repo, caches := range byRepo {
h.evictTo(db, caches, h.policy.RepoSizeLimit, "repository "+repo)
}
}
h.evictTotal(db)
}
// evictTotal caps the store as a whole. It re-queries because the per-repo pass may have
// deleted rows an earlier result still holds.
func (h *Handler) evictTotal(db *bolthold.Store) {
if h.policy.SizeLimit <= 0 {
return
}
h.evictTo(db, h.completedByUse(db), h.policy.SizeLimit, "the cache")
}
// evictRepo reclaims space when a commit pushes a repo over, rather than at the next sweep.
func (h *Handler) evictRepo(db *bolthold.Store, repo string) {
if h.policy.RepoSizeLimit <= 0 {
return
}
h.evictTo(db, h.cachesByUse(db, bolthold.Where("Repo").Eq(repo).And("Complete").Eq(true).Index("Repo")), h.policy.RepoSizeLimit, "repository "+repo)
}
// evictTo deletes until caches fit limit. caches must be ordered by UsedAt ascending.
func (h *Handler) evictTo(db *bolthold.Store, caches []*Cache, limit int64, scope string) {
// An entry bigger than the limit never fits, so it goes on its own account instead of
// dragging every neighbour out first and then following them next sweep.
fits := caches[:0]
for _, cache := range caches {
if cache.Size <= limit {
fits = append(fits, cache)
continue
}
if !inUse(cache) {
h.logger.Warnf("cache %q is %d MiB on its own, over the limit for %s; dropping it", cache.Key, cache.Size/miB, scope)
h.deleteCache(db, cache)
}
}
caches = fits
total := totalSize(caches)
var freed int64
for _, cache := range caches {
if total <= limit {
break
}
if inUse(cache) || !h.deleteCache(db, cache) {
continue
}
total -= cache.Size
freed += cache.Size
}
if freed > 0 {
h.logger.Warnf("evicted %d MiB from %s, least recently used first", freed/miB, scope)
}
}
// inUse reports whether an entry was read or written recently enough that removing it
// could break a download in progress.
func inUse(cache *Cache) bool {
return time.Since(time.Unix(cache.UsedAt, 0)) < inUseGrace
}
// touch stamps UsedAt through the caller's store, a bolt write on the read path. It cannot
// go through touchCache, which opens its own store and would block on the exclusive lock
// for as long as the caller holds one.
func (h *Handler) touch(db *bolthold.Store, cache *Cache) {
cache.UsedAt = time.Now().Unix()
if err := db.Update(cache.ID, cache); err != nil {
h.logger.Warnf("touch cache: %v", err)
}
}
func (h *Handler) sweep(db *bolthold.Store, query *bolthold.Query) {
for _, cache := range h.caches(db, query) {
h.deleteCache(db, cache)
}
}
func (h *Handler) caches(db *bolthold.Store, query *bolthold.Query) []*Cache {
var caches []*Cache
if err := db.Find(&caches, query); err != nil {
h.logger.Warnf("find caches: %v", err)
}
return caches
}
// cachesByUse returns matches least recently accessed first, sorting here rather than with
// bolthold's SortBy, which reflects over every field it compares.
func (h *Handler) cachesByUse(db *bolthold.Store, query *bolthold.Query) []*Cache {
caches := h.caches(db, query)
slices.SortFunc(caches, func(a, b *Cache) int { return cmp.Compare(a.UsedAt, b.UsedAt) })
return caches
}
// completedByUse returns every entry the size limits count, least recently accessed first.
func (h *Handler) completedByUse(db *bolthold.Store) []*Cache {
return h.cachesByUse(db, bolthold.Where("Complete").Eq(true).Index("Complete"))
}
func totalSize(caches []*Cache) int64 {
var total int64
for _, cache := range caches {
total += cache.Size
}
return total
}
// deleteCache drops an entry and its bytes, reporting whether it went fully. The blob goes
// first, so a failed unlink leaves the row for the next sweep instead of orphaning bytes.
func (h *Handler) deleteCache(db *bolthold.Store, cache *Cache) bool {
if err := h.storage.Remove(cache.ID); err != nil {
h.logger.Warnf("remove cache blob: %v", err)
return false
}
if err := db.Delete(cache.ID, cache); err != nil {
h.logger.Warnf("delete cache: %v", err)
return false
}
h.logger.Infof("deleted cache: %+v", cache)
return true
}
func (h *Handler) responseJSON(w http.ResponseWriter, r *http.Request, code int, v ...any) { func (h *Handler) responseJSON(w http.ResponseWriter, r *http.Request, code int, v ...any) {
w.Header().Set("Content-Type", "application/json; charset=utf-8") w.Header().Set("Content-Type", "application/json; charset=utf-8")
var data []byte var data []byte
+289 -31
View File
@@ -8,6 +8,7 @@ import (
"bytes" "bytes"
"crypto/rand" "crypto/rand"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
@@ -41,16 +42,19 @@ func (b *bearerTransport) RoundTrip(r *http.Request) (*http.Response, error) {
var testClient = &http.Client{Transport: &bearerTransport{token: testToken}} var testClient = &http.Client{Transport: &bearerTransport{token: testToken}}
// testRetention mirrors config.DefaultCacheRetention; Policy has no defaults of its own.
const testRetention = 7 * 24 * time.Hour
// signArtifactURL builds a signed download URL the same way the server does; // signArtifactURL builds a signed download URL the same way the server does;
// tests use it to reach the get handler directly without going through a // tests use it to reach the get handler directly without going through a
// find/cache-hit round trip. // find/cache-hit round trip.
func signArtifactURL(h *Handler, id int64) string { func signArtifactURL(h *Handler, id int64) string {
return h.signedArtifactURL(uint64(id), time.Now().Add(artifactURLTTL)) return h.signedArtifactURL(JobCredential{}, uint64(id), time.Now().Add(artifactURLTTL))
} }
func TestHandler(t *testing.T) { func TestHandler(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
@@ -656,7 +660,7 @@ func backdateCache(t *testing.T, handler *Handler, key string, age time.Duration
require.NoError(t, db.Update(caches[0].ID, caches[0])) require.NoError(t, db.Update(caches[0].ID, caches[0]))
} }
func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) { //nolint:unparam // pre-existing issue from nektos/act func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) {
var id uint64 var id uint64
{ {
body, err := json.Marshal(&Request{ body, err := json.Marshal(&Request{
@@ -722,7 +726,7 @@ func uploadCacheNormally(t *testing.T, base, key, version string, content []byte
func TestHandler_gcCache(t *testing.T) { func TestHandler_gcCache(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir, Policy: Policy{Retention: testRetention}})
require.NoError(t, err) require.NoError(t, err)
defer func() { defer func() {
@@ -752,8 +756,8 @@ func TestHandler_gcCache(t *testing.T) {
Key: "test_key_2", Key: "test_key_2",
Version: "test_version", Version: "test_version",
Complete: false, Complete: false,
UsedAt: now.Add(-(keepTemp + time.Second)).Unix(), UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(),
CreatedAt: now.Add(-(keepTemp + time.Hour)).Unix(), CreatedAt: now.Add(-(inUseGrace + time.Hour)).Unix(),
}, },
Kept: false, Kept: false,
}, },
@@ -763,21 +767,21 @@ func TestHandler_gcCache(t *testing.T) {
Key: "test_key_3", Key: "test_key_3",
Version: "test_version", Version: "test_version",
Complete: true, Complete: true,
UsedAt: now.Add(-(keepUnused + time.Second)).Unix(), UsedAt: now.Add(-(testRetention + time.Second)).Unix(),
CreatedAt: now.Add(-(keepUnused + time.Hour)).Unix(), CreatedAt: now.Add(-(testRetention + time.Hour)).Unix(),
}, },
Kept: false, Kept: false,
}, },
{ {
// should be removed, since it's used but too old. // should be kept, since age alone does not retire an entry that is still used.
Cache: &Cache{ Cache: &Cache{
Key: "test_key_3", Key: "test_key_3",
Version: "test_version", Version: "test_version",
Complete: true, Complete: true,
UsedAt: now.Unix(), UsedAt: now.Unix(),
CreatedAt: now.Add(-(keepUsed + time.Second)).Unix(), CreatedAt: now.Add(-365 * 24 * time.Hour).Unix(),
}, },
Kept: false, Kept: true,
}, },
{ {
// should be kept, since it has a newer edition but be used recently. // should be kept, since it has a newer edition but be used recently.
@@ -785,7 +789,7 @@ func TestHandler_gcCache(t *testing.T) {
Key: "test_key_1", Key: "test_key_1",
Version: "test_version", Version: "test_version",
Complete: true, Complete: true,
UsedAt: now.Add(-(keepOld - time.Minute)).Unix(), UsedAt: now.Add(-(inUseGrace - time.Minute)).Unix(),
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(), CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
}, },
Kept: true, Kept: true,
@@ -796,7 +800,7 @@ func TestHandler_gcCache(t *testing.T) {
Key: "test_key_1", Key: "test_key_1",
Version: "test_version", Version: "test_version",
Complete: true, Complete: true,
UsedAt: now.Add(-(keepOld + time.Second)).Unix(), UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(),
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(), CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
}, },
Kept: false, Kept: false,
@@ -829,11 +833,265 @@ func TestHandler_gcCache(t *testing.T) {
require.NoError(t, db.Close()) require.NoError(t, db.Close())
} }
// TestHandler_evictPolicy covers the non-default policies; TestHandler_gcCache covers the
// defaults across every pass.
func TestHandler_evictPolicy(t *testing.T) {
now := time.Now()
stale := func(d time.Duration) int64 { return now.Add(-d).Unix() }
mib := func(n int64) int64 { return n * miB }
for _, tc := range []struct {
name string
policy Policy
entries []*Cache
kept []string
}{
{
name: "a zero retention keeps an entry nothing has touched",
policy: Policy{Retention: 0},
entries: []*Cache{
{Key: "idle", UsedAt: stale(testRetention + time.Hour)},
},
kept: []string{"idle"},
},
{
name: "evicts least recently accessed until the repository fits",
policy: Policy{RepoSizeLimit: mib(10)},
entries: []*Cache{
{Repo: "o/a", Key: "oldest", Size: mib(4), UsedAt: stale(3 * time.Hour)},
{Repo: "o/a", Key: "middle", Size: mib(4), UsedAt: stale(2 * time.Hour)},
{Repo: "o/a", Key: "newest", Size: mib(4), UsedAt: stale(time.Hour)},
},
kept: []string{"middle", "newest"},
},
{
name: "spares entries that may still be downloading",
policy: Policy{RepoSizeLimit: mib(10)},
entries: []*Cache{
{Repo: "o/a", Key: "fresh_1", Size: mib(6), UsedAt: stale(time.Minute)},
{Repo: "o/a", Key: "fresh_2", Size: mib(6), UsedAt: stale(time.Minute)},
},
kept: []string{"fresh_1", "fresh_2"},
},
{
name: "one repository over its limit leaves another alone",
policy: Policy{RepoSizeLimit: mib(10)},
entries: []*Cache{
{Repo: "o/a", Key: "a_old", Size: mib(6), UsedAt: stale(3 * time.Hour)},
{Repo: "o/a", Key: "a_new", Size: mib(6), UsedAt: stale(time.Hour)},
{Repo: "o/b", Key: "b_old", Size: mib(6), UsedAt: stale(4 * time.Hour)},
},
kept: []string{"a_new", "b_old"},
},
{
name: "the total limit evicts across repositories once each fits its own",
policy: Policy{RepoSizeLimit: mib(10), SizeLimit: mib(12)},
entries: []*Cache{
{Repo: "o/a", Key: "a_old", Size: mib(8), UsedAt: stale(3 * time.Hour)},
{Repo: "o/b", Key: "b_new", Size: mib(8), UsedAt: stale(time.Hour)},
},
kept: []string{"b_new"},
},
{
// Retention below inUseGrace would otherwise drop an entry whose signed URL a job
// is still holding.
name: "a retention shorter than the grace still spares a just-served entry",
policy: Policy{Retention: time.Minute},
entries: []*Cache{
{Key: "just_served", UsedAt: stale(2 * time.Minute)},
{Key: "idle", UsedAt: stale(time.Hour)},
},
kept: []string{"just_served"},
},
{
name: "an entry over the limit goes without emptying the repository",
policy: Policy{RepoSizeLimit: mib(10)},
entries: []*Cache{
{Repo: "o/a", Key: "keeps", Size: mib(4), UsedAt: stale(3 * time.Hour)},
{Repo: "o/a", Key: "huge", Size: mib(20), UsedAt: stale(2 * time.Hour)},
},
kept: []string{"keeps"},
},
{
name: "a zero limit keeps everything",
policy: Policy{RepoSizeLimit: 0},
entries: []*Cache{
{Repo: "o/a", Key: "huge_1", Size: mib(100), UsedAt: stale(3 * time.Hour)},
{Repo: "o/a", Key: "huge_2", Size: mib(100), UsedAt: stale(2 * time.Hour)},
},
kept: []string{"huge_1", "huge_2"},
},
} {
t.Run(tc.name, func(t *testing.T) {
for _, e := range tc.entries {
e.Complete = true // only completed entries carry a measured size, so only they count
}
handler := newTestHandler(t, tc.policy, tc.entries...)
handler.gcAt = time.Time{} // ensure gcCache will not skip
handler.gcCache()
assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, tc.entries))
})
}
}
// TestHandler_evictForFreeSpace proves the volume backstop sheds only what it must, and only
// when the disk is actually short.
func TestHandler_evictForFreeSpace(t *testing.T) {
free := func(n int64) func(string) (uint64, error) {
return func(string) (uint64, error) { return uint64(n), nil }
}
for _, tc := range []struct {
name string
freeDisk func(string) (uint64, error)
kept []string
}{
{"ample free space evicts nothing", free(defaultMinFreeDisk), []string{"oldest", "middle", "newest"}},
{"a small shortfall sheds one entry", free(defaultMinFreeDisk - 4*miB), []string{"middle", "newest"}},
{"a shortfall the cache cannot cover sheds all of it", free(0), nil},
{
"an unreadable volume is treated as unavailable, not as full",
func(string) (uint64, error) { return 0, errors.New("unsupported") },
[]string{"oldest", "middle", "newest"},
},
} {
t.Run(tc.name, func(t *testing.T) {
now := time.Now()
entries := []*Cache{
{Key: "oldest", Complete: true, Size: 4 * miB, UsedAt: now.Add(-3 * time.Hour).Unix()},
{Key: "middle", Complete: true, Size: 4 * miB, UsedAt: now.Add(-2 * time.Hour).Unix()},
{Key: "newest", Complete: true, Size: 4 * miB, UsedAt: now.Add(-time.Hour).Unix()},
}
handler := newTestHandler(t, Policy{}, entries...)
handler.freeDisk = tc.freeDisk
db, err := handler.openDB()
require.NoError(t, err)
handler.evictForFreeSpace(db)
require.NoError(t, db.Close())
assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, entries))
})
}
}
// TestHandler_SweepKeepsEntryWhenBlobSurvives proves a failed unlink leaves the row in place,
// so the next sweep retries rather than orphaning bytes no row points at and no limit counts.
func TestHandler_SweepKeepsEntryWhenBlobSurvives(t *testing.T) {
cache := &Cache{Key: "stuck", Complete: true, UsedAt: time.Now().Add(-(testRetention + time.Hour)).Unix()}
handler := newTestHandler(t, Policy{Retention: testRetention}, cache)
// A non-empty directory where the blob belongs makes os.Remove fail on every platform.
blob := handler.storage.filename(cache.ID)
require.NoError(t, os.MkdirAll(blob, 0o755))
require.NoError(t, os.WriteFile(filepath.Join(blob, "held"), []byte("x"), 0o600))
handler.gcAt = time.Time{}
handler.gcCache()
assert.Equal(t, []string{"stuck"}, keptKeys(t, handler, []*Cache{cache}), "the entry must outlive a blob that could not be removed")
}
// TestHandler_FindProtectsFromEviction covers the window between a find handing out a signed
// download URL and the GET that redeems it: the entry promised to a job must not be the next
// eviction victim just because its last access predates the find.
func TestHandler_FindProtectsFromEviction(t *testing.T) {
// 12 MiB against a 10 MiB limit, so exactly one entry has to go.
wanted := &Cache{Repo: testRepo, Key: "wanted", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-3 * time.Hour).Unix()}
other := &Cache{Repo: testRepo, Key: "other", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-2 * time.Hour).Unix()}
newest := &Cache{Repo: testRepo, Key: "newest", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()}
handler := newTestHandler(t, Policy{RepoSizeLimit: 10 * miB}, wanted, other, newest)
writeBlob(t, handler, wanted.ID) // find only reports a hit when the blob is on disk
resp, err := testClient.Get(fmt.Sprintf("%s%s/cache?keys=wanted&version=v", handler.ExternalURL(), apiPath))
require.NoError(t, err)
defer resp.Body.Close()
require.Equal(t, 200, resp.StatusCode)
// Evict directly: the request above kicked off an async gcCache, and writing gcAt here
// to drive gcCache would race its read.
db, err := handler.openDB()
require.NoError(t, err)
defer func() { require.NoError(t, db.Close()) }()
handler.evictOversized(db)
require.NoError(t, db.Get(wanted.ID, &Cache{}), "the entry just promised to a job must survive")
assert.ErrorIs(t, db.Get(other.ID, &Cache{}), bolthold.ErrNotFound, "the next least recently used goes instead")
}
// TestHandler_evictOnCommit proves a repository that goes over its limit gets space back at
// once, rather than waiting out the collection interval.
func TestHandler_evictOnCommit(t *testing.T) {
full := &Cache{Repo: testRepo, Key: "full", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()}
handler := newTestHandler(t, Policy{RepoSizeLimit: 4 * miB}, full)
// StartHandler already stamped gcAt, so the periodic sweep stays rate-limited out and
// only the commit path can evict.
uploadCacheNormally(t, handler.ExternalURL()+apiPath, "new", "v", []byte("some content"))
assert.Empty(t, keptKeys(t, handler, []*Cache{full}))
}
func TestHandler_gcCacheInterval(t *testing.T) {
cache := &Cache{Key: "temp", UsedAt: time.Now().Add(-time.Hour).Unix()}
// Half the default, so a sweep 45m ago is still inside the default but past this one.
handler := newTestHandler(t, Policy{SweepInterval: 30 * time.Minute}, cache)
handler.gcAt = time.Now().Add(-45 * time.Minute) // past the configured interval, still inside the default
handler.gcCache()
assert.Empty(t, keptKeys(t, handler, []*Cache{cache}))
}
// newTestHandler starts a handler with testToken registered, seeded with entries.
func newTestHandler(t *testing.T, policy Policy, entries ...*Cache) *Handler {
t.Helper()
handler, err := StartHandler(Options{
Dir: filepath.Join(t.TempDir(), "artifactcache"),
OutboundIP: "127.0.0.1",
Policy: policy,
})
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, handler.Close()) })
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
db, err := handler.openDB()
require.NoError(t, err)
for _, e := range entries {
require.NoError(t, insertCache(db, e))
}
require.NoError(t, db.Close())
return handler
}
// keptKeys reports which of entries are still in the store.
func keptKeys(t *testing.T, handler *Handler, entries []*Cache) []string {
t.Helper()
db, err := handler.openDB()
require.NoError(t, err)
defer func() { require.NoError(t, db.Close()) }()
var kept []string
for _, e := range entries {
if err := db.Get(e.ID, &Cache{}); err == nil {
kept = append(kept, e.Key)
}
}
return kept
}
// writeBlob gives an entry the on-disk bytes that find and get require.
func writeBlob(t *testing.T, handler *Handler, id uint64) {
t.Helper()
require.NoError(t, handler.storage.Write(id, 0, strings.NewReader("a")))
_, err := handler.storage.Commit(id, 1)
require.NoError(t, err)
}
// TestHandler_RejectsMissingBearer covers the advisory's root cause: // TestHandler_RejectsMissingBearer covers the advisory's root cause:
// unauthenticated access to management endpoints is now refused with 401. // unauthenticated access to management endpoints is now refused with 401.
func TestHandler_RejectsMissingBearer(t *testing.T) { func TestHandler_RejectsMissingBearer(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -866,7 +1124,7 @@ func TestHandler_RejectsMissingBearer(t *testing.T) {
// accepted after RegisterJob; stale/forged tokens cannot be replayed. // accepted after RegisterJob; stale/forged tokens cannot be replayed.
func TestHandler_RejectsUnknownBearer(t *testing.T) { func TestHandler_RejectsUnknownBearer(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -886,7 +1144,7 @@ func TestHandler_RejectsUnknownBearer(t *testing.T) {
// working the moment the job ends instead of living for the runner's lifetime. // working the moment the job ends instead of living for the runner's lifetime.
func TestHandler_UnregisterRevokes(t *testing.T) { func TestHandler_UnregisterRevokes(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -917,7 +1175,7 @@ func TestHandler_UnregisterRevokes(t *testing.T) {
// invisible to queries scoped to repoB. // invisible to queries scoped to repoB.
func TestHandler_CrossRepoIsolation(t *testing.T) { func TestHandler_CrossRepoIsolation(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
handler.RegisterJob("token-a", JobCredential{Repo: "owner/repoA"}) handler.RegisterJob("token-a", JobCredential{Repo: "owner/repoA"})
@@ -983,7 +1241,7 @@ func TestHandler_CrossRepoIsolation(t *testing.T) {
// working after artifactURLTTL even if the bearer token is still registered. // working after artifactURLTTL even if the bearer token is still registered.
func TestHandler_ArtifactSignature(t *testing.T) { func TestHandler_ArtifactSignature(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
@@ -998,7 +1256,7 @@ func TestHandler_ArtifactSignature(t *testing.T) {
}) })
t.Run("tampered signature", func(t *testing.T) { t.Run("tampered signature", func(t *testing.T) {
good := handler.signedArtifactURL(1, time.Now().Add(artifactURLTTL)) good := signArtifactURL(handler, 1)
bad := good[:len(good)-4] + "dead" bad := good[:len(good)-4] + "dead"
resp, err := testClient.Get(bad) resp, err := testClient.Get(bad)
require.NoError(t, err) require.NoError(t, err)
@@ -1007,7 +1265,7 @@ func TestHandler_ArtifactSignature(t *testing.T) {
}) })
t.Run("expired signature", func(t *testing.T) { t.Run("expired signature", func(t *testing.T) {
expired := handler.signedArtifactURL(1, time.Now().Add(-time.Second)) expired := handler.signedArtifactURL(JobCredential{}, 1, time.Now().Add(-time.Second))
resp, err := testClient.Get(expired) resp, err := testClient.Get(expired)
require.NoError(t, err) require.NoError(t, err)
resp.Body.Close() resp.Body.Close()
@@ -1016,10 +1274,10 @@ func TestHandler_ArtifactSignature(t *testing.T) {
t.Run("signature from a different server", func(t *testing.T) { t.Run("signature from a different server", func(t *testing.T) {
dir2 := filepath.Join(t.TempDir(), "artifactcache2") dir2 := filepath.Join(t.TempDir(), "artifactcache2")
other, err := StartHandler(dir2, "", 0, "", nil) other, err := StartHandler(Options{Dir: dir2})
require.NoError(t, err) require.NoError(t, err)
defer other.Close() defer other.Close()
otherURL := other.signedArtifactURL(1, time.Now().Add(artifactURLTTL)) otherURL := signArtifactURL(other, 1)
// Rewrite the host so the request still lands on our handler, but // Rewrite the host so the request still lands on our handler, but
// the signature was computed with a different secret. // the signature was computed with a different secret.
parts := strings.SplitN(otherURL, apiPath, 2) parts := strings.SplitN(otherURL, apiPath, 2)
@@ -1038,13 +1296,13 @@ func TestHandler_ArtifactSignature(t *testing.T) {
func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) { func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
first, err := StartHandler(dir, "127.0.0.1", 0, "", nil) first, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
exp := time.Now().Add(artifactURLTTL).Unix() exp := time.Now().Add(artifactURLTTL).Unix()
sig := first.computeSignature("", 42, exp) sig := first.computeSignature("", 42, exp)
require.NoError(t, first.Close()) require.NoError(t, first.Close())
second, err := StartHandler(dir, "127.0.0.1", 0, "", nil) second, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
defer second.Close() defer second.Close()
@@ -1056,7 +1314,7 @@ func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
// the auth refactor. // the auth refactor.
func TestHandler_ArtifactSignatureDownload(t *testing.T) { func TestHandler_ArtifactSignatureDownload(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
handler.RegisterJob(testToken, JobCredential{Repo: testRepo}) handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
@@ -1096,7 +1354,7 @@ func TestHandler_ArtifactSignatureDownload(t *testing.T) {
// (restart mid-task, retry), which must not kill the live job's auth. // (restart mid-task, retry), which must not kill the live job's auth.
func TestHandler_RegisterJob_RefCounted(t *testing.T) { func TestHandler_RegisterJob_RefCounted(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -1125,10 +1383,10 @@ func TestHandler_RegisterJob_RefCounted(t *testing.T) {
// TestHandler_GC_PerRepoDedup ensures duplicate-pruning does not evict // TestHandler_GC_PerRepoDedup ensures duplicate-pruning does not evict
// another repo's entry. Two repos reserve the same (key, version); after the // another repo's entry. Two repos reserve the same (key, version); after the
// keepOld window, GC must keep the one from each repo. // inUseGrace window, GC must keep the one from each repo.
func TestHandler_GC_PerRepoDedup(t *testing.T) { func TestHandler_GC_PerRepoDedup(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
handler.RegisterJob("tok-a", JobCredential{Repo: "owner/repoA"}) handler.RegisterJob("tok-a", JobCredential{Repo: "owner/repoA"})
@@ -1142,7 +1400,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) {
db, err := handler.openDB() db, err := handler.openDB()
require.NoError(t, err) require.NoError(t, err)
now := time.Now().Unix() now := time.Now().Unix()
stale := time.Now().Add(-keepOld - time.Minute).Unix() stale := time.Now().Add(-inUseGrace - time.Minute).Unix()
a := &Cache{Repo: "owner/repoA", Key: key, Version: version, Complete: true, CreatedAt: stale, UsedAt: stale, Size: 1} a := &Cache{Repo: "owner/repoA", Key: key, Version: version, Complete: true, CreatedAt: stale, UsedAt: stale, Size: 1}
b := &Cache{Repo: "owner/repoB", Key: key, Version: version, Complete: true, CreatedAt: now, UsedAt: now, Size: 1} b := &Cache{Repo: "owner/repoB", Key: key, Version: version, Complete: true, CreatedAt: now, UsedAt: now, Size: 1}
require.NoError(t, insertCache(db, a)) require.NoError(t, insertCache(db, a))
@@ -1179,7 +1437,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) {
// register/revoke when the feature is off. // register/revoke when the feature is off.
func TestHandler_InternalAPI_Disabled(t *testing.T) { func TestHandler_InternalAPI_Disabled(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := StartHandler(dir, "", 0, "", nil) handler, err := StartHandler(Options{Dir: dir})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -1197,7 +1455,7 @@ func TestHandler_InternalAPI_Disabled(t *testing.T) {
func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) { func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
const secret = "internal-secret" const secret = "internal-secret"
handler, err := StartHandler(dir, "", 0, secret, nil) handler, err := StartHandler(Options{Dir: dir, InternalSecret: secret})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
+3 -2
View File
@@ -77,6 +77,7 @@ func (h *Handler) v2CreateCacheEntry(w http.ResponseWriter, r *http.Request, _ h
h.twirpError(w, r, twirpInternal, err) h.twirpError(w, r, twirpInternal, err)
return return
} else if existing != nil { } else if existing != nil {
h.touch(db, existing) // the client skips the upload, so this is the only sign the entry is still in use
h.twirpNotOK(w, r) h.twirpNotOK(w, r)
return return
} }
@@ -97,7 +98,7 @@ func (h *Handler) v2CreateCacheEntry(w http.ResponseWriter, r *http.Request, _ h
h.responseJSON(w, r, http.StatusOK, map[string]any{ h.responseJSON(w, r, http.StatusOK, map[string]any{
"ok": true, "ok": true,
"signed_upload_url": h.signedURL(blobPath, blobUploadPurpose, cache.ID, time.Now().Add(blobUploadURLTTL)), "signed_upload_url": h.signedURL(cred, blobPath, blobUploadPurpose, cache.ID, time.Now().Add(blobUploadURLTTL)),
}) })
} }
@@ -168,7 +169,7 @@ func (h *Handler) v2GetCacheEntryDownloadURL(w http.ResponseWriter, r *http.Requ
h.responseJSON(w, r, http.StatusOK, map[string]any{ h.responseJSON(w, r, http.StatusOK, map[string]any{
"ok": true, "ok": true,
"signed_download_url": h.signedArtifactURL(cache.ID, time.Now().Add(artifactURLTTL)), "signed_download_url": h.signedArtifactURL(cred, cache.ID, time.Now().Add(artifactURLTTL)),
"matched_key": cache.Key, "matched_key": cache.Key,
}) })
} }
+16 -14
View File
@@ -10,8 +10,8 @@ import (
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
"path/filepath"
"strconv" "strconv"
"strings"
"testing" "testing"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -66,16 +66,6 @@ func getURL(t *testing.T, url string) []byte {
return body return body
} }
func startTestHandler(t *testing.T) *Handler {
t.Helper()
handler, err := StartHandler(filepath.Join(t.TempDir(), "artifactcache"), "127.0.0.1", 0, "", nil)
require.NoError(t, err)
t.Cleanup(func() { _ = handler.Close() })
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
return handler
}
// saveV2 runs the reserve/upload/finalize sequence and returns the finalize response along // saveV2 runs the reserve/upload/finalize sequence and returns the finalize response along
// with the upload URL it used. // with the upload URL it used.
func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) (finalized map[string]any, uploadURL string) { func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) (finalized map[string]any, uploadURL string) {
@@ -97,7 +87,7 @@ func saveV2(t *testing.T, handler *Handler, key, version string, content []byte)
// URLs it is handed: unsigned requests are refused, an upload URL cannot be replayed to read // URLs it is handed: unsigned requests are refused, an upload URL cannot be replayed to read
// or to replace a finalized entry. // or to replace a finalized entry.
func TestCacheServiceV2RoundTrip(t *testing.T) { func TestCacheServiceV2RoundTrip(t *testing.T) {
handler := startTestHandler(t) handler := newTestHandler(t, Policy{})
content := []byte("the cached archive") content := []byte("the cached archive")
unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath) unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath)
@@ -126,7 +116,7 @@ func TestCacheServiceV2RoundTrip(t *testing.T) {
// A large archive is staged as blocks and only put in order by the final block list, so // A large archive is staged as blocks and only put in order by the final block list, so
// blocks that arrive out of order must still be assembled the way the client asked. // blocks that arrive out of order must still be assembled the way the client asked.
func TestCacheServiceV2BlockUpload(t *testing.T) { func TestCacheServiceV2BlockUpload(t *testing.T) {
handler := startTestHandler(t) handler := newTestHandler(t, Policy{})
created := v2Call(t, handler, testClient, "CreateCacheEntry", map[string]any{"key": "blocks", "version": "v1"}) created := v2Call(t, handler, testClient, "CreateCacheEntry", map[string]any{"key": "blocks", "version": "v1"})
uploadURL, _ := created["signed_upload_url"].(string) uploadURL, _ := created["signed_upload_url"].(string)
@@ -163,7 +153,7 @@ func TestCacheServiceV2BlockUpload(t *testing.T) {
} }
func TestCacheServiceV2Lookups(t *testing.T) { func TestCacheServiceV2Lookups(t *testing.T) {
handler := startTestHandler(t) handler := newTestHandler(t, Policy{})
saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x")) saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x"))
require.Equal(t, true, saved["ok"]) require.Equal(t, true, saved["ok"])
@@ -198,6 +188,18 @@ func TestCacheServiceV2Lookups(t *testing.T) {
assert.NotEmpty(t, reserved["signed_upload_url"]) assert.NotEmpty(t, reserved["signed_upload_url"])
}) })
t.Run("a proxied job is handed the address its runner registered", func(t *testing.T) {
const proxy = "https://cache.example.invalid"
handler.RegisterJob("proxied", JobCredential{Repo: testRepo, PublicURL: proxy + "/"})
client := &http.Client{Transport: &bearerTransport{token: "proxied"}}
created := v2Call(t, handler, client, "CreateCacheEntry", map[string]any{"key": "proxied-key", "version": "v1"})
assert.True(t, strings.HasPrefix(created["signed_upload_url"].(string), proxy+blobPath+"/"))
got := v2Call(t, handler, client, "GetCacheEntryDownloadURL", map[string]any{"key": "deps-abc", "version": "v1"})
assert.True(t, strings.HasPrefix(got["signed_download_url"].(string), proxy+apiPath+"/artifacts/"))
})
t.Run("finalizing without a reservation is not ok", func(t *testing.T) { t.Run("finalizing without a reservation is not ok", func(t *testing.T) {
got := v2Call(t, handler, testClient, "FinalizeCacheEntryUpload", map[string]any{ got := v2Call(t, handler, testClient, "FinalizeCacheEntryUpload", map[string]any{
"key": "never-reserved", "version": "v1", "size_bytes": 1, "key": "never-reserved", "version": "v1", "size_bytes": 1,
+1 -1
View File
@@ -24,7 +24,7 @@ func TestFrontResultsService(t *testing.T) {
})) }))
defer gitea.Close() defer gitea.Close()
handler, err := StartHandler(t.TempDir(), "127.0.0.1", 0, "", nil) handler, err := StartHandler(Options{Dir: t.TempDir(), OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
const token = "forward-token" const token = "forward-token"
+7 -3
View File
@@ -143,9 +143,13 @@ func (s *Storage) Serve(w http.ResponseWriter, r *http.Request, id uint64) {
http.ServeFile(w, r, name) http.ServeFile(w, r, name)
} }
func (s *Storage) Remove(id uint64) { // Remove deletes an entry's blob and any staged parts. It reports failure so the caller can
_ = os.Remove(s.filename(id)) // keep the entry and retry, rather than dropping the only reference to bytes on disk.
_ = os.RemoveAll(s.tempDir(id)) func (s *Storage) Remove(id uint64) error {
if err := os.Remove(s.filename(id)); err != nil && !os.IsNotExist(err) {
return err
}
return os.RemoveAll(s.tempDir(id))
} }
func (s *Storage) filename(id uint64) string { func (s *Storage) filename(id uint64) string {
+15 -2
View File
@@ -345,6 +345,16 @@ func gitOptions(token string) (fetchOptions git.FetchOptions, pullOptions git.Pu
return fetchOptions, pullOptions return fetchOptions, pullOptions
} }
// staleRefreshErr reports why a failed refresh must abort: the resolve and
// checkout that follow are local and succeed on a cancelled context, which
// would hand back the cached revision as if it were fresh.
func staleRefreshErr(ctx context.Context, err error) error {
if err == nil || errors.Is(err, git.NoErrAlreadyUpToDate) {
return nil
}
return ctx.Err()
}
// NewGitCloneExecutor creates an executor to clone git repos // NewGitCloneExecutor creates an executor to clone git repos
func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
return func(ctx context.Context) error { return func(ctx context.Context) error {
@@ -385,7 +395,7 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
} }
if !isOfflineMode { if !isOfflineMode {
err = r.Fetch(&fetchOptions) err = r.FetchContext(ctx, &fetchOptions)
if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) { if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
return err return err
} }
@@ -454,9 +464,12 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
switch { switch {
case !isOfflineMode && !shallow: case !isOfflineMode && !shallow:
// In shallow mode the depth-limited fetch above already advanced the ref. // In shallow mode the depth-limited fetch above already advanced the ref.
if err = w.Pull(&pullOptions); err != nil && err != git.NoErrAlreadyUpToDate { if err = w.PullContext(ctx, &pullOptions); err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
logger.Debugf("Unable to pull %s: %v", refName, err) logger.Debugf("Unable to pull %s: %v", refName, err)
} }
if err := staleRefreshErr(ctx, err); err != nil {
return err
}
case isOfflineMode && reused: case isOfflineMode && reused:
reusedMsg = " (reused in offline mode)" reusedMsg = " (reused in offline mode)"
} }
+58
View File
@@ -6,18 +6,24 @@ package git
import ( import (
"context" "context"
"errors"
"fmt" "fmt"
"net/http"
"net/http/httptest"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
"strconv" "strconv"
"strings" "strings"
"sync"
"syscall" "syscall"
"testing" "testing"
"time" "time"
"gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common"
gogit "github.com/go-git/go-git/v5"
gogitconfig "github.com/go-git/go-git/v5/config"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
logrustest "github.com/sirupsen/logrus/hooks/test" logrustest "github.com/sirupsen/logrus/hooks/test"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
@@ -610,3 +616,55 @@ func TestAcquireCloneLock(t *testing.T) {
} }
}) })
} }
// An unresponsive remote must not pin a job: the refresh has to be interruptible.
func TestNewGitCloneExecutorFetchHonoursContext(t *testing.T) {
block := make(chan struct{})
reached := make(chan struct{})
var once sync.Once
server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
once.Do(func() { close(reached) })
<-block
}))
t.Cleanup(func() {
close(block)
server.Close()
})
dir := filepath.Join(t.TempDir(), "cached-action")
repo, err := gogit.PlainInit(dir, false)
require.NoError(t, err)
_, err = repo.CreateRemote(&gogitconfig.RemoteConfig{Name: "origin", URLs: []string{server.URL}})
require.NoError(t, err)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
done := make(chan error, 1)
go func() {
done <- NewGitCloneExecutor(NewGitCloneExecutorInput{URL: server.URL, Ref: "main", Dir: dir})(ctx)
}()
select {
case <-reached:
case <-time.After(10 * time.Second):
t.Fatal("the executor never reached the remote")
}
cancel()
select {
case err := <-done:
require.Error(t, err)
case <-time.After(10 * time.Second):
t.Fatal("fetch ignored context cancellation")
}
}
func TestStaleRefreshErr(t *testing.T) {
ctx, cancel := context.WithCancel(t.Context())
require.NoError(t, staleRefreshErr(ctx, errors.New("remote hung up")))
cancel()
require.ErrorIs(t, staleRefreshErr(ctx, errors.New("remote hung up")), context.Canceled)
require.NoError(t, staleRefreshErr(ctx, gogit.NoErrAlreadyUpToDate))
}
+13 -7
View File
@@ -88,14 +88,14 @@ func (cr *containerReference) connectToNetwork(name string, aliases []string) co
} }
} }
// supportsContainerImagePlatform returns true if the underlying Docker server // supportsContainerImagePlatform reports whether the Docker server API version
// API version is 1.41 and beyond // is 1.41 and beyond
func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) bool { func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) (bool, error) {
ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{}) ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{})
if err != nil { if err != nil {
common.Logger(ctx).Panicf("Failed to get Docker API Version: %s", err) return false, fmt.Errorf("get docker API version: %w", err)
} }
return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41") return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41"), nil
} }
func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor { func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor {
@@ -682,11 +682,17 @@ func (cr *containerReference) create(capAdd, capDrop []string) common.Executor {
} }
var platSpecs *specs.Platform var platSpecs *specs.Platform
if cr.input.Platform != "" && supportsContainerImagePlatform(ctx, cr.cli) { if cr.input.Platform != "" {
platSpecs, err = parsePlatform(cr.input.Platform) // Dropping the platform silently would build for the host arch.
supported, err := supportsContainerImagePlatform(ctx, cr.cli)
if err != nil { if err != nil {
return err return err
} }
if supported {
if platSpecs, err = parsePlatform(cr.input.Platform); err != nil {
return err
}
}
} }
hostConfig := &container.HostConfig{ hostConfig := &container.HostConfig{
+17
View File
@@ -79,6 +79,11 @@ type mockDockerClient struct {
mock.Mock mock.Mock
} }
func (m *mockDockerClient) ServerVersion(ctx context.Context, opts mobyclient.ServerVersionOptions) (mobyclient.ServerVersionResult, error) {
args := m.Called(ctx, opts)
return args.Get(0).(mobyclient.ServerVersionResult), args.Error(1)
}
func (m *mockDockerClient) ExecCreate(ctx context.Context, id string, opts mobyclient.ExecCreateOptions) (mobyclient.ExecCreateResult, error) { func (m *mockDockerClient) ExecCreate(ctx context.Context, id string, opts mobyclient.ExecCreateOptions) (mobyclient.ExecCreateResult, error) {
args := m.Called(ctx, id, opts) args := m.Called(ctx, id, opts)
return args.Get(0).(mobyclient.ExecCreateResult), args.Error(1) return args.Get(0).(mobyclient.ExecCreateResult), args.Error(1)
@@ -930,3 +935,15 @@ func TestMergeContainerConfigsVolumesReplaceRunnerMounts(t *testing.T) {
assert.Equal(t, []string{"/var/run/docker.sock:/var/run/docker.sock", "/host/tools:/opt/hostedtoolcache"}, hostConf.Binds) assert.Equal(t, []string{"/var/run/docker.sock:/var/run/docker.sock", "/host/tools:/opt/hostedtoolcache"}, hostConf.Binds)
assert.Empty(t, hostConf.Mounts) assert.Empty(t, hostConf.Mounts)
} }
// A dead daemon must fail the job, not panic through logrus and not silently
// drop the requested platform.
func TestSupportsContainerImagePlatformDaemonError(t *testing.T) {
cli := &mockDockerClient{}
cli.On("ServerVersion", mock.Anything, mock.Anything).
Return(mobyclient.ServerVersionResult{}, errors.New("cannot connect to the Docker daemon"))
supported, err := supportsContainerImagePlatform(t.Context(), cli)
require.ErrorContains(t, err, "cannot connect to the Docker daemon")
assert.False(t, supported)
}
+5
View File
@@ -167,6 +167,11 @@ func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actio
defer git.AcquireCloneLock(actionDir)() defer git.AcquireCloneLock(actionDir)()
if !rc.Config.NoActionPatch {
// A concurrent job's prepare resets this directory, so patch under the copy's lock.
patchActions(ctx, actionScriptPaths(filepath.Join(actionDir, actionPath), step.getActionModel()))
}
if err := removeGitIgnore(ctx, actionDir); err != nil { if err := removeGitIgnore(ctx, actionDir); err != nil {
return err return err
} }
@@ -12,7 +12,6 @@ import (
"strings" "strings"
"gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/act/common/git"
"gitea.dev/actionslib/pkg/model" "gitea.dev/actionslib/pkg/model"
) )
@@ -41,8 +40,7 @@ const (
cacheURLEnv = "ACTIONS_CACHE_URL" cacheURLEnv = "ACTIONS_CACHE_URL"
resultsURLEnv = "ACTIONS_RESULTS_URL" resultsURLEnv = "ACTIONS_RESULTS_URL"
// localhostHost is the suffix isGhes accepts; emptying the test is what opens the gate, // localhostHost is the suffix isGhes accepts.
// because every hostname ends with the empty string.
localhostHost = ".LOCALHOST" localhostHost = ".LOCALHOST"
// artifactRefusal is the only thing the gate guards in @actions/artifact, which is what makes // artifactRefusal is the only thing the gate guards in @actions/artifact, which is what makes
@@ -50,13 +48,6 @@ const (
// runner has not looked at, and is left alone. // runner has not looked at, and is left alone.
artifactRefusal = "GHESNotSupportedError" artifactRefusal = "GHESNotSupportedError"
// sidecarSuffix names the directory of untouched copies, a sibling of the action directory
// because that directory is copied wholesale into job containers.
sidecarSuffix = ".toolkit-patch"
// skipMarker in the sidecar means a patched bundle already failed once here.
skipMarker = "skip"
maxBundleSize = 64 << 20 maxBundleSize = 64 << 20
) )
@@ -101,156 +92,64 @@ func actionScriptPaths(dir string, action *model.Action) []string {
} }
var paths []string var paths []string
for _, script := range []string{action.Runs.Pre, action.Runs.Main, action.Runs.Post} { for _, script := range []string{action.Runs.Pre, action.Runs.Main, action.Runs.Post} {
if script != "" { if script == "" {
paths = append(paths, filepath.Join(dir, script)) continue
} }
path := filepath.Join(dir, script)
// `runs` is the action's own yaml, and a key pointing outside its directory is not ours.
if rel, err := filepath.Rel(dir, path); err != nil || strings.HasPrefix(rel, "..") {
continue
}
paths = append(paths, path)
} }
return paths return paths
} }
// patchToolkit edits the toolkit in an action's bundles, keeping each original beside them. Every // patchActions edits the toolkit in an action's bundles. The caller holds the action directory's
// failure is silent and leaves the bundle as it was, which costs the cache client the v2 API and // clone lock, which is what keeps another job's checkout from resetting them before the copy.
// an artifact action nothing at all. func patchActions(ctx context.Context, scripts []string) {
func patchToolkit(ctx context.Context, actionDir string, scripts []string) {
if len(scripts) == 0 {
return
}
if _, err := os.Stat(filepath.Join(sidecarDir(actionDir), skipMarker)); err == nil {
return
}
defer git.AcquireCloneLock(actionDir)()
for _, script := range scripts { for _, script := range scripts {
if err := patchBundle(script, originalFor(actionDir, script)); err != nil { switch patched, err := patchBundle(script); {
common.Logger(ctx).Debugf("actions toolkit: %s left unpatched: %v", filepath.Base(script), err) case err != nil:
common.Logger(ctx).Warnf("actions toolkit: %s left unpatched: %v", script, err)
case patched:
common.Logger(ctx).Debugf("actions toolkit: patched %s", script)
} }
} }
} }
// revertToolkit puts the originals back and stops this action being patched again, so the next job func patchBundle(script string) (bool, error) {
// runs it exactly as shipped. Called when a step failed with a patched bundle; it does not re-run
// the step, because a step's outputs and env-file writes are already recorded by then.
func revertToolkit(ctx context.Context, actionDir string, scripts []string) {
if len(scripts) == 0 {
return
}
if _, err := os.Stat(sidecarDir(actionDir)); err != nil {
return
}
defer git.AcquireCloneLock(actionDir)()
reverted := false
for _, script := range scripts {
original := originalFor(actionDir, script)
if !isPatchOf(original, script) {
continue
}
if err := os.Rename(original, script); err == nil {
reverted = true
}
}
if reverted {
_ = os.WriteFile(filepath.Join(sidecarDir(actionDir), skipMarker), nil, 0o600)
common.Logger(ctx).Warnf("actions toolkit: restored the original %s, it will not be patched again", filepath.Base(actionDir))
}
}
// sidecarDir holds an action's untouched bundles, and the marker that stops it being patched.
func sidecarDir(actionDir string) string {
return actionDir + sidecarSuffix
}
// originalFor is where a script's untouched copy lives, or "" for a script the action's own
// `runs` keys placed outside its directory, which is not this runner's to rewrite.
func originalFor(actionDir, script string) string {
rel, err := filepath.Rel(actionDir, script)
if err != nil || strings.HasPrefix(rel, "..") {
return ""
}
return filepath.Join(sidecarDir(actionDir), rel)
}
// patchBundle rewrites one entrypoint in place. The untouched copy kept beside it is what marks
// the bundle as already patched.
func patchBundle(script, original string) error {
if original == "" {
return nil
}
if _, err := os.Stat(original); err == nil {
if isPatchOf(original, script) {
return nil
}
// The action's ref moved and git checked the new bundle out over the patched one, so
// the pair no longer belongs together. Patch afresh rather than keep an original that
// would restore an older version of the action.
if err := os.Remove(original); err != nil {
return err
}
}
info, err := os.Stat(script) info, err := os.Stat(script)
if err != nil { if err != nil {
return err return false, err
} }
if info.Size() > maxBundleSize { if info.Size() > maxBundleSize {
return nil return false, nil
} }
data, err := os.ReadFile(script) data, err := os.ReadFile(script)
if err != nil { if err != nil {
return err return false, err
} }
patched, ok := patchedBundle(data) patched, ok := patchedBundle(data)
if !ok { if !ok {
return nil return false, nil
} }
if err := os.MkdirAll(filepath.Dir(original), 0o755); err != nil { // No atomic write needed: every prepare checks the action out and hard resets it.
return err return true, os.WriteFile(script, patched, info.Mode().Perm())
}
// The copy is taken before the bundle is replaced, so a write that fails part way can put the
// action back as it was. A crash needs no handling: the clone executor checks the action out
// and hard resets it on every prepare, so a half-written bundle never outlives the job.
if err := os.WriteFile(original, data, info.Mode().Perm()); err != nil {
return err
}
if err := os.WriteFile(script, patched, info.Mode().Perm()); err != nil {
_ = os.Rename(original, script)
return err
}
return nil
}
// isPatchOf reports whether script is exactly what patching original produced. It is what proves
// the two still belong together: an action whose ref moved is checked out over the patched bundle,
// leaving an original that would restore the version before the move.
func isPatchOf(original, script string) bool {
data, err := os.ReadFile(original)
if err != nil {
return false
}
current, err := os.ReadFile(script)
if err != nil {
return false
}
patched, ok := patchedBundle(data)
return ok && bytes.Equal(patched, current)
} }
// patchedBundle opens the GHES gate, and where the cache toolkit is present, points the cache // patchedBundle opens the GHES gate, and where the cache toolkit is present, points the cache
// service at the cache server. A bundle this runner cannot account for comes back untouched. // service at the cache server. A bundle this runner cannot account for comes back untouched.
func patchedBundle(data []byte) ([]byte, bool) { func patchedBundle(data []byte) ([]byte, bool) {
if !localhostTest.Match(data) { // Literals before regex: most bundles carry neither toolkit and stop here. The artifact gate
// guards a refusal with no URL to move, so it opens alone; the cache gate opens only with its
// service URL, since a bundle whose getter this cannot find is better left on v1.
artifact := bytes.Contains(data, []byte(artifactRefusal))
cache := bytes.Contains(data, []byte(CacheServiceV2Env)) && serviceURLBranches.Match(data)
if !artifact && !cache {
return data, false return data, false
} }
switch { if !localhostTest.Match(data) {
case bytes.Contains(data, []byte(CacheServiceV2Env)):
// The cache toolkit: both edits or neither, because choosing v2 without redirecting the
// URL would send the client to a results URL that serves no cache service.
if !serviceURLBranches.Match(data) {
return data, false
}
case bytes.Contains(data, []byte(artifactRefusal)):
// The artifact toolkit, where the gate is a plain refusal and there is no URL to move:
// artifacts already go to Gitea, which implements that service.
default:
return data, false return data, false
} }
@@ -259,5 +158,8 @@ func patchedBundle(data []byte) ([]byte, bool) {
// quoting survives and the result stays valid even inside a string literal. // quoting survives and the result stays valid even inside a string literal.
return bytes.Replace(test, []byte(localhostHost), nil, 1) return bytes.Replace(test, []byte(localhostHost), nil, 1)
}) })
return serviceURLBranches.ReplaceAll(opened, cacheURLFirst), true if cache {
opened = serviceURLBranches.ReplaceAll(opened, cacheURLFirst)
}
return opened, true
} }
@@ -108,17 +108,15 @@ func runActionEntrypoint(t *testing.T, script string, env jobEnv, inputs map[str
return string(out) return string(out)
} }
// patchedAction downloads one entrypoint and patches it exactly as a downloaded action would be, // patchedAction downloads one entrypoint and patches it exactly as a downloaded action would be.
// keeping the untouched original in the sidecar beside it.
func patchedAction(t *testing.T, repo, ref, entrypoint string) string { func patchedAction(t *testing.T, repo, ref, entrypoint string) string {
t.Helper() t.Helper()
body, err := os.ReadFile(bundleFromGitHub(t, repo, ref, entrypoint)) body, err := os.ReadFile(bundleFromGitHub(t, repo, ref, entrypoint))
require.NoError(t, err) require.NoError(t, err)
dir := tempDirPath(t) script := filepath.Join(tempDirPath(t), filepath.Base(entrypoint))
script := filepath.Join(dir, filepath.Base(entrypoint))
require.NoError(t, os.WriteFile(script, body, 0o600)) require.NoError(t, os.WriteFile(script, body, 0o600))
patchToolkit(t.Context(), dir, []string{script}) patchActions(t.Context(), []string{script})
return script return script
} }
@@ -144,7 +142,7 @@ func TestCacheServiceV2EndToEnd(t *testing.T) {
restore := patchedAction(t, "actions/cache", actionsCacheRef, "dist/restore/index.js") restore := patchedAction(t, "actions/cache", actionsCacheRef, "dist/restore/index.js")
save := patchedAction(t, "actions/cache", actionsCacheRef, "dist/save/index.js") save := patchedAction(t, "actions/cache", actionsCacheRef, "dist/save/index.js")
handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil) handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
t.Cleanup(func() { _ = handler.Close() }) t.Cleanup(func() { _ = handler.Close() })
const token, repo = "e2e-runtime-token", "testuser/testrepo" const token, repo = "e2e-runtime-token", "testuser/testrepo"
@@ -184,7 +182,7 @@ func TestCacheServiceV2EndToEnd(t *testing.T) {
// cache server on its own address. That is what a runner without a results service of its own // cache server on its own address. That is what a runner without a results service of its own
// leaves its jobs with, so it has to round trip too. // leaves its jobs with, so it has to round trip too.
env.workspace = tempDirPath(t) env.workspace = tempDirPath(t)
v1 := runActionEntrypoint(t, filepath.Join(sidecarDir(filepath.Dir(restore)), "index.js"), env, inputs) v1 := runActionEntrypoint(t, bundleFromGitHub(t, "actions/cache", actionsCacheRef, "dist/restore/index.js"), env, inputs)
require.Contains(t, v1, "Cache service version: v1") require.Contains(t, v1, "Cache service version: v1")
require.Contains(t, v1, "Cache restored from key: "+key) require.Contains(t, v1, "Cache restored from key: "+key)
} }
@@ -194,7 +192,7 @@ func TestCacheServiceV2EndToEnd(t *testing.T) {
// anchored on that distance. One entrypoint from each of the families that bundle the cache // anchored on that distance. One entrypoint from each of the families that bundle the cache
// toolkit, patched but not run, is what keeps a future release from quietly matching only one of // toolkit, patched but not run, is what keeps a future release from quietly matching only one of
// the two shapes and leaving every cache on v1. // the two shapes and leaving every cache on v1.
func TestToolkitPatchAcrossActions(t *testing.T) { func TestPatchedBundleAcrossActions(t *testing.T) {
for _, tc := range []struct { for _, tc := range []struct {
repo, ref, path string repo, ref, path string
wantPatched bool wantPatched bool
@@ -291,7 +289,7 @@ func TestUploadArtifactThroughTheResultsService(t *testing.T) {
})) }))
defer gitea.Close() defer gitea.Close()
handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil) handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
t.Cleanup(func() { _ = handler.Close() }) t.Cleanup(func() { _ = handler.Close() })
// The artifact client decodes the runtime token for the run ids it puts in its requests, where // The artifact client decodes the runtime token for the run ids it puts in its requests, where
@@ -345,7 +343,7 @@ func TestSetupActionFindsTheCacheService(t *testing.T) {
setup := patchedAction(t, "actions/setup-node", "v7.0.0", "dist/setup/index.js") setup := patchedAction(t, "actions/setup-node", "v7.0.0", "dist/setup/index.js")
handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil) handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
t.Cleanup(func() { _ = handler.Close() }) t.Cleanup(func() { _ = handler.Close() })
const token = "setup-runtime-token" const token = "setup-runtime-token"
@@ -5,7 +5,6 @@ package runner
import ( import (
"context" "context"
"errors"
"os" "os"
"os/exec" "os/exec"
"path/filepath" "path/filepath"
@@ -14,6 +13,7 @@ import (
"gitea.dev/actionslib/pkg/model" "gitea.dev/actionslib/pkg/model"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -172,51 +172,42 @@ func TestPatchedBundleSurvivesInsideAStringLiteral(t *testing.T) {
require.NoError(t, err, "%s", checked) require.NoError(t, err, "%s", checked)
} }
func TestPatchBundleKeepsTheOriginal(t *testing.T) { // An action with a pre step is copied, and so patched, twice.
dir, script := bundleFile(t, gateTSC) func TestPatchBundleIsIdempotent(t *testing.T) {
original := originalFor(dir, script) script := bundleFile(t, gateTSC)
require.NoError(t, patchBundle(script, original)) done, err := patchBundle(script)
require.NoError(t, err)
require.True(t, done)
patched, err := os.ReadFile(script) patched, err := os.ReadFile(script)
require.NoError(t, err) require.NoError(t, err)
assert.True(t, gateOpened(string(patched))) require.True(t, gateOpened(string(patched)))
kept, err := os.ReadFile(original) done, err = patchBundle(script)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, gateTSC, string(kept), "the untouched bundle is kept outside the action tree") assert.False(t, done, "a patched bundle is not patched again")
assert.NotContains(t, original, dir+string(filepath.Separator), "originals must not ship into job containers")
// Patching again must not stack, and must not overwrite the kept original.
require.NoError(t, patchBundle(script, original))
again, err := os.ReadFile(script) again, err := os.ReadFile(script)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, string(patched), string(again)) assert.Equal(t, string(patched), string(again))
kept, err = os.ReadFile(original)
require.NoError(t, err)
assert.Equal(t, gateTSC, string(kept))
} }
// A bundle with nothing to patch is left exactly as it was, with no original kept beside it.
func TestPatchBundleLeavesOtherActionsAlone(t *testing.T) { func TestPatchBundleLeavesOtherActionsAlone(t *testing.T) {
dir, script := bundleFile(t, `console.log("checkout")`) script := bundleFile(t, `console.log("checkout")`)
original := originalFor(dir, script)
require.NoError(t, patchBundle(script, original)) done, err := patchBundle(script)
require.NoError(t, err)
assert.False(t, done)
body, err := os.ReadFile(script) body, err := os.ReadFile(script)
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, `console.log("checkout")`, string(body)) assert.Equal(t, `console.log("checkout")`, string(body))
_, err = os.Stat(original)
assert.True(t, os.IsNotExist(err), "no original is kept for a bundle that was not patched")
} }
// bundleFile writes one entrypoint into a fresh action directory. func bundleFile(t *testing.T, body string) string {
func bundleFile(t *testing.T, body string) (dir, script string) {
t.Helper() t.Helper()
dir = t.TempDir() script := filepath.Join(t.TempDir(), "index.js")
script = filepath.Join(dir, "index.js")
require.NoError(t, os.WriteFile(script, []byte(body), 0o600)) require.NoError(t, os.WriteFile(script, []byte(body), 0o600))
return dir, script return script
} }
func TestActionScriptPaths(t *testing.T) { func TestActionScriptPaths(t *testing.T) {
@@ -226,101 +217,50 @@ func TestActionScriptPaths(t *testing.T) {
// Only a node action has a bundle to patch. // Only a node action has a bundle to patch.
assert.Nil(t, actionScriptPaths("/a", &model.Action{Runs: model.ActionRuns{Using: "docker", Image: "alpine"}})) assert.Nil(t, actionScriptPaths("/a", &model.Action{Runs: model.ActionRuns{Using: "docker", Image: "alpine"}}))
assert.Nil(t, actionScriptPaths("/a", nil)) assert.Nil(t, actionScriptPaths("/a", nil))
// An action naming a file outside its own directory does not get it rewritten.
escaping := &model.Action{Runs: model.ActionRuns{Using: "node20", Main: "../../elsewhere/index.js"}}
assert.Nil(t, actionScriptPaths("/a", escaping))
} }
// A step that fails with a patched bundle gets the untouched bundle back, and the action is not // The bundle has to be patched whatever state the shared action directory is in, because a
// patched again, so later jobs run it exactly as its author shipped it. // concurrent job's prepare checks the action out again and resets it.
func TestRevertToolkit(t *testing.T) { func TestPatchActionsAtTheContainerCopy(t *testing.T) {
dir, script := bundleFile(t, gateTSC) copiedBundle := func(t *testing.T, noPatch bool) string {
scripts := []string{script}
patchToolkit(t.Context(), dir, scripts)
body, err := os.ReadFile(script)
require.NoError(t, err)
require.True(t, gateOpened(string(body)), "precondition: the bundle is patched")
revertToolkit(t.Context(), dir, scripts)
body, err = os.ReadFile(script)
require.NoError(t, err)
assert.Equal(t, gateTSC, string(body), "the original bundle is back")
// The skip marker survives, so the action stays unpatched from now on.
patchToolkit(t.Context(), dir, scripts)
body, err = os.ReadFile(script)
require.NoError(t, err)
assert.Equal(t, gateTSC, string(body), "a reverted action stays unpatched")
}
// An action whose ref moves is checked out over the patched bundle. The kept original then
// belongs to the version before the move, and must not be restored over the new one.
func TestPatchBundleAfterTheActionMoved(t *testing.T) {
dir, script := bundleFile(t, gateTSC)
original := originalFor(dir, script)
scripts := []string{script}
patchToolkit(t.Context(), dir, scripts)
require.NoError(t, os.WriteFile(script, []byte(gateWebpack), 0o600)) // the new version lands
// Reverting must not roll the action back to the version the original came from.
revertToolkit(t.Context(), dir, scripts)
body, err := os.ReadFile(script)
require.NoError(t, err)
assert.Equal(t, gateWebpack, string(body))
// Nothing was reverted, so the action is not marked off either: the new version is patched
// in its own right, and keeps its own original.
require.NoFileExists(t, filepath.Join(sidecarDir(dir), skipMarker))
require.NoError(t, patchBundle(script, original))
body, err = os.ReadFile(script)
require.NoError(t, err)
assert.True(t, gateOpened(string(body)))
kept, err := os.ReadFile(original)
require.NoError(t, err)
assert.Equal(t, gateWebpack, string(kept))
}
// The wiring: a step patches its own bundles only when the runner serves the v2 API, and a step
// that fails gets them back. The action's path inside its repository is part of where they live.
func TestStepActionRemoteToolkitPatch(t *testing.T) {
newStep := func(t *testing.T, patch bool) (*stepActionRemote, string) {
t.Helper() t.Helper()
cm := &containerMock{}
sar := &stepActionRemote{ sar := &stepActionRemote{
Step: &model.Step{Uses: "owner/repo/sub@v1"}, Step: &model.Step{Uses: "owner/repo/sub@v1"},
remoteAction: &remoteAction{Org: "owner", Repo: "repo", Path: "sub", Ref: "v1"}, remoteAction: &remoteAction{Org: "owner", Repo: "repo", Path: "sub", Ref: "v1"},
action: &model.Action{Runs: model.ActionRuns{Using: "node20", Main: "index.js"}}, action: &model.Action{Runs: model.ActionRuns{Using: "node20", Main: "index.js"}},
RunContext: &RunContext{ RunContext: &RunContext{
Config: &Config{ActionCacheDir: t.TempDir(), PatchToolkit: patch}, Config: &Config{ActionCacheDir: t.TempDir(), NoActionPatch: noPatch},
JobContainer: cm,
}, },
} }
script := filepath.Join(sar.actionDir(), "sub", "index.js") script := filepath.Join(sar.actionDir(), "sub", "index.js")
require.NoError(t, os.MkdirAll(filepath.Dir(script), 0o755)) require.NoError(t, os.MkdirAll(filepath.Dir(script), 0o755))
require.NoError(t, os.WriteFile(script, []byte(gateTSC), 0o600)) require.NoError(t, os.WriteFile(script, []byte(gateTSC), 0o600))
return sar, script
var copied string
cm.On("CopyDir", mock.Anything, mock.Anything, mock.Anything).Return(func(context.Context) error {
body, err := os.ReadFile(script)
require.NoError(t, err)
copied = string(body)
return nil
})
require.NoError(t, maybeCopyToActionDir(t.Context(), sar, sar.actionDir(), "sub", "/var/run/act/actions/repo/sub"))
return copied
} }
t.Run("left alone when the runner does not patch", func(t *testing.T) { t.Run("patched on its way in", func(t *testing.T) {
sar, script := newStep(t, false) assert.True(t, gateOpened(copiedBundle(t, false)))
require.NoError(t, sar.patchActionToolkit(t.Context()))
body, err := os.ReadFile(script)
require.NoError(t, err)
assert.Equal(t, gateTSC, string(body))
}) })
t.Run("patched, and put back when the step fails", func(t *testing.T) { // The escape hatch, for an action the edit breaks: the artifact actions refuse again, and the
sar, script := newStep(t, true) // cache client keeps to v1.
require.NoError(t, sar.patchActionToolkit(t.Context())) t.Run("as shipped when the runner is told not to patch", func(t *testing.T) {
assert.Equal(t, gateTSC, copiedBundle(t, true))
body, err := os.ReadFile(script)
require.NoError(t, err)
require.True(t, gateOpened(string(body)))
failed := errors.New("the step failed")
require.ErrorIs(t, sar.revertToolkitOnFailure(func(context.Context) error { return failed })(t.Context()), failed)
body, err = os.ReadFile(script)
require.NoError(t, err)
assert.Equal(t, gateTSC, string(body))
}) })
} }
+15 -4
View File
@@ -229,14 +229,19 @@ func (rc *RunContext) containerDaemonSocket() string {
return rc.Config.ContainerDaemonSocket return rc.Config.ContainerDaemonSocket
} }
const sharedToolCacheVolume = "act-toolcache" // mounted only when the tool cache is shared
// validVolumes returns the volumes allowed on this job's containers: the configured base // validVolumes returns the volumes allowed on this job's containers: the configured base
// plus the volumes the runner mounts automatically. It derives a fresh slice every call and // plus the volumes the runner mounts automatically. It derives a fresh slice every call and
// never mutates the shared Config (see containerDaemonSocket). // never mutates the shared Config (see containerDaemonSocket).
func (rc *RunContext) validVolumes() []string { func (rc *RunContext) validVolumes() []string {
name := rc.jobContainerName() name := rc.jobContainerName()
volumes := slices.Clone(rc.Config.ValidVolumes) volumes := slices.Clone(rc.Config.ValidVolumes)
if rc.Config.SharedToolCache {
volumes = append(volumes, sharedToolCacheVolume)
}
// TODO: add a new configuration to control whether the docker daemon can be mounted // TODO: add a new configuration to control whether the docker daemon can be mounted
return append(volumes, "act-toolcache", name, name+"-env", return append(volumes, name, name+"-env",
getDockerDaemonSocketMountPath(rc.containerDaemonSocket())) getDockerDaemonSocketMountPath(rc.containerDaemonSocket()))
} }
@@ -309,8 +314,10 @@ func (rc *RunContext) GetBindsAndMounts() ([]string, map[string]string) {
if daemonSocket := rc.containerDaemonSocket(); daemonSocket != "-" && !claimed["/var/run/docker.sock"] { if daemonSocket := rc.containerDaemonSocket(); daemonSocket != "-" && !claimed["/var/run/docker.sock"] {
binds = append(binds, getDockerDaemonSocketMountPath(daemonSocket)+":/var/run/docker.sock") binds = append(binds, getDockerDaemonSocketMountPath(daemonSocket)+":/var/run/docker.sock")
} }
if toolCache := rc.toolCache(container.DefaultToolCache); !claimed[toolCache] { if rc.Config.SharedToolCache {
mounts["act-toolcache"] = toolCache if toolCache := rc.toolCache(container.DefaultToolCache); !claimed[toolCache] {
mounts[sharedToolCacheVolume] = toolCache
}
} }
mounts[name+"-env"] = ext.GetActPath() // runner-internal, never overridable mounts[name+"-env"] = ext.GetActPath() // runner-internal, never overridable
@@ -360,7 +367,11 @@ func (rc *RunContext) startHostEnvironment() common.Executor {
if err := os.MkdirAll(runnerTmp, 0o777); err != nil { if err := os.MkdirAll(runnerTmp, 0o777); err != nil {
return err return err
} }
toolCache := rc.toolCache(filepath.Join(cacheDir, "tool_cache")) toolCacheParent := miscpath // per job, so cleanup removes it with the job
if rc.Config.SharedToolCache {
toolCacheParent = cacheDir
}
toolCache := rc.toolCache(filepath.Join(toolCacheParent, "tool_cache"))
if err := os.MkdirAll(toolCache, 0o777); err != nil { if err := os.MkdirAll(toolCache, 0o777); err != nil {
return err return err
} }
+24 -4
View File
@@ -502,7 +502,8 @@ func TestRunContext_GetBindsAndMounts(t *testing.T) {
}, },
}, },
Config: &Config{ Config: &Config{
BindWorkdir: false, BindWorkdir: false,
SharedToolCache: true, // so OverridesToolCache has a mount to displace
}, },
} }
rc.Run.JobID = "job1" rc.Run.JobID = "job1"
@@ -543,25 +544,44 @@ func TestRunContext_GetBindsAndMounts(t *testing.T) {
}) })
} }
}) })
t.Run("ToolCacheMount", func(t *testing.T) {
rc := &RunContext{
Name: "TestRCName",
Run: &model.Run{Workflow: &model.Workflow{Name: "TestWorkflowName"}},
Config: &Config{},
}
_, gotmount := rc.GetBindsAndMounts()
assert.NotContains(t, gotmount, sharedToolCacheVolume)
rc.Config.SharedToolCache = true
_, gotmount = rc.GetBindsAndMounts()
assert.Equal(t, container.DefaultToolCache, gotmount[sharedToolCacheVolume])
})
} }
func TestRunContextValidVolumes(t *testing.T) { func TestRunContextValidVolumes(t *testing.T) {
rc := &RunContext{ rc := &RunContext{
Name: "job", Name: "job",
Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}}, Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}},
Config: &Config{ValidVolumes: []string{"my-vol", "/host/path"}}, Config: &Config{ValidVolumes: []string{"my-vol", "/host/path"}, SharedToolCache: true},
} }
name := rc.jobContainerName() name := rc.jobContainerName()
got := rc.validVolumes() got := rc.validVolumes()
// the configured volumes plus the four the runner mounts automatically // the configured volumes plus the ones the runner mounts automatically
assert.Subset(t, got, []string{"my-vol", "/host/path", "act-toolcache", name, name + "-env", "/var/run/docker.sock"}) assert.Subset(t, got, []string{"my-vol", "/host/path", sharedToolCacheVolume, name, name + "-env", "/var/run/docker.sock"})
// deriving the list must never mutate or grow the shared Config slice: parallel matrix // deriving the list must never mutate or grow the shared Config slice: parallel matrix
// combinations share one *Config, and the previous in-place append was a data race. // combinations share one *Config, and the previous in-place append was a data race.
assert.Equal(t, []string{"my-vol", "/host/path"}, rc.Config.ValidVolumes) assert.Equal(t, []string{"my-vol", "/host/path"}, rc.Config.ValidVolumes)
assert.Len(t, rc.validVolumes(), len(got), "repeated calls must be stable, not accumulate") assert.Len(t, rc.validVolumes(), len(got), "repeated calls must be stable, not accumulate")
// a job may mount it only while the runner does
rc.Config.SharedToolCache = false
assert.NotContains(t, rc.validVolumes(), sharedToolCacheVolume)
} }
func TestCleanupJobResourcesCleansServicesWithoutJobContainer(t *testing.T) { func TestCleanupJobResourcesCleansServicesWithoutJobContainer(t *testing.T) {
+2 -1
View File
@@ -74,7 +74,7 @@ type Config struct {
ContainerNetworkCreateOptions container.NewDockerNetworkCreateExecutorInput // the default network create options ContainerNetworkCreateOptions container.NewDockerNetworkCreateExecutorInput // the default network create options
ActionCache ActionCache // Use a custom ActionCache Implementation ActionCache ActionCache // Use a custom ActionCache Implementation
ProxyEnv map[string]string // the proxy variables the job runs with, also given to service containers and image builds ProxyEnv map[string]string // the proxy variables the job runs with, also given to service containers and image builds
PatchToolkit bool // edit the @actions toolkit bundled into an action so it works against Gitea, see toolkit_patch.go NoActionPatch bool // run actions exactly as published, applying no compatibility patches, see patch_actions.go
PresetGitHubContext *model.GithubContext // the preset github context, overrides some fields like DefaultBranch, Env, Secrets etc. PresetGitHubContext *model.GithubContext // the preset github context, overrides some fields like DefaultBranch, Env, Secrets etc.
EventJSON string // the content of JSON file to use for event.json in containers, overrides EventPath EventJSON string // the content of JSON file to use for event.json in containers, overrides EventPath
@@ -91,6 +91,7 @@ type Config struct {
PlatformPicker func(labels []string) string // platform picker, it will take precedence over Platforms if isn't nil PlatformPicker func(labels []string) string // platform picker, it will take precedence over Platforms if isn't nil
JobLoggerLevel *log.Level // the level of job logger JobLoggerLevel *log.Level // the level of job logger
ValidVolumes []string // only volumes (and bind mounts) in this slice can be mounted on the job container or service containers ValidVolumes []string // only volumes (and bind mounts) in this slice can be mounted on the job container or service containers
SharedToolCache bool // one tool cache for all jobs instead of one per job
InsecureSkipTLS bool // whether to skip verifying TLS certificate of the Gitea instance InsecureSkipTLS bool // whether to skip verifying TLS certificate of the Gitea instance
MaxParallel int // max parallel jobs to run across all workflows (0 = no limit, uses CPU count) MaxParallel int // max parallel jobs to run across all workflows (0 = no limit, uses CPU count)
AllocatePTY bool // allocate a pseudo-TTY for each step's process AllocatePTY bool // allocate a pseudo-TTY for each step's process
+3 -40
View File
@@ -180,9 +180,6 @@ func (sar *stepActionRemote) prepareActionExecutor() common.Executor {
sar.action = actionModel sar.action = actionModel
return err return err
}, },
// A stage of its own: it takes the same clone lock, and it has to land before
// runAction copies the action into the job container.
sar.patchActionToolkit,
)(ctx) )(ctx)
} }
} }
@@ -201,7 +198,7 @@ func (sar *stepActionRemote) pre() common.Executor {
return common.NewPipelineExecutor( return common.NewPipelineExecutor(
sar.prepareActionExecutor(), sar.prepareActionExecutor(),
runStepExecutor(sar, stepStagePre, sar.revertToolkitOnFailure(runPreStep(sar))).If(hasPreStep(sar)).If(shouldRunPreStep(sar))) runStepExecutor(sar, stepStagePre, runPreStep(sar)).If(hasPreStep(sar)).If(shouldRunPreStep(sar)))
} }
func (sar *stepActionRemote) main() common.Executor { func (sar *stepActionRemote) main() common.Executor {
@@ -223,47 +220,13 @@ func (sar *stepActionRemote) main() common.Executor {
return sar.RunContext.JobContainer.CopyDir(copyToPath, sar.RunContext.Config.Workdir+string(filepath.Separator)+".", sar.RunContext.Config.UseGitIgnore)(ctx) return sar.RunContext.JobContainer.CopyDir(copyToPath, sar.RunContext.Config.Workdir+string(filepath.Separator)+".", sar.RunContext.Config.UseGitIgnore)(ctx)
} }
actionDir := sar.actionDir() return sar.runAction(sar, sar.actionDir(), sar.remoteAction)(ctx)
return sar.revertToolkitOnFailure(sar.runAction(sar, actionDir, sar.remoteAction))(ctx)
}), }),
) )
} }
func (sar *stepActionRemote) post() common.Executor { func (sar *stepActionRemote) post() common.Executor {
return runStepExecutor(sar, stepStagePost, sar.revertToolkitOnFailure(runPostStep(sar))).If(hasPostStep(sar)).If(shouldRunPostStep(sar)) return runStepExecutor(sar, stepStagePost, runPostStep(sar)).If(hasPostStep(sar)).If(shouldRunPostStep(sar))
}
// toolkitBundles is the action directory and the entrypoints the toolkit may live in.
func (sar *stepActionRemote) toolkitBundles() (string, []string) {
if sar.remoteAction == nil {
return "", nil
}
dir := sar.actionDir()
return dir, actionScriptPaths(filepath.Join(dir, sar.remoteAction.Path), sar.action)
}
// patchActionToolkit edits the bundled toolkit so it works against Gitea: the artifact actions
// stop refusing, and the cache client keeps to the cache server whichever API version it picks.
func (sar *stepActionRemote) patchActionToolkit(ctx context.Context) error {
if sar.RunContext.Config.PatchToolkit {
dir, scripts := sar.toolkitBundles()
patchToolkit(ctx, dir, scripts)
}
return nil
}
// revertToolkitOnFailure restores the untouched bundles when the action fails, so a later job
// runs it as shipped rather than repeating a failure the patch may have caused.
func (sar *stepActionRemote) revertToolkitOnFailure(exec common.Executor) common.Executor {
return func(ctx context.Context) error {
err := exec(ctx)
if err != nil {
dir, scripts := sar.toolkitBundles()
revertToolkit(ctx, dir, scripts)
}
return err
}
} }
func (sar *stepActionRemote) actionDir() string { func (sar *stepActionRemote) actionDir() string {
+2
View File
@@ -11,6 +11,8 @@ Each example persists **two** things, and it is worth knowing which is which:
- `/data` is the runner's working directory. It holds the `.runner` registration file and, optionally, the config file — so the runner re-attaches to the server instead of registering again. - `/data` is the runner's working directory. It holds the `.runner` registration file and, optionally, the config file — so the runner re-attaches to the server instead of registering again.
- The Docker daemon's data root holds the images pulled for jobs (`/var/lib/docker` for the dind sidecar, `/home/rootless/.local/share/docker` for `dind-rootless`). It is *not* under `/data`. If you drop this volume, the examples still work, but the image cache is discarded whenever the pod is recreated and every job re-pulls its images. - The Docker daemon's data root holds the images pulled for jobs (`/var/lib/docker` for the dind sidecar, `/home/rootless/.local/share/docker` for `dind-rootless`). It is *not* under `/data`. If you drop this volume, the examples still work, but the image cache is discarded whenever the pod is recreated and every job re-pulls its images.
- Kubernetes SIGKILLs a pod 30s after SIGTERM by default, long before a job finishes and reports its result, which leaves tasks the server can only reap as zombies. The manifests raise `terminationGracePeriodSeconds` to three hours, matching the systemd example and the `runner.timeout` job ceiling; set `runner.shutdown_timeout` below that so the runner drains jobs within the window rather than being killed mid-cleanup.
Files in this directory: Files in this directory:
- [`dind-docker.yaml`](dind-docker.yaml) - [`dind-docker.yaml`](dind-docker.yaml)
+1
View File
@@ -56,6 +56,7 @@ spec:
app: runner app: runner
spec: spec:
restartPolicy: Always restartPolicy: Always
terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README
volumes: volumes:
- name: docker-socket - name: docker-socket
emptyDir: {} emptyDir: {}
+38
View File
@@ -0,0 +1,38 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package kubernetes_test
import (
"os"
"path/filepath"
"regexp"
"strconv"
"strings"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
var gracePeriod = regexp.MustCompile(`terminationGracePeriodSeconds: (\d+)`)
// Without it Kubernetes SIGKILLs the pod 30s after SIGTERM, mid-job.
func TestManifestsSetTerminationGracePeriod(t *testing.T) {
files, err := filepath.Glob("*.yaml")
require.NoError(t, err)
require.NotEmpty(t, files)
for _, file := range files {
content, err := os.ReadFile(file)
require.NoError(t, err)
if !strings.Contains(string(content), "containers:") {
continue
}
match := gracePeriod.FindStringSubmatch(string(content))
require.NotNil(t, match, file)
seconds, err := strconv.Atoi(match[1])
require.NoError(t, err)
assert.GreaterOrEqual(t, seconds, 3600, file)
}
}
+1
View File
@@ -56,6 +56,7 @@ spec:
app: runner app: runner
spec: spec:
restartPolicy: Always restartPolicy: Always
terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README
volumes: volumes:
- name: runner-data - name: runner-data
persistentVolumeClaim: persistentVolumeClaim:
@@ -33,6 +33,7 @@ spec:
app: runner app: runner
spec: spec:
restartPolicy: Always restartPolicy: Always
terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README
volumes: volumes:
- name: docker-socket - name: docker-socket
emptyDir: {} emptyDir: {}
+3 -3
View File
@@ -12,8 +12,9 @@ require (
github.com/containerd/errdefs v1.0.0 github.com/containerd/errdefs v1.0.0
github.com/creack/pty v1.1.24 github.com/creack/pty v1.1.24
github.com/distribution/reference v0.6.0 github.com/distribution/reference v0.6.0
github.com/docker/cli v29.6.2+incompatible github.com/docker/cli v29.7.1+incompatible
github.com/docker/go-connections v0.8.1 github.com/docker/go-connections v0.8.1
github.com/docker/go-units v0.5.0
github.com/go-git/go-billy/v5 v5.9.1 github.com/go-git/go-billy/v5 v5.9.1
github.com/go-git/go-git/v5 v5.19.2 github.com/go-git/go-git/v5 v5.19.2
github.com/gobwas/glob v0.2.3 github.com/gobwas/glob v0.2.3
@@ -22,7 +23,7 @@ require (
github.com/julienschmidt/httprouter v1.3.0 github.com/julienschmidt/httprouter v1.3.0
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
github.com/mattn/go-isatty v0.0.24 github.com/mattn/go-isatty v0.0.24
github.com/moby/go-archive v0.2.1 github.com/moby/go-archive v0.3.2
github.com/moby/moby/api v1.55.0 github.com/moby/moby/api v1.55.0
github.com/moby/moby/client v0.5.1 github.com/moby/moby/client v0.5.1
github.com/moby/patternmatcher v0.6.1 github.com/moby/patternmatcher v0.6.1
@@ -61,7 +62,6 @@ require (
github.com/cyphar/filepath-securejoin v0.6.1 // indirect github.com/cyphar/filepath-securejoin v0.6.1 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect github.com/davecgh/go-spew v1.1.1 // indirect
github.com/docker/docker-credential-helpers v0.9.6 // indirect github.com/docker/docker-credential-helpers v0.9.6 // indirect
github.com/docker/go-units v0.5.0 // indirect
github.com/emirpasic/gods v1.18.1 // indirect github.com/emirpasic/gods v1.18.1 // indirect
github.com/fatih/color v1.19.0 // indirect github.com/fatih/color v1.19.0 // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect
+4 -4
View File
@@ -45,8 +45,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
github.com/docker/cli v29.6.2+incompatible h1:/bjePvcbbFTnRrMfWJBY7AjfICdsiLVgHn6LwTVOcqw= github.com/docker/cli v29.7.1+incompatible h1:ILZpP6B7fedIr6ANy824QkDp1WMJuouIq0O2SrBkB2w=
github.com/docker/cli v29.6.2+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8= github.com/docker/cli v29.7.1+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
github.com/docker/docker-credential-helpers v0.9.6 h1:cT2PbRPSlnMmNTfT2TDMXRyQ1KMWHG7xoTLBcn1ZNv0= github.com/docker/docker-credential-helpers v0.9.6 h1:cT2PbRPSlnMmNTfT2TDMXRyQ1KMWHG7xoTLBcn1ZNv0=
github.com/docker/docker-credential-helpers v0.9.6/go.mod h1:v1S+hepowrQXITkEfw6o4+BMbGot02wiKpzWhGUZK6c= github.com/docker/docker-credential-helpers v0.9.6/go.mod h1:v1S+hepowrQXITkEfw6o4+BMbGot02wiKpzWhGUZK6c=
github.com/docker/go-connections v0.8.1 h1:JibmG5hULs5qXSr/cp/w3Pw5fZuStt4MOHMUExb29/M= github.com/docker/go-connections v0.8.1 h1:JibmG5hULs5qXSr/cp/w3Pw5fZuStt4MOHMUExb29/M=
@@ -123,8 +123,8 @@ github.com/mattn/go-shellwords v1.0.12 h1:M2zGm7EW6UQJvDeQxo4T51eKPurbeFbe8WtebG
github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y= github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y=
github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
github.com/moby/go-archive v0.2.1 h1:fAa0wUS/ikZKyx7o/1fhUYmhZ7RgpthdeoDhJvunTLc= github.com/moby/go-archive v0.3.2 h1:x893kC3zRygv2C+k4Y9kMxYRPLCj4XEJB0srbAP06Hw=
github.com/moby/go-archive v0.2.1/go.mod h1:Npdv43fFqlhZW7Xo8fbm3ZMYFvAGNviUPqX21VERbcE= github.com/moby/go-archive v0.3.2/go.mod h1:Npdv43fFqlhZW7Xo8fbm3ZMYFvAGNviUPqX21VERbcE=
github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc= github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc=
github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs=
github.com/moby/moby/client v0.5.1 h1:tYNaJno4c0HXz12y5BiqEDy0rVTYkWzI26lGvnTMiJw= github.com/moby/moby/client v0.5.1 h1:tYNaJno4c0HXz12y5BiqEDy0rVTYkWzI26lGvnTMiJw=
+9 -7
View File
@@ -10,6 +10,7 @@ import (
"os/signal" "os/signal"
"gitea.com/gitea/runner/act/artifactcache" "gitea.com/gitea/runner/act/artifactcache"
"gitea.com/gitea/runner/internal/app/run"
"gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/config"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
@@ -52,13 +53,14 @@ func runCacheServer(configFile *string, cacheArgs *cacheServerArgs) func(cmd *co
if secret == "" { if secret == "" {
return errors.New("cache.external_secret (or cache.external_secret_file) must be set for cache-server; configure the same value on each runner that points at this server via cache.external_server") return errors.New("cache.external_secret (or cache.external_secret_file) must be set for cache-server; configure the same value on each runner that points at this server via cache.external_server")
} }
cacheHandler, err := artifactcache.StartHandler( cacheHandler, err := artifactcache.StartHandler(artifactcache.Options{
dir, Dir: dir,
host, OutboundIP: host,
port, Port: port,
secret, InternalSecret: secret,
log.StandardLogger().WithField("module", "cache_request"), Policy: run.CachePolicy(cfg),
) Logger: log.StandardLogger().WithField("module", "cache_request"),
})
if err != nil { if err != nil {
return err return err
} }
+1
View File
@@ -158,6 +158,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
cfg.Runner.Insecure, cfg.Runner.Insecure,
reg.UUID, reg.UUID,
reg.Token, reg.Token,
config.RequestTimeout,
) )
runner := run.NewRunner(cfg, reg, cli) runner := run.NewRunner(cfg, reg, cli)
+24 -4
View File
@@ -13,6 +13,7 @@ import (
"maps" "maps"
"os" "os"
"path/filepath" "path/filepath"
"slices"
"strconv" "strconv"
"strings" "strings"
"time" "time"
@@ -22,6 +23,7 @@ import (
"gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common"
"gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/app/run" "gitea.com/gitea/runner/internal/app/run"
"gitea.com/gitea/runner/internal/pkg/config"
"gitea.dev/actionslib/pkg/model" "gitea.dev/actionslib/pkg/model"
"github.com/joho/godotenv" "github.com/joho/godotenv"
@@ -67,6 +69,15 @@ type executeArgs struct {
cacheHandler *artifactcache.Handler cacheHandler *artifactcache.Handler
network string network string
githubInstance string githubInstance string
toolCacheMode string
}
// sharedToolCache reports whether mode mounts one tool cache for every job.
func sharedToolCache(mode string) (bool, error) {
if !slices.Contains(config.ToolCacheModes, mode) {
return false, fmt.Errorf("invalid --tool-cache-mode %q: must be one of %q", mode, config.ToolCacheModes)
}
return mode == config.ToolCacheModeShared, nil
} }
// WorkflowsPath returns path to workflow file(s) // WorkflowsPath returns path to workflow file(s)
@@ -374,7 +385,10 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
} }
// init a cache server // init a cache server
handler, err := artifactcache.StartHandler("", "", 0, "", log.StandardLogger().WithField("module", "cache_request")) handler, err := artifactcache.StartHandler(artifactcache.Options{
Policy: run.CachePolicy(&config.Config{Cache: config.DefaultCache()}),
Logger: log.StandardLogger().WithField("module", "cache_request"),
})
if err != nil { if err != nil {
return err return err
} }
@@ -423,11 +437,15 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
proxyEnv := run.JobProxyEnv(env, env["ACTIONS_CACHE_URL"], nil) proxyEnv := run.JobProxyEnv(env, env["ACTIONS_CACHE_URL"], nil)
maps.Copy(env, proxyEnv) maps.Copy(env, proxyEnv)
shared, err := sharedToolCache(execArgs.toolCacheMode)
if err != nil {
return err
}
// run the plan // run the plan
config := &runner.Config{ config := &runner.Config{
Workdir: execArgs.Workdir(), Workdir: execArgs.Workdir(),
BindWorkdir: false, BindWorkdir: false,
PatchToolkit: true, // the cache server started above is what the patch points at
ReuseContainers: false, ReuseContainers: false,
ForcePull: execArgs.forcePull, ForcePull: execArgs.forcePull,
ForceRebuild: execArgs.forceRebuild, ForceRebuild: execArgs.forceRebuild,
@@ -463,7 +481,8 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
PlatformPicker: func(_ []string) string { PlatformPicker: func(_ []string) string {
return execArgs.image return execArgs.image
}, },
ValidVolumes: []string{"**"}, // All volumes are allowed for `exec` command ValidVolumes: []string{"**"}, // All volumes are allowed for `exec` command
SharedToolCache: shared,
} }
config.Env["ACT_EXEC"] = "true" config.Env["ACT_EXEC"] = "true"
@@ -538,7 +557,8 @@ func loadExecCmd(ctx context.Context) *cobra.Command {
execCmd.PersistentFlags().BoolVarP(&execArg.noSkipCheckout, "no-skip-checkout", "", false, "Do not skip actions/checkout") execCmd.PersistentFlags().BoolVarP(&execArg.noSkipCheckout, "no-skip-checkout", "", false, "Do not skip actions/checkout")
execCmd.PersistentFlags().BoolVarP(&execArg.debug, "debug", "d", false, "enable debug log") execCmd.PersistentFlags().BoolVarP(&execArg.debug, "debug", "d", false, "enable debug log")
execCmd.PersistentFlags().BoolVarP(&execArg.dryrun, "dryrun", "n", false, "dryrun mode") execCmd.PersistentFlags().BoolVarP(&execArg.dryrun, "dryrun", "n", false, "dryrun mode")
execCmd.PersistentFlags().StringVarP(&execArg.image, "image", "i", "docker.gitea.com/runner-images:ubuntu-latest", "Docker image to use. Use \"-self-hosted\" to run directly on the host.") execCmd.PersistentFlags().StringVarP(&execArg.image, "image", "i", config.DefaultImage, "Docker image to use. Use \"-self-hosted\" to run directly on the host.")
execCmd.PersistentFlags().StringVarP(&execArg.toolCacheMode, "tool-cache-mode", "", config.ToolCacheModeNone, "What to mount at RUNNER_TOOL_CACHE: none, or shared to reuse one tool cache across runs")
execCmd.PersistentFlags().StringVarP(&execArg.network, "network", "", "", "Specify the network to which the container will connect") execCmd.PersistentFlags().StringVarP(&execArg.network, "network", "", "", "Specify the network to which the container will connect")
execCmd.PersistentFlags().StringVarP(&execArg.githubInstance, "gitea-instance", "", "", "Gitea instance to use.") execCmd.PersistentFlags().StringVarP(&execArg.githubInstance, "gitea-instance", "", "", "Gitea instance to use.")
+15
View File
@@ -12,6 +12,8 @@ import (
"strings" "strings"
"testing" "testing"
"gitea.com/gitea/runner/internal/pkg/config"
"gitea.dev/actionslib/pkg/model" "gitea.dev/actionslib/pkg/model"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"go.yaml.in/yaml/v4" "go.yaml.in/yaml/v4"
@@ -28,6 +30,19 @@ func TestExecuteArgsResolve(t *testing.T) {
require.Equal(t, abs, args.resolve(abs)) require.Equal(t, abs, args.resolve(abs))
} }
func TestSharedToolCache(t *testing.T) {
shared, err := sharedToolCache(config.ToolCacheModeShared)
require.NoError(t, err)
require.True(t, shared)
shared, err = sharedToolCache(config.ToolCacheModeNone)
require.NoError(t, err)
require.False(t, shared)
_, err = sharedToolCache("everyone")
require.ErrorContains(t, err, "tool-cache-mode")
}
func TestExecuteArgsPaths(t *testing.T) { func TestExecuteArgsPaths(t *testing.T) {
workdir := t.TempDir() workdir := t.TempDir()
args := &executeArgs{ args := &executeArgs{
+1
View File
@@ -366,6 +366,7 @@ func doRegister(ctx context.Context, cfg *config.Config, inputs *registerInputs)
cfg.Runner.Insecure, cfg.Runner.Insecure,
"", "",
"", "",
config.RequestTimeout,
) )
for { for {
+18 -6
View File
@@ -68,6 +68,8 @@ type Poller struct {
type workerState struct { type workerState struct {
consecutiveEmpty int64 consecutiveEmpty int64
consecutiveErrors int64 consecutiveErrors int64
// fetchTimedOut suppresses repeats of the fetch timeout warning.
fetchTimedOut bool
// lastBackoff is the last interval reported to the PollBackoffSeconds gauge; // lastBackoff is the last interval reported to the PollBackoffSeconds gauge;
// used to suppress redundant no-op Set calls when the backoff plateaus // used to suppress redundant no-op Set calls when the backoff plateaus
// (e.g. at FetchIntervalMax). // (e.g. at FetchIntervalMax).
@@ -346,16 +348,20 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
TasksVersion: v, TasksVersion: v,
})) }))
// DeadlineExceeded is the designed idle path for a long-poll: the server // Our own deadline proves nothing either way: today Gitea answers immediately,
// found no work within FetchTimeout. Treat it as an empty response and do // so it means a slow server, and once it holds the request open it is an idle
// not record the duration — the timeout value would swamp the histogram. // poll. Back off without claiming the server is healthy, warn once a streak so
// neither case floods, and keep it out of the latency histogram.
if errors.Is(err, context.DeadlineExceeded) { if errors.Is(err, context.DeadlineExceeded) {
p.markHealthyPoll() if !s.fetchTimedOut {
s.fetchTimedOut = true
log.Warnf("fetching a task timed out after %s, raise runner.fetch_timeout if this persists", p.cfg.Runner.FetchTimeout)
}
s.consecutiveEmpty++ s.consecutiveEmpty++
s.consecutiveErrors = 0 // timeout is a healthy idle response
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc() metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
return nil, false return nil, false
} }
s.fetchTimedOut = false
metrics.PollFetchDuration.Observe(time.Since(start).Seconds()) metrics.PollFetchDuration.Observe(time.Since(start).Seconds())
if err != nil { if err != nil {
@@ -368,7 +374,13 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
p.shutdownPolling() p.shutdownPolling()
return nil, false return nil, false
} }
log.WithError(err).Error("failed to fetch task") // Not a long poll, so a deadline can mean the server assigned a task
// this runner never received.
if errors.Is(err, context.DeadlineExceeded) {
log.WithError(err).Errorf("fetching a task timed out after %s", p.cfg.Runner.FetchTimeout)
} else {
log.WithError(err).Error("failed to fetch task")
}
p.lastPollFailed.Store(true) p.lastPollFailed.Store(true)
s.consecutiveErrors++ s.consecutiveErrors++
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc() metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc()
+29 -5
View File
@@ -16,6 +16,7 @@ import (
connect_go "connectrpc.com/connect" connect_go "connectrpc.com/connect"
runnerv1 "gitea.dev/actionslib/runner/v1" runnerv1 "gitea.dev/actionslib/runner/v1"
"github.com/sirupsen/logrus/hooks/test"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock" "github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@@ -61,11 +62,7 @@ func TestPoller_WorkerStateCounters(t *testing.T) {
// increments only the per-worker error counter, not the empty counter. // increments only the per-worker error counter, not the empty counter.
func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) { func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
client := mocks.NewClient(t) client := mocks.NewClient(t)
client.On("FetchTask", mock.Anything, mock.Anything).Return( client.On("FetchTask", mock.Anything, mock.Anything).Return(nil, errors.New("network unreachable"))
func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
return nil, errors.New("network unreachable")
},
)
cfg, err := config.LoadDefault("") cfg, err := config.LoadDefault("")
require.NoError(t, err) require.NoError(t, err)
@@ -78,6 +75,33 @@ func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
assert.Equal(t, int64(0), s.consecutiveEmpty) assert.Equal(t, int64(0), s.consecutiveEmpty)
} }
// A deadline is the idle path once the server holds the request open, and a
// symptom before then, so it must neither reset the error count nor assert that
// the server is reachable.
func TestPoller_FetchTimeoutIsNoSignal(t *testing.T) {
client := mocks.NewClient(t)
client.On("FetchTask", mock.Anything, mock.Anything).Return(nil, context.DeadlineExceeded)
cfg, err := config.LoadDefault("")
require.NoError(t, err)
p := &Poller{client: client, cfg: cfg}
p.lastPollFailed.Store(true)
hook := test.NewGlobal()
defer hook.Reset()
s := &workerState{consecutiveErrors: 2}
_, ok := p.fetchTask(context.Background(), s)
require.False(t, ok)
assert.Equal(t, int64(2), s.consecutiveErrors)
assert.Equal(t, int64(1), s.consecutiveEmpty)
assert.True(t, p.lastPollFailed.Load(), "a timeout must not clear a known failure")
// An idle runner against a server holding the request open must not flood.
_, _ = p.fetchTask(context.Background(), s)
assert.Len(t, hook.AllEntries(), 1)
}
// TestPoller_FetchUnauthenticatedStopsPolling verifies that an Unauthenticated // TestPoller_FetchUnauthenticatedStopsPolling verifies that an Unauthenticated
// response marks the runner as unregistered and cancels the polling context so // response marks the runner as unregistered and cancels the polling context so
// the daemon can exit instead of retrying forever. // the daemon can exit instead of retrying forever.
+90 -23
View File
@@ -27,6 +27,8 @@ import (
"gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/pkg/client" "gitea.com/gitea/runner/internal/pkg/client"
"gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/config"
"gitea.com/gitea/runner/internal/pkg/disk"
"gitea.com/gitea/runner/internal/pkg/envcheck"
"gitea.com/gitea/runner/internal/pkg/labels" "gitea.com/gitea/runner/internal/pkg/labels"
"gitea.com/gitea/runner/internal/pkg/metrics" "gitea.com/gitea/runner/internal/pkg/metrics"
"gitea.com/gitea/runner/internal/pkg/report" "gitea.com/gitea/runner/internal/pkg/report"
@@ -89,17 +91,18 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client)
var cacheHandler *artifactcache.Handler var cacheHandler *artifactcache.Handler
if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled { if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled {
if cfg.Cache.ExternalServer != "" { if cfg.Cache.ExternalServer != "" {
warnIgnoredCachePolicy(cfg)
// The v1 client appends its path to this without a separator, so the slash is required. // The v1 client appends its path to this without a separator, so the slash is required.
envs["ACTIONS_CACHE_URL"] = strings.TrimRight(cfg.Cache.ExternalServer, "/") + "/" envs["ACTIONS_CACHE_URL"] = strings.TrimRight(cfg.Cache.ExternalServer, "/") + "/"
} else { } else {
warnIgnoredCacheSecret(cfg) warnIgnoredCacheSecret(cfg)
handler, err := artifactcache.StartHandler( handler, err := artifactcache.StartHandler(artifactcache.Options{
cfg.Cache.Dir, Dir: cfg.Cache.Dir,
cfg.Cache.Host, OutboundIP: cfg.Cache.Host,
cfg.Cache.Port, Port: cfg.Cache.Port,
"", Policy: CachePolicy(cfg),
log.StandardLogger().WithField("module", "cache_request"), Logger: log.StandardLogger().WithField("module", "cache_request"),
) })
if err != nil { if err != nil {
log.Errorf("cannot init cache server, it will be disabled: %v", err) log.Errorf("cannot init cache server, it will be disabled: %v", err)
// go on // go on
@@ -157,8 +160,8 @@ func (r *Runner) OnIdle(ctx context.Context) {
} }
// Host mode: reclaim per-job scratch dirs left behind when HostEnvironment // Host mode: reclaim per-job scratch dirs left behind when HostEnvironment
// cleanup timed out (e.g. a delete stalled by an AV/EDR filter driver). They // cleanup timed out (e.g. a delete stalled by an AV/EDR filter driver). They
// sit under the host workdir parent alongside the shared tool_cache, which // sit under the host workdir parent next to a shared tool_cache, which the name
// the name match leaves untouched. No-op when no host-mode job ever ran. // match leaves untouched. No-op when no host-mode job ever ran.
if hostRoot := filepath.FromSlash(r.cfg.Host.WorkdirParent); hostRoot != "" { if hostRoot := filepath.FromSlash(r.cfg.Host.WorkdirParent); hostRoot != "" {
r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir) r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir)
} }
@@ -170,7 +173,7 @@ func (r *Runner) OnIdle(ctx context.Context) {
// directories above, a task beginning during the pass is safe because the cutoff keeps a // directories above, a task beginning during the pass is safe because the cutoff keeps a
// network it has created but not yet attached a container to out of scope. // network it has created but not yet attached a container to out of scope.
func (r *Runner) cleanupOrphanNetworks(ctx context.Context) { func (r *Runner) cleanupOrphanNetworks(ctx context.Context) {
if r.uuid == "" || !r.labels.RequireDocker() && !r.cfg.Container.RequireDocker { if r.uuid == "" || !r.requiresDocker() {
return return
} }
cutoff := r.now().Add(-r.cfg.Runner.WorkdirCleanupAge) cutoff := r.now().Add(-r.cfg.Runner.WorkdirCleanupAge)
@@ -217,7 +220,7 @@ func isTaskIDDir(name string) bool {
// isHostScratchDir reports whether name is a per-job host-mode scratch dir: // isHostScratchDir reports whether name is a per-job host-mode scratch dir:
// hex.EncodeToString of 8 random bytes, i.e. exactly 16 lowercase hex chars // hex.EncodeToString of 8 random bytes, i.e. exactly 16 lowercase hex chars
// (see startHostEnvironment in act/runner/run_context.go). The narrow match // (see startHostEnvironment in act/runner/run_context.go). The narrow match
// leaves the sibling shared "tool_cache" dir and any operator data untouched. // leaves a sibling shared "tool_cache" dir and any operator data untouched.
func isHostScratchDir(name string) bool { func isHostScratchDir(name string) bool {
if len(name) != 16 { if len(name) != 16 {
return false return false
@@ -345,6 +348,27 @@ func (r *Runner) isSelfHostedActionsURL(task *runnerv1.Task) bool {
return giteaDefaultActionsURL != "" && giteaDefaultActionsURL != "https://github.com" return giteaDefaultActionsURL != "" && giteaDefaultActionsURL != "https://github.com"
} }
// dockerReachable is a variable so tests can substitute one that needs no Docker daemon. It
// probes the environment act connects through, not container.docker_host, which act ignores.
var dockerReachable = func(ctx context.Context) bool {
ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
return envcheck.CheckIfDockerRunning(ctx, "") == nil
}
func (r *Runner) requiresDocker() bool {
return r.labels.RequireDocker() || r.cfg.Container.RequireDocker
}
// fallbackPlatform is where a job runs whose runs-on matches no label, as any job without a
// runs-on does, since Gitea sends those to every runner.
func (r *Runner) fallbackPlatform(ctx context.Context) string {
if r.requiresDocker() || dockerReachable(ctx) {
return r.cfg.Runner.DefaultImage
}
return labels.SelfHostedPlatform
}
func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.Reporter) (err error) { func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.Reporter) (err error) {
defer func() { defer func() {
if r := recover(); r != nil { if r := recover(); r != nil {
@@ -437,10 +461,13 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
// is that server's responsibility to authenticate requests. // is that server's responsibility to authenticate requests.
revokeCache, resultsURL := r.registerCacheForTask(giteaRuntimeToken, preset.Repository, reporter) revokeCache, resultsURL := r.registerCacheForTask(giteaRuntimeToken, preset.Repository, reporter)
defer revokeCache() defer revokeCache()
// A cache server that agreed to forward the artifact half is the whole results service, so // A cache server that agreed to forward the artifact half is the whole results service, so the
// the job is pointed at it and the v2 variable is finally true. // job is pointed at it.
if resultsURL != "" { if resultsURL != "" {
envs["ACTIONS_RESULTS_URL"], envs[runner.CacheServiceV2Env] = resultsURL, "true" envs["ACTIONS_RESULTS_URL"] = resultsURL
if r.cacheServiceV2() {
envs[runner.CacheServiceV2Env] = "true"
}
} }
eventJSON, err := json.Marshal(preset.Event) eventJSON, err := json.Marshal(preset.Event)
@@ -473,6 +500,15 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
// Without bind_workdir, the workspace path omits the task id; concurrent host-mode jobs // Without bind_workdir, the workspace path omits the task id; concurrent host-mode jobs
// for the same repository would share this directory and can race with per-job cleanup. // for the same repository would share this directory and can race with per-job cleanup.
// act asks for the platform once per step, so resolve the fallback at most once per task.
fallbackPlatform := sync.OnceValue(func() string { return r.fallbackPlatform(ctx) })
platformPicker := func(runsOn []string) string {
if platform := r.labels.PickPlatform(runsOn); platform != "" {
return platform
}
return fallbackPlatform()
}
runnerConfig := &runner.Config{ runnerConfig := &runner.Config{
// On Linux, Workdir will be like "/<parent_directory>/<owner>/<repo>" // On Linux, Workdir will be like "/<parent_directory>/<owner>/<repo>"
// On Windows, Workdir will be like "\<parent_directory>\<owner>\<repo>" // On Windows, Workdir will be like "\<parent_directory>\<owner>\<repo>"
@@ -482,7 +518,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
AllocatePTY: r.cfg.Runner.AllocatePTY, AllocatePTY: r.cfg.Runner.AllocatePTY,
ActionOfflineMode: r.cfg.Cache.OfflineMode, ActionOfflineMode: r.cfg.Cache.OfflineMode,
ActionCloneDepth: actionCloneDepth, ActionCloneDepth: actionCloneDepth,
PatchToolkit: r.patchToolkit(), NoActionPatch: r.cfg.Runner.PatchActions != nil && !*r.cfg.Runner.PatchActions,
ReuseContainers: false, ReuseContainers: false,
ForcePull: r.cfg.Container.ForcePull, ForcePull: r.cfg.Container.ForcePull,
@@ -515,11 +551,12 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
Privileged: r.cfg.Container.Privileged, Privileged: r.cfg.Container.Privileged,
DefaultActionInstance: r.getDefaultActionsURL(task), DefaultActionInstance: r.getDefaultActionsURL(task),
DefaultActionInstanceIsSelfHosted: r.isSelfHostedActionsURL(task), DefaultActionInstanceIsSelfHosted: r.isSelfHostedActionsURL(task),
PlatformPicker: r.labels.PickPlatform, PlatformPicker: platformPicker,
JobStartedHook: r.cfg.Runner.Hooks.JobStarted, JobStartedHook: r.cfg.Runner.Hooks.JobStarted,
JobCompletedHook: r.cfg.Runner.Hooks.JobCompleted, JobCompletedHook: r.cfg.Runner.Hooks.JobCompleted,
Vars: task.Vars, Vars: task.Vars,
ValidVolumes: r.cfg.Container.ValidVolumes, ValidVolumes: r.cfg.Container.ValidVolumes,
SharedToolCache: r.cfg.Runner.ToolCacheMode == config.ToolCacheModeShared,
InsecureSkipTLS: r.cfg.Runner.Insecure, InsecureSkipTLS: r.cfg.Runner.Insecure,
RunnerName: r.name, RunnerName: r.name,
} }
@@ -556,10 +593,10 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
return execErr return execErr
} }
// patchToolkit reports whether act should edit the toolkit bundled into an action. It follows the // cacheServiceV2 reports whether jobs are told the cache service speaks v2. It is all cache.v2
// cache URL, because that is what the edits point the client at; see act/runner/toolkit_patch.go. // turns off: the bundle edit that reaches it is what the artifact actions need too.
func (r *Runner) patchToolkit() bool { func (r *Runner) cacheServiceV2() bool {
return r.envs["ACTIONS_CACHE_URL"] != "" && (r.cfg.Cache.V2 == nil || *r.cfg.Cache.V2) return r.cfg.Cache.V2 == nil || *r.cfg.Cache.V2
} }
// registerCacheForTask tells the cache server to accept requests authenticated // registerCacheForTask tells the cache server to accept requests authenticated
@@ -605,14 +642,15 @@ func (r *Runner) registerExternalCacheJob(token string, cred artifactcache.JobCr
resultsURL := "" resultsURL := ""
if body, err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, map[string]any{ if body, err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, map[string]any{
"token": token, "repo": cred.Repo, "results": cred.Results, "insecure_tls": cred.InsecureTLS, "token": token, "repo": cred.Repo, "results": cred.Results, "insecure_tls": cred.InsecureTLS,
"public_url": base,
}); err != nil { }); err != nil {
log.Warnf("cache external_server register failed (%s): %v", base, err) log.Warnf("cache external_server register failed (%s): %v", base, err)
if reporter != nil { if reporter != nil {
reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf( reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf(
"cache external_server register failed (%s): %v — cache requests from this job will be unauthenticated and likely return 401", base, err))) "cache external_server register failed (%s): %v — cache requests from this job will be unauthenticated and likely return 401", base, err)))
} }
} else { } else if forwarded, _ := body["results_url"].(string); forwarded != "" {
resultsURL, _ = body["results_url"].(string) // absent from a server too old to forward resultsURL = base // the answer only says it forwards, its own address need not be the job's
} }
return func() { return func() {
if _, err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret, if _, err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret,
@@ -692,7 +730,7 @@ func checkFreeDisk(cfg *config.Config) (bool, string) {
root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/")) root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/"))
} }
root = nearestExistingPath(root) root = nearestExistingPath(root)
available, err := freeDiskBytes(root) available, err := disk.FreeBytes(root)
if err != nil { if err != nil {
return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err) return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err)
} }
@@ -725,6 +763,35 @@ func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Respons
})) }))
} }
// minFreeDisk keeps the cache from growing past the point where the runner stops taking
// work, but only when health checks are on, since the key is documented as opt-in.
func minFreeDisk(cfg *config.Config) int64 {
if !cfg.HealthCheck.Enabled {
return 0
}
return cfg.HealthCheck.MinFreeDiskSpaceMB * 1024 * 1024
}
// CachePolicy maps the cache config onto the cache server's own type, in bytes not MiB.
func CachePolicy(cfg *config.Config) artifactcache.Policy {
return artifactcache.Policy{
Retention: cfg.Cache.Retention,
RepoSizeLimit: int64(cfg.Cache.RepoSizeLimit),
SizeLimit: int64(cfg.Cache.SizeLimit),
SweepInterval: cfg.Cache.SweepInterval,
MinFreeDisk: minFreeDisk(cfg),
}
}
// warnIgnoredCachePolicy flags eviction settings configured on a runner that points at an external cache server.
func warnIgnoredCachePolicy(cfg *config.Config) {
defaults := config.DefaultCache()
if cfg.Cache.Retention != defaults.Retention || cfg.Cache.RepoSizeLimit != defaults.RepoSizeLimit ||
cfg.Cache.SizeLimit != defaults.SizeLimit || cfg.Cache.SweepInterval != defaults.SweepInterval {
log.Warn("cache eviction settings are ignored when cache.external_server is set; configure them on that server instead")
}
}
// warnIgnoredCacheSecret flags an external cache server secret configured on a runner that uses the built-in cache server. // warnIgnoredCacheSecret flags an external cache server secret configured on a runner that uses the built-in cache server.
func warnIgnoredCacheSecret(cfg *config.Config) { func warnIgnoredCacheSecret(cfg *config.Config) {
if cfg.Cache.ExternalServer != "" { if cfg.Cache.ExternalServer != "" {
+10 -9
View File
@@ -25,7 +25,7 @@ func emptyCfg() *config.Config { return &config.Config{} }
func TestRunner_registerCacheForTask(t *testing.T) { func TestRunner_registerCacheForTask(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil) handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -62,7 +62,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) {
t.Run("empty token", func(t *testing.T) { t.Run("empty token", func(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil) handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -77,7 +77,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) {
// /find, no auth on the signed archiveLocation download. // /find, no auth on the signed archiveLocation download.
func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) { func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) {
dir := filepath.Join(t.TempDir(), "artifactcache") dir := filepath.Join(t.TempDir(), "artifactcache")
handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil) handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"})
require.NoError(t, err) require.NoError(t, err)
defer handler.Close() defer handler.Close()
@@ -157,14 +157,15 @@ func decodeJSON(resp *http.Response, v any) error {
// End-to-end against a remote cache-server: token unknown → 401, register → // End-to-end against a remote cache-server: token unknown → 401, register →
// reserve/upload/commit/find/download all OK, revoke → 401 again. Registering also names the // reserve/upload/commit/find/download all OK, revoke → 401 again. Registering also names the
// instance, and the server answering with its own address is what makes a shared cache server the // instance and the address its jobs reach the server at, which makes a shared cache server the
// whole results service, as the built-in one is. // whole results service, as the built-in one is.
func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) { func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
dir := filepath.Join(t.TempDir(), "remote-cache") dir := filepath.Join(t.TempDir(), "remote-cache")
const secret = "shared-secret-for-tests" const secret = "shared-secret-for-tests"
remote, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, secret, nil) remote, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.2", InternalSecret: secret}) // advertised, never dialled
require.NoError(t, err) require.NoError(t, err)
defer remote.Close() defer remote.Close()
external := strings.Replace(remote.ExternalURL(), "127.0.0.2", "127.0.0.1", 1)
gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_, _ = io.WriteString(w, `{"ok":true}`) _, _ = io.WriteString(w, `{"ok":true}`)
})) }))
@@ -172,7 +173,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
r := &Runner{ r := &Runner{
cfg: &config.Config{Cache: config.Cache{ cfg: &config.Config{Cache: config.Cache{
ExternalServer: remote.ExternalURL(), ExternalServer: external,
ExternalSecret: secret, ExternalSecret: secret,
}}, }},
envs: map[string]string{"ACTIONS_RESULTS_URL": gitea.URL}, envs: map[string]string{"ACTIONS_RESULTS_URL": gitea.URL},
@@ -180,7 +181,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
token := "external-task-token" token := "external-task-token"
repo := "owner/repoX" repo := "owner/repoX"
base := remote.ExternalURL() + "/_apis/artifactcache" base := external + "/_apis/artifactcache"
probe := func() int { probe := func() int {
req, _ := http.NewRequest(http.MethodGet, base+"/cache?keys=k&version=v", nil) req, _ := http.NewRequest(http.MethodGet, base+"/cache?keys=k&version=v", nil)
req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("Authorization", "Bearer "+token)
@@ -198,7 +199,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
"token must be accepted after registerCacheForTask") "token must be accepted after registerCacheForTask")
// The server took the results service over, so the artifact half reaches Gitea through it. // The server took the results service over, so the artifact half reaches Gitea through it.
require.Equal(t, remote.ExternalURL(), resultsURL) require.Equal(t, external, resultsURL)
artifact, err := http.NewRequestWithContext(t.Context(), http.MethodPost, artifact, err := http.NewRequestWithContext(t.Context(), http.MethodPost,
resultsURL+"/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", nil) resultsURL+"/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", nil)
require.NoError(t, err) require.NoError(t, err)
@@ -248,7 +249,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
ArchiveLocation string `json:"archiveLocation"` ArchiveLocation string `json:"archiveLocation"`
} }
require.NoError(t, decodeJSON(resp, &hit)) require.NoError(t, decodeJSON(resp, &hit))
require.NotEmpty(t, hit.ArchiveLocation) require.True(t, strings.HasPrefix(hit.ArchiveLocation, external), hit.ArchiveLocation)
dl, err := http.Get(hit.ArchiveLocation) dl, err := http.Get(hit.ArchiveLocation)
require.NoError(t, err) require.NoError(t, err)
+36 -4
View File
@@ -12,6 +12,7 @@ import (
"gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/act/runner"
clientmocks "gitea.com/gitea/runner/internal/pkg/client/mocks" clientmocks "gitea.com/gitea/runner/internal/pkg/client/mocks"
"gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/config"
"gitea.com/gitea/runner/internal/pkg/labels"
"gitea.com/gitea/runner/internal/pkg/ver" "gitea.com/gitea/runner/internal/pkg/ver"
"connectrpc.com/connect" "connectrpc.com/connect"
@@ -98,6 +99,34 @@ func TestNewRunnerInitializesLabelsAndEnvironment(t *testing.T) {
require.Empty(t, r.envs[runner.CacheServiceV2Env], "no cache server, nothing to serve v2 from") require.Empty(t, r.envs[runner.CacheServiceV2Env], "no cache server, nothing to serve v2 from")
} }
func TestRunnerFallbackPlatform(t *testing.T) {
tests := []struct {
name string
label string
dockerRunning bool
want string
}{
{"a docker label needs no daemon probe", "ubuntu:docker://node:18", false, "mirror.example/ci:noble"},
{"host labels keep the image where docker runs", "ubuntu:host", true, "mirror.example/ci:noble"},
{"host labels without docker run on the host", "ubuntu:host", false, labels.SelfHostedPlatform},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
reachable := dockerReachable
dockerReachable = func(context.Context) bool { return tt.dockerRunning }
t.Cleanup(func() { dockerReachable = reachable })
label, err := labels.Parse(tt.label)
require.NoError(t, err)
cfg := &config.Config{}
cfg.Runner.DefaultImage = "mirror.example/ci:noble"
r := &Runner{cfg: cfg, labels: labels.Labels{label}}
require.Equal(t, tt.want, r.fallbackPlatform(t.Context()))
})
}
}
// Proxy variables are assembled per task, because a job's service containers have to be // Proxy variables are assembled per task, because a job's service containers have to be
// reached directly and they are only known once the workflow is parsed. // reached directly and they are only known once the workflow is parsed.
func TestNewRunnerLeavesProxyToTheTask(t *testing.T) { func TestNewRunnerLeavesProxyToTheTask(t *testing.T) {
@@ -140,7 +169,6 @@ func TestNewRunnerCacheServiceV2(t *testing.T) {
assert.Equal(t, "https://gitea.example", r.envs["ACTIONS_RESULTS_URL"]) assert.Equal(t, "https://gitea.example", r.envs["ACTIONS_RESULTS_URL"])
assert.Empty(t, r.envs[runner.CacheServiceV2Env], "a promise the runner has not made yet") assert.Empty(t, r.envs[runner.CacheServiceV2Env], "a promise the runner has not made yet")
assert.True(t, r.patchToolkit())
// The registration is what makes it true: the cache server takes the results service over, // The registration is what makes it true: the cache server takes the results service over,
// having been told which instance to forward the artifact half to. // having been told which instance to forward the artifact half to.
@@ -161,6 +189,11 @@ func TestNewRunnerCacheServiceV2(t *testing.T) {
defer resp.Body.Close() defer resp.Body.Close()
assert.Equal(t, http.StatusOK, resp.StatusCode, "the advertised results service serves no cache service") assert.Equal(t, http.StatusOK, resp.StatusCode, "the advertised results service serves no cache service")
// Turning v2 off withdraws the advertisement and nothing else.
assert.True(t, r.cacheServiceV2())
cfg.Cache.V2 = new(bool)
assert.False(t, r.cacheServiceV2())
} }
// The v1 cache client appends its path to ACTIONS_CACHE_URL without a separator, so a configured // The v1 cache client appends its path to ACTIONS_CACHE_URL without a separator, so a configured
@@ -174,9 +207,8 @@ func TestNewRunnerNormalizesTheExternalCacheServer(t *testing.T) {
r := NewRunner(cfg, &config.Registration{Name: "runner"}, cli) r := NewRunner(cfg, &config.Registration{Name: "runner"}, cli)
assert.Equal(t, "http://cache.local:8088/", r.envs["ACTIONS_CACHE_URL"]) assert.Equal(t, "http://cache.local:8088/", r.envs["ACTIONS_CACHE_URL"])
// Nothing to front the results service with, so the variable stays unset, but the bundles are // Nothing to front the results service with, so the variable stays unset and the client keeps
// still patched: artifacts v4 need that, and the patch keeps the cache client on the cache URL. // to v1, which reads the cache URL first.
assert.Equal(t, "https://gitea.example", r.envs["ACTIONS_RESULTS_URL"]) assert.Equal(t, "https://gitea.example", r.envs["ACTIONS_RESULTS_URL"])
assert.Empty(t, r.envs[runner.CacheServiceV2Env]) assert.Empty(t, r.envs[runner.CacheServiceV2Env])
assert.True(t, r.patchToolkit())
} }
+19 -1
View File
@@ -5,8 +5,10 @@ package run
import ( import (
"fmt" "fmt"
"maps"
"os" "os"
"runtime" "runtime"
"slices"
"strconv" "strconv"
"strings" "strings"
@@ -14,6 +16,7 @@ import (
"gitea.com/gitea/runner/internal/pkg/ver" "gitea.com/gitea/runner/internal/pkg/ver"
runnerv1 "gitea.dev/actionslib/runner/v1" runnerv1 "gitea.dev/actionslib/runner/v1"
"google.golang.org/protobuf/types/known/structpb"
) )
// osReleasePath describes the host distribution on Linux; absent elsewhere, where the platform // osReleasePath describes the host distribution on Linux; absent elsewhere, where the platform
@@ -46,12 +49,27 @@ func (r *Runner) setupLines(task *runnerv1.Task) []string {
"Repository: "+fields["repository"].GetStringValue(), "Repository: "+fields["repository"].GetStringValue(),
"Triggered by event: "+fields["event_name"].GetStringValue(), "Triggered by event: "+fields["event_name"].GetStringValue(),
"::endgroup::", "::endgroup::",
"::group::Operating System",
) )
lines = append(lines, inputLines(fields)...)
lines = append(lines, "::group::Operating System")
lines = append(lines, osInfo()...) lines = append(lines, osInfo()...)
return append(lines, "::endgroup::") return append(lines, "::endgroup::")
} }
// inputLines lists the inputs the run was triggered with, as workflow_dispatch and workflow_call
// carry them in the event payload. Empty when the event has none.
func inputLines(fields map[string]*structpb.Value) []string {
inputs := fields["event"].GetStructValue().GetFields()["inputs"].GetStructValue().GetFields()
if len(inputs) == 0 {
return nil
}
lines := []string{"::group::Inputs"}
for _, name := range slices.Sorted(maps.Keys(inputs)) {
lines = append(lines, fmt.Sprintf("%s: %v", name, inputs[name].AsInterface()))
}
return append(lines, "::endgroup::")
}
// osInfo describes the host the runner executes on. // osInfo describes the host the runner executes on.
func osInfo() []string { func osInfo() []string {
lines := make([]string, 0, 2) lines := make([]string, 0, 2)
+39
View File
@@ -56,6 +56,45 @@ func TestSetupLines(t *testing.T) {
}, r.setupLines(&runnerv1.Task{Id: 268506, Context: taskCtx})) }, r.setupLines(&runnerv1.Task{Id: 268506, Context: taskCtx}))
} }
func TestSetupLinesInputs(t *testing.T) {
original := osReleasePath
osReleasePath = filepath.Join(t.TempDir(), "absent")
defer func() { osReleasePath = original }()
r := &Runner{name: "gitea-com-gitea-0003"}
taskCtx, err := structpb.NewStruct(map[string]any{
"job": "test",
"repository": "gitea/runner",
"event_name": "workflow_dispatch",
"event": map[string]any{
"inputs": map[string]any{
"with_default": "default",
"required": "required input",
"boolean": true,
},
},
})
require.NoError(t, err)
assert.Equal(t, []string{
"gitea-com-gitea-0003(version:" + ver.Version() + ")",
"::group::Runner Information",
"Task: 1",
"Job: test",
"Repository: gitea/runner",
"Triggered by event: workflow_dispatch",
"::endgroup::",
"::group::Inputs",
"boolean: true",
"required: required input",
"with_default: default",
"::endgroup::",
"::group::Operating System",
fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH),
"::endgroup::",
}, r.setupLines(&runnerv1.Task{Id: 1, Context: taskCtx}))
}
func TestPrettyOSName(t *testing.T) { func TestPrettyOSName(t *testing.T) {
tests := map[string]struct { tests := map[string]struct {
osRelease string osRelease string
+7 -5
View File
@@ -17,7 +17,7 @@ import (
"gitea.dev/actionslib/runner/v1/runnerv1connect" "gitea.dev/actionslib/runner/v1/runnerv1connect"
) )
func getHTTPClient(endpoint string, insecure bool) *http.Client { func getHTTPClient(endpoint string, insecure bool, timeout time.Duration) *http.Client {
transport := &http.Transport{ transport := &http.Transport{
Proxy: http.ProxyFromEnvironment, Proxy: http.ProxyFromEnvironment,
MaxIdleConns: 10, MaxIdleConns: 10,
@@ -29,11 +29,13 @@ func getHTTPClient(endpoint string, insecure bool) *http.Client {
InsecureSkipVerify: true, InsecureSkipVerify: true,
} }
} }
return &http.Client{Transport: transport} return &http.Client{Transport: transport, Timeout: timeout}
} }
// New returns a new runner client. // New returns a new runner client. timeout bounds every RPC: without it a
func New(endpoint string, insecure bool, uuid, token string, opts ...connect.ClientOption) *HTTPClient { // stalled connection parks the reporter for the whole job context, so logs and
// heartbeats stop together and the task is reaped as a zombie.
func New(endpoint string, insecure bool, uuid, token string, timeout time.Duration, opts ...connect.ClientOption) *HTTPClient {
baseURL := strings.TrimRight(endpoint, "/") + "/api/actions" baseURL := strings.TrimRight(endpoint, "/") + "/api/actions"
opts = append(opts, connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { opts = append(opts, connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
@@ -49,7 +51,7 @@ func New(endpoint string, insecure bool, uuid, token string, opts ...connect.Cli
} }
}))) })))
httpClient := getHTTPClient(endpoint, insecure) httpClient := getHTTPClient(endpoint, insecure, timeout)
return &HTTPClient{ return &HTTPClient{
PingServiceClient: pingv1connect.NewPingServiceClient( PingServiceClient: pingv1connect.NewPingServiceClient(
httpClient, httpClient,
+7 -5
View File
@@ -8,6 +8,7 @@ import (
"net/http/httptest" "net/http/httptest"
"strings" "strings"
"testing" "testing"
"time"
"connectrpc.com/connect" "connectrpc.com/connect"
pingv1 "gitea.dev/actionslib/ping/v1" pingv1 "gitea.dev/actionslib/ping/v1"
@@ -17,7 +18,8 @@ import (
func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) { func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) {
t.Setenv("HTTP_PROXY", "http://proxy.example.com:8080") t.Setenv("HTTP_PROXY", "http://proxy.example.com:8080")
client := getHTTPClient("http://gitea.example.com", false) client := getHTTPClient("http://gitea.example.com", false, time.Minute)
require.Equal(t, time.Minute, client.Timeout)
transport, ok := client.Transport.(*http.Transport) transport, ok := client.Transport.(*http.Transport)
require.True(t, ok) require.True(t, ok)
@@ -32,7 +34,7 @@ func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) {
func TestGetHTTPClientInsecureTLS(t *testing.T) { func TestGetHTTPClientInsecureTLS(t *testing.T) {
// insecure only takes effect for https endpoints // insecure only takes effect for https endpoints
httpsInsecure := getHTTPClient("https://gitea.example.com", true) httpsInsecure := getHTTPClient("https://gitea.example.com", true, time.Minute)
transport, ok := httpsInsecure.Transport.(*http.Transport) transport, ok := httpsInsecure.Transport.(*http.Transport)
require.True(t, ok) require.True(t, ok)
require.NotNil(t, transport.TLSClientConfig) require.NotNil(t, transport.TLSClientConfig)
@@ -47,7 +49,7 @@ func TestGetHTTPClientInsecureTLS(t *testing.T) {
{"http insecure ignored", "http://gitea.example.com", true}, {"http insecure ignored", "http://gitea.example.com", true},
} { } {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
c := getHTTPClient(tc.endpoint, tc.insecure) c := getHTTPClient(tc.endpoint, tc.insecure, time.Minute)
tr, ok := c.Transport.(*http.Transport) tr, ok := c.Transport.(*http.Transport)
require.True(t, ok) require.True(t, ok)
require.Nil(t, tr.TLSClientConfig) require.Nil(t, tr.TLSClientConfig)
@@ -66,7 +68,7 @@ func TestNewSetsBaseURLAndHeaders(t *testing.T) {
defer server.Close() defer server.Close()
// trailing slash must be trimmed before "/api/actions" is appended // trailing slash must be trimmed before "/api/actions" is appended
c := New(server.URL+"/", false, "the-uuid", "the-token") c := New(server.URL+"/", false, "the-uuid", "the-token", time.Minute)
// Address returns the endpoint as supplied (untrimmed) // Address returns the endpoint as supplied (untrimmed)
require.Equal(t, server.URL+"/", c.Address()) require.Equal(t, server.URL+"/", c.Address())
require.False(t, c.Insecure()) require.False(t, c.Insecure())
@@ -87,7 +89,7 @@ func TestNewOmitsEmptyHeaders(t *testing.T) {
})) }))
defer server.Close() defer server.Close()
c := New(server.URL, false, "", "") c := New(server.URL, false, "", "", time.Minute)
_, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"})) _, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"}))
require.Empty(t, gotHeaders.Get(UUIDHeader)) require.Empty(t, gotHeaders.Get(UUIDHeader))
+44 -8
View File
@@ -1,13 +1,20 @@
# Every option with its default value, all commented out. Read this file, do not copy it. # Every option with its default value, all commented out. Read this file, do not copy it.
# `./gitea-runner config init` writes a config file to copy the lines you change into. # `./gitea-runner config init` writes a config file to copy the lines you change into.
# Logging for the runner process itself (messages printed to stderr). # Logging for the runner process itself (messages printed to stderr), plus the copy of
# This does not control how workflow step output is streamed to the Gitea UI; # each task's log kept under log.job. Neither controls how workflow step output is streamed
# tune that with runner.log_report_* below. # to the Gitea UI; tune that with runner.log_report_* below.
log: log:
# logrus severity: trace, debug, info, warn, error, fatal, panic. # logrus severity: trace, debug, info, warn, error, fatal, panic.
# trace and debug turn on caller/file:line in log lines. Default if omitted: info. # trace and debug turn on caller/file:line in log lines. Default if omitted: info.
#level: info #level: info
# Write a copy of each task's log to dir as <start time>-task-<id>.log, so a job's output
# survives a failure to send it to Gitea. A path turns them on, empty turns them off.
# retention is how long a log is kept (0s keeps all), max_size caps one log (0 is no limit).
#job:
# dir: ""
# retention: 168h
# max_size: 1GB
runner: runner:
# Where to store the registration result. # Where to store the registration result.
@@ -34,6 +41,7 @@ runner:
# Whether skip verifying the TLS certificate of the Gitea instance. # Whether skip verifying the TLS certificate of the Gitea instance.
#insecure: false #insecure: false
# The timeout for fetching the job from the Gitea instance. # The timeout for fetching the job from the Gitea instance.
# Values above the 60s RPC timeout are capped to it.
#fetch_timeout: 5s #fetch_timeout: 5s
# The interval for fetching the job from the Gitea instance. # The interval for fetching the job from the Gitea instance.
#fetch_interval: 2s #fetch_interval: 2s
@@ -79,6 +87,10 @@ runner:
# When true (the default), inject the ACT=true environment variable into jobs. # When true (the default), inject the ACT=true environment variable into jobs.
# Set to false so workflows gated on `if: ${{ !env.ACT }}` behave like they do on GitHub. # Set to false so workflows gated on `if: ${{ !env.ACT }}` behave like they do on GitHub.
#set_act_env: true #set_act_env: true
# When true (the default), apply compatibility patches to the actions a job runs, so actions
# written for GitHub work against this instance. Set to false to run them exactly as published,
# at the price of the stock artifact actions refusing and the cache client keeping to v1.
#patch_actions: true
# The labels of a runner are used to determine which jobs the runner can run, and how to run them. # The labels of a runner are used to determine which jobs the runner can run, and how to run them.
# Like: "macos-arm64:host" or "ubuntu-latest:docker://docker.gitea.com/runner-images:ubuntu-latest" # Like: "macos-arm64:host" or "ubuntu-latest:docker://docker.gitea.com/runner-images:ubuntu-latest"
# Find more images provided by Gitea at https://gitea.com/gitea/runner-images . # Find more images provided by Gitea at https://gitea.com/gitea/runner-images .
@@ -93,6 +105,15 @@ runner:
# terminal; tools like `docker build` emit redrawing progress frames into the captured log # terminal; tools like `docker build` emit redrawing progress frames into the captured log
# when a TTY is present. # when a TTY is present.
#allocate_pty: false #allocate_pty: false
# Image for a job whose runs-on matches none of the labels above. A runner without a
# docker daemon runs such a job on the host instead.
#default_image: "docker.gitea.com/runner-images:ubuntu-latest"
# What to mount at RUNNER_TOOL_CACHE (/opt/hostedtoolcache), where setup actions install tools:
# none: nothing. A docker job sees what its image ships there, a host job an empty dir, and
# either way what it installs is gone when the job ends.
# shared: one tool cache every job reuses. Two jobs writing the same tool version at once
# corrupt it, so use it only with capacity 1.
#tool_cache_mode: none
# Optional executable on the host, run once after each task's built-in cleanup # Optional executable on the host, run once after each task's built-in cleanup
# (post-steps, container teardown, bind-workdir removal). Additive only. # (post-steps, container teardown, bind-workdir removal). Additive only.
# #
@@ -137,6 +158,7 @@ cache:
#port: 0 #port: 0
# URL of a shared `gitea-runner cache-server` to use instead of starting a local one. # URL of a shared `gitea-runner cache-server` to use instead of starting a local one.
# Set on every runner that should share a cache pool. A trailing slash is optional. # Set on every runner that should share a cache pool. A trailing slash is optional.
# Jobs reach the server at this URL too, so set it to the reverse proxy when one fronts the server.
# Example: "http://cache-host:8088/" # Example: "http://cache-host:8088/"
# Requires external_secret (below) to match the value on the cache-server. # Requires external_secret (below) to match the value on the cache-server.
#external_server: "" #external_server: ""
@@ -153,11 +175,25 @@ cache:
# A moved tag (e.g. a re-tagged "v6") or an updated branch stays at the cached commit # A moved tag (e.g. a re-tagged "v6") or an updated branch stays at the cached commit
# until its cache entry expires or is manually removed. # until its cache entry expires or is manually removed.
#offline_mode: false #offline_mode: false
# Serve the actions cache service v2 API, used by actions/cache@v4.2 and later. Those actions # Serve the actions cache service v2 API. The actions that use it refuse any host they do not
# refuse any host they do not take for GitHub, so reaching it means editing that check out of # take for GitHub, so reaching it means editing that check out of their own bundle, undone
# the action's own bundle, keeping the untouched copy beside it. The same edit lets the stock # whenever it is downloaded again. That edit is made either way, this only governs the API
# upload-artifact and download-artifact work here. A bundle that does not match is left alone. # advertised. A bundle that does not match is left alone.
#v2: true #v2: true
# How the cache server discards entries, ignored when external_server is set since that
# server applies its own. Leave a setting out for its default; 0s or 0 turns the three
# limits off. Sizes accept 10GB, 512mb, 1TiB or a plain byte count, binary either way.
# Whatever these allow, the cache still sheds entries to keep free space on its volume
# above health_check.min_free_disk_space_mb when health checks are enabled.
# Remove entries nothing has read or written within this window. Only last access counts.
#retention: 168h
# Cap one repository, removing its least recently accessed entries until it fits. An entry
# larger than the limit is dropped rather than emptying the repository to make room.
#repo_size_limit: 10GB
# Cap the whole cache the same way. Off by default, since the free space floor bounds it.
#size_limit: 0
# Minimum time between two eviction sweeps. This one has no "off".
#sweep_interval: 1h
container: container:
# Specifies the network to which the container will connect. # Specifies the network to which the container will connect.
@@ -180,7 +216,7 @@ container:
#privileged: false #privileged: false
# Any other options to be used when the container is started, for example: # Any other options to be used when the container is started, for example:
# options: --add-host=my.gitea.url:host-gateway # options: --add-host=my.gitea.url:host-gateway
# A volume declared here replaces the one the runner mounts on the same container path, so the # A volume declared here replaces the one the runner would mount on the same container path, so the
# tool cache can be kept on the host. Its source must also be allowed by valid_volumes below: # tool cache can be kept on the host. Its source must also be allowed by valid_volumes below:
# options: --volume /host/toolcache:/opt/hostedtoolcache # options: --volume /host/toolcache:/opt/hostedtoolcache
#options: #options:
+89 -3
View File
@@ -10,14 +10,22 @@ import (
"maps" "maps"
"os" "os"
"path/filepath" "path/filepath"
"slices"
"strings" "strings"
"time" "time"
"github.com/docker/go-units"
"github.com/joho/godotenv" "github.com/joho/godotenv"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"go.yaml.in/yaml/v4" "go.yaml.in/yaml/v4"
) )
// RequestTimeout bounds every RPC to Gitea, and with it runner.fetch_timeout.
const RequestTimeout = 60 * time.Second
// DefaultImage is the image jobs run in unless a label or runner.default_image names another.
const DefaultImage = "docker.gitea.com/runner-images:ubuntu-latest"
// DefaultPostTaskScriptTimeout is the fallback cap on how long the post-task // DefaultPostTaskScriptTimeout is the fallback cap on how long the post-task
// script may run when post_task_script is set without an explicit timeout. It is // script may run when post_task_script is set without an explicit timeout. It is
// applied both at config load (for a configured script) and at the point of use // applied both at config load (for a configured script) and at the point of use
@@ -30,9 +38,17 @@ const Minimal = `# Minimal config file. Every option it does not set keeps its d
# "gitea-runner config generate" prints all options, "config set <key> <value>" sets one here. # "gitea-runner config generate" prints all options, "config set <key> <value>" sets one here.
` `
// Log represents the configuration for logging. // Log represents the runner process's own logging, plus the copy of each task's log it keeps.
type Log struct { type Log struct {
Level string `yaml:"level"` // Level indicates the logging level. Level string `yaml:"level"` // Level indicates the logging level.
Job LogJob `yaml:"job"` // Job configures the copy of each task's log kept on the runner host.
}
// LogJob represents the configuration for the copy of each task's log kept on the runner host.
type LogJob struct {
Dir string `yaml:"dir"` // Dir is the directory the runner writes each task's log to. Empty, the default, writes none.
Retention time.Duration `yaml:"retention"` // Retention deletes a task's log directory once it is older than this. Default 168h, 0s keeps them regardless of age.
MaxSize Size `yaml:"max_size"` // MaxSize caps one task's log file. Default 1GB, 0 is no limit.
} }
// Runner represents the configuration for the runner. // Runner represents the configuration for the runner.
@@ -58,7 +74,10 @@ type Runner struct {
GithubMirror string `yaml:"github_mirror"` // GithubMirror defines what mirrors should be used when using github GithubMirror string `yaml:"github_mirror"` // GithubMirror defines what mirrors should be used when using github
ActionShallowClone *bool `yaml:"action_shallow_clone"` // ActionShallowClone fetches only the requested ref of an action repository at depth 1 instead of cloning every branch's full history. It is a pointer to distinguish between false and not set; if not set, it defaults to true. ActionShallowClone *bool `yaml:"action_shallow_clone"` // ActionShallowClone fetches only the requested ref of an action repository at depth 1 instead of cloning every branch's full history. It is a pointer to distinguish between false and not set; if not set, it defaults to true.
SetActEnv *bool `yaml:"set_act_env"` // SetActEnv controls whether the ACT=true environment variable is injected into jobs. It is a pointer to distinguish between false and not set; if not set, it defaults to true. Set it to false so workflows gated on `if: ${{ !env.ACT }}` behave like on GitHub. SetActEnv *bool `yaml:"set_act_env"` // SetActEnv controls whether the ACT=true environment variable is injected into jobs. It is a pointer to distinguish between false and not set; if not set, it defaults to true. Set it to false so workflows gated on `if: ${{ !env.ACT }}` behave like on GitHub.
PatchActions *bool `yaml:"patch_actions"` // PatchActions applies compatibility patches to the actions a job runs, so actions written for GitHub work against Gitea, see act/runner/patch_actions.go. It is a pointer to distinguish between false and not set; if not set, it defaults to true. Set it to false to run every action exactly as published, at the price of the artifact actions refusing.
AllocatePTY bool `yaml:"allocate_pty"` // AllocatePTY allocates a pseudo-TTY for each step's process. Default is false, matching GitHub's actions/runner. Enable only for jobs that need an interactive terminal; tools like docker build emit redrawing progress frames into the captured log when a TTY is present. Applies to both host and docker backends. AllocatePTY bool `yaml:"allocate_pty"` // AllocatePTY allocates a pseudo-TTY for each step's process. Default is false, matching GitHub's actions/runner. Enable only for jobs that need an interactive terminal; tools like docker build emit redrawing progress frames into the captured log when a TTY is present. Applies to both host and docker backends.
DefaultImage string `yaml:"default_image"` // DefaultImage is the image a job runs in when its runs-on matches none of the runner's labels. A runner without docker runs such a job on the host instead.
ToolCacheMode string `yaml:"tool_cache_mode"` // ToolCacheMode is what the runner mounts at RUNNER_TOOL_CACHE on both backends: ToolCacheModeNone or ToolCacheModeShared.
PostTaskScript string `yaml:"post_task_script"` // PostTaskScript is the path to an executable script run on the host after each task's cleanup completes. Empty disables the hook. On Windows use .exe/.bat/.cmd; PowerShell (.ps1) is not supported yet as the configured path. PostTaskScript string `yaml:"post_task_script"` // PostTaskScript is the path to an executable script run on the host after each task's cleanup completes. Empty disables the hook. On Windows use .exe/.bat/.cmd; PowerShell (.ps1) is not supported yet as the configured path.
PostTaskScriptTimeout time.Duration `yaml:"post_task_script_timeout"` // PostTaskScriptTimeout caps how long the post-task script may run. Default is 5m when post_task_script is set. PostTaskScriptTimeout time.Duration `yaml:"post_task_script_timeout"` // PostTaskScriptTimeout caps how long the post-task script may run. Default is 5m when post_task_script is set.
Hooks RunnerHooks `yaml:"hooks"` // Hooks are scripts run inside the job environment around the job's steps. Hooks RunnerHooks `yaml:"hooks"` // Hooks are scripts run inside the job environment around the job's steps.
@@ -80,7 +99,48 @@ type Cache struct {
ExternalSecret string `yaml:"external_secret"` // ExternalSecret is a shared secret between this runner and an external gitea-runner cache-server, enabling per-job ACTIONS_RUNTIME_TOKEN authentication and repo scoping over the network. Required whenever ExternalServer is set; ExternalSecretFile is the alternative way to provide it. ExternalSecret string `yaml:"external_secret"` // ExternalSecret is a shared secret between this runner and an external gitea-runner cache-server, enabling per-job ACTIONS_RUNTIME_TOKEN authentication and repo scoping over the network. Required whenever ExternalServer is set; ExternalSecretFile is the alternative way to provide it.
ExternalSecretFile string `yaml:"external_secret_file"` // ExternalSecretFile is the path to a file holding the ExternalSecret value, so the secret can be mounted instead of stored in the config file. LoadDefault reads it into ExternalSecret; setting both is an error. ExternalSecretFile string `yaml:"external_secret_file"` // ExternalSecretFile is the path to a file holding the ExternalSecret value, so the secret can be mounted instead of stored in the config file. LoadDefault reads it into ExternalSecret; setting both is an error.
OfflineMode bool `yaml:"offline_mode"` // OfflineMode reuses a cached action without fetching from the remote; a moved tag or branch stays at the cached commit until the cache entry is removed. OfflineMode bool `yaml:"offline_mode"` // OfflineMode reuses a cached action without fetching from the remote; a moved tag or branch stays at the cached commit until the cache entry is removed.
V2 *bool `yaml:"v2"` // V2 serves the actions cache service v2 API to jobs, used by actions/cache@v4.2 and later, and edits the action bundles that would otherwise refuse it. Unset means enabled. V2 *bool `yaml:"v2"` // V2 advertises the actions cache service v2 API to jobs. The bundle edit that reaches it is made either way, the artifact actions need it too. Unset means enabled.
// Eviction settings, ignored when ExternalServer is set since that server applies its own.
Retention time.Duration `yaml:"retention"` // Retention removes entries nothing has read or written within this window. Default 168h, 0 keeps them regardless of age.
RepoSizeLimit Size `yaml:"repo_size_limit"` // RepoSizeLimit caps one repository, evicting least recently accessed first. Default 10GB, 0 is no limit.
SizeLimit Size `yaml:"size_limit"` // SizeLimit caps the whole cache the same way. No limit by default.
SweepInterval time.Duration `yaml:"sweep_interval"` // SweepInterval is the minimum time between two eviction sweeps. Default 1h; a cadence has no "off".
}
// DefaultCache returns the cache eviction defaults, seeded before the file is read so a
// written 0 can mean off. SizeLimit stays zero: the free space floor bounds the whole cache.
func DefaultCache() Cache {
return Cache{
Retention: 7 * 24 * time.Hour,
RepoSizeLimit: 10 * 1024 * 1024 * 1024,
SweepInterval: time.Hour,
}
}
// Size is a byte count written the way people say it: 10GB, 512mb, 1TiB, or a plain number
// of bytes. Units are binary and case-insensitive, so GB and GiB both mean 1024³.
type Size int64
func (s *Size) UnmarshalYAML(value *yaml.Node) error {
if value.Kind != yaml.ScalarNode {
return fmt.Errorf("line %d: size must be a scalar such as 10GB", value.Line)
}
size, err := parseSize(value.Value)
if err != nil {
return fmt.Errorf("line %d: %w", value.Line, err)
}
*s = size
return nil
}
// parseSize reads a Size such as 10GB, 512mb, 1TiB or a plain byte count.
func parseSize(value string) (Size, error) {
bytes, err := units.RAMInBytes(strings.TrimSpace(value))
if err != nil {
return 0, fmt.Errorf("%q is not a size such as 10GB, 512MB or a plain byte count", value)
}
return Size(bytes), nil
} }
// Container represents the configuration for the container. // Container represents the configuration for the container.
@@ -101,6 +161,14 @@ type Container struct {
ServiceReadyTimeout time.Duration `yaml:"service_ready_timeout"` // ServiceReadyTimeout bounds how long a job waits for a service container that declares a healthcheck to report healthy. Negative disables waiting. ServiceReadyTimeout time.Duration `yaml:"service_ready_timeout"` // ServiceReadyTimeout bounds how long a job waits for a service container that declares a healthcheck to report healthy. Negative disables waiting.
} }
// Values of Runner.ToolCacheMode: the runner mounts no tool cache, or one that every job reuses.
const (
ToolCacheModeNone = "none"
ToolCacheModeShared = "shared"
)
var ToolCacheModes = []string{ToolCacheModeNone, ToolCacheModeShared}
type ContainerNetworkCreateOptions struct { type ContainerNetworkCreateOptions struct {
EnableIPv4 *bool `yaml:"enable_ipv4"` // Enable or disable IPv4 for the network (true for docker by default) EnableIPv4 *bool `yaml:"enable_ipv4"` // Enable or disable IPv4 for the network (true for docker by default)
EnableIPv6 *bool `yaml:"enable_ipv6"` // Enable or disable IPv6 for the network (false for docker by default) EnableIPv6 *bool `yaml:"enable_ipv6"` // Enable or disable IPv6 for the network (false for docker by default)
@@ -142,7 +210,8 @@ type Config struct {
// LoadDefault returns the default configuration. // LoadDefault returns the default configuration.
// If file is not empty, it will be used to load the configuration. // If file is not empty, it will be used to load the configuration.
func LoadDefault(file string) (*Config, error) { func LoadDefault(file string) (*Config, error) {
cfg := &Config{} // Seeded before the file is read, so a written 0 can mean off.
cfg := &Config{Cache: DefaultCache(), Log: Log{Job: LogJob{Retention: 7 * 24 * time.Hour, MaxSize: 1024 * 1024 * 1024}}}
definedRunnerKeys := map[string]bool{} definedRunnerKeys := map[string]bool{}
if file != "" { if file != "" {
content, err := os.ReadFile(file) content, err := os.ReadFile(file)
@@ -215,6 +284,15 @@ func LoadDefault(file string) (*Config, error) {
if cfg.Container.WorkdirParent == "" { if cfg.Container.WorkdirParent == "" {
cfg.Container.WorkdirParent = "workspace" cfg.Container.WorkdirParent = "workspace"
} }
if cfg.Runner.DefaultImage == "" {
cfg.Runner.DefaultImage = DefaultImage
}
if cfg.Runner.ToolCacheMode == "" {
cfg.Runner.ToolCacheMode = ToolCacheModeNone
}
if !slices.Contains(ToolCacheModes, cfg.Runner.ToolCacheMode) {
return nil, fmt.Errorf("invalid runner.tool_cache_mode %q: must be one of %q", cfg.Runner.ToolCacheMode, ToolCacheModes)
}
if cfg.Host.WorkdirParent == "" { if cfg.Host.WorkdirParent == "" {
home, err := os.UserHomeDir() home, err := os.UserHomeDir()
if err != nil { if err != nil {
@@ -272,6 +350,14 @@ func LoadDefault(file string) (*Config, error) {
} }
// Validate and fix invalid config combinations to prevent confusing behavior. // Validate and fix invalid config combinations to prevent confusing behavior.
if cfg.Runner.ToolCacheMode == ToolCacheModeShared && cfg.Runner.Capacity > 1 {
log.Warnf("runner.tool_cache_mode %q with capacity %d: two jobs writing the same tool version at once corrupt it",
ToolCacheModeShared, cfg.Runner.Capacity)
}
if cfg.Runner.FetchTimeout > RequestTimeout {
log.Warnf("fetch_timeout (%v) exceeds the RPC timeout (%v), capping it", cfg.Runner.FetchTimeout, RequestTimeout)
cfg.Runner.FetchTimeout = RequestTimeout
}
if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval { if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval {
log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval", log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval",
cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval) cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval)
+69
View File
@@ -42,11 +42,29 @@ cache:
require.NoError(t, err) require.NoError(t, err)
} }
func TestLoadDefault_ToolCacheMode(t *testing.T) {
cfg, err := LoadDefault("")
require.NoError(t, err)
assert.Equal(t, ToolCacheModeNone, cfg.Runner.ToolCacheMode)
path := filepath.Join(t.TempDir(), "config.yaml")
require.NoError(t, os.WriteFile(path, []byte("runner:\n tool_cache_mode: shared\n"), 0o600))
cfg, err = LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, ToolCacheModeShared, cfg.Runner.ToolCacheMode)
require.NoError(t, os.WriteFile(path, []byte("runner:\n tool_cache_mode: everyone\n"), 0o600))
_, err = LoadDefault(path)
require.Error(t, err)
assert.Contains(t, err.Error(), "tool_cache_mode")
}
func TestLoadDefault_DefaultsWorkdirCleanupAge(t *testing.T) { func TestLoadDefault_DefaultsWorkdirCleanupAge(t *testing.T) {
cfg, err := LoadDefault("") cfg, err := LoadDefault("")
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, 24*time.Hour, cfg.Runner.WorkdirCleanupAge) assert.Equal(t, 24*time.Hour, cfg.Runner.WorkdirCleanupAge)
assert.Equal(t, 10*time.Minute, cfg.Runner.IdleCleanupInterval) assert.Equal(t, 10*time.Minute, cfg.Runner.IdleCleanupInterval)
assert.Equal(t, DefaultImage, cfg.Runner.DefaultImage)
} }
func TestLoadDefault_HealthChecksAreOptIn(t *testing.T) { func TestLoadDefault_HealthChecksAreOptIn(t *testing.T) {
@@ -168,6 +186,45 @@ runner:
assert.Equal(t, 5*time.Minute, cfg.Runner.PostTaskScriptTimeout) assert.Equal(t, 5*time.Minute, cfg.Runner.PostTaskScriptTimeout)
} }
func write(t *testing.T, body string) string {
t.Helper()
path := filepath.Join(t.TempDir(), "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(body), 0o600))
return path
}
func TestLoadDefault_LoadsCacheEviction(t *testing.T) {
t.Run("sizes accept any spelling of the unit", func(t *testing.T) {
cfg, err := LoadDefault(write(t, "cache:\n retention: 336h\n repo_size_limit: 50gb\n size_limit: 1TiB\n sweep_interval: 15m\n"))
require.NoError(t, err)
assert.Equal(t, 336*time.Hour, cfg.Cache.Retention)
assert.Equal(t, Size(50*1024*1024*1024), cfg.Cache.RepoSizeLimit)
assert.Equal(t, Size(1024*1024*1024*1024), cfg.Cache.SizeLimit)
assert.Equal(t, 15*time.Minute, cfg.Cache.SweepInterval)
})
t.Run("zero turns a limit off where an absent key keeps its default", func(t *testing.T) {
cfg, err := LoadDefault(write(t, "cache:\n repo_size_limit: 0\n"))
require.NoError(t, err)
assert.Zero(t, cfg.Cache.RepoSizeLimit)
assert.Equal(t, DefaultCache().Retention, cfg.Cache.Retention, "an absent key still defaults")
})
t.Run("a bad size names the offending value", func(t *testing.T) {
_, err := LoadDefault(write(t, "cache:\n repo_size_limit: banana\n"))
require.Error(t, err)
assert.Contains(t, err.Error(), "banana")
})
}
func TestLoadDefault_LoadsJobLogs(t *testing.T) {
cfg, err := LoadDefault(write(t, "log:\n job:\n dir: /var/log/jobs\n retention: 0s\n max_size: 100MB\n"))
require.NoError(t, err)
assert.Equal(t, "/var/log/jobs", cfg.Log.Job.Dir)
assert.Zero(t, cfg.Log.Job.Retention, "zero keeps every task directory")
assert.Equal(t, Size(100*1024*1024), cfg.Log.Job.MaxSize)
}
func TestLoadDefault_LoadsJobHooks(t *testing.T) { func TestLoadDefault_LoadsJobHooks(t *testing.T) {
dir := t.TempDir() dir := t.TempDir()
path := filepath.Join(dir, "config.yaml") path := filepath.Join(dir, "config.yaml")
@@ -368,3 +425,15 @@ func TestLoadDefault_ShippedConfigsChangeNothing(t *testing.T) {
} }
assert.Empty(t, hook.AllEntries()) assert.Empty(t, hook.AllEntries())
} }
func TestLoadDefault_ClampsFetchTimeout(t *testing.T) {
path := filepath.Join(t.TempDir(), "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
fetch_timeout: 120s
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, RequestTimeout, cfg.Runner.FetchTimeout)
}
+11 -1
View File
@@ -26,7 +26,10 @@ const (
kindSection kindSection
) )
var durationType = reflect.TypeFor[time.Duration]() var (
durationType = reflect.TypeFor[time.Duration]()
sizeType = reflect.TypeFor[Size]()
)
// GetValue renders a flat list or mapping one entry per line, and anything nested as YAML. // GetValue renders a flat list or mapping one entry per line, and anything nested as YAML.
func GetValue(file, path string) (string, error) { func GetValue(file, path string) (string, error) {
@@ -537,6 +540,13 @@ func scalarNode(typ reflect.Type, value string) (*yaml.Node, error) {
return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: duration.String()}, nil return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: duration.String()}, nil
} }
if typ == sizeType {
if _, err := parseSize(value); err != nil {
return nil, err
}
return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: value}, nil
}
switch typ.Kind() { switch typ.Kind() {
case reflect.Bool: case reflect.Bool:
parsed, err := strconv.ParseBool(value) parsed, err := strconv.ParseBool(value)
+7
View File
@@ -0,0 +1,7 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
// Package disk reports free space on the volume holding a path. Platforms without an
// implementation return an error, so callers treat the check as unavailable rather than
// as a full disk.
package disk
@@ -3,10 +3,11 @@
//go:build !aix && !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows //go:build !aix && !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows
package run package disk
import "fmt" import "fmt"
func freeDiskBytes(path string) (uint64, error) { // FreeBytes reports the space available to an unprivileged user on the volume holding path.
func FreeBytes(path string) (uint64, error) {
return 0, fmt.Errorf("free disk space checks are not supported for %s", path) return 0, fmt.Errorf("free disk space checks are not supported for %s", path)
} }
@@ -3,11 +3,12 @@
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris //go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
package run package disk
import "golang.org/x/sys/unix" import "golang.org/x/sys/unix"
func freeDiskBytes(path string) (uint64, error) { // FreeBytes reports the space available to an unprivileged user on the volume holding path.
func FreeBytes(path string) (uint64, error) {
var stat unix.Statfs_t var stat unix.Statfs_t
if err := unix.Statfs(path, &stat); err != nil { if err := unix.Statfs(path, &stat); err != nil {
return 0, err return 0, err
@@ -3,11 +3,12 @@
//go:build windows //go:build windows
package run package disk
import "golang.org/x/sys/windows" import "golang.org/x/sys/windows"
func freeDiskBytes(path string) (uint64, error) { // FreeBytes reports the space available to an unprivileged user on the volume holding path.
func FreeBytes(path string) (uint64, error) {
pathPtr, err := windows.UTF16PtrFromString(path) pathPtr, err := windows.UTF16PtrFromString(path)
if err != nil { if err != nil {
return 0, err return 0, err
+6 -13
View File
@@ -11,6 +11,9 @@ import (
const ( const (
SchemeHost = "host" SchemeHost = "host"
SchemeDocker = "docker" SchemeDocker = "docker"
// SelfHostedPlatform is the platform marker act treats as "run on the host".
SelfHostedPlatform = "-self-hosted"
) )
type Label struct { type Label struct {
@@ -61,6 +64,7 @@ func (l Labels) RequireDocker() bool {
return false return false
} }
// PickPlatform returns the platform of the first runs-on entry this runner has a label for, or "".
func (l Labels) PickPlatform(runsOn []string) string { func (l Labels) PickPlatform(runsOn []string) string {
platforms := make(map[string]string, len(l)) platforms := make(map[string]string, len(l))
for _, label := range l { for _, label := range l {
@@ -69,7 +73,7 @@ func (l Labels) PickPlatform(runsOn []string) string {
// "//" will be ignored // "//" will be ignored
platforms[label.Name] = strings.TrimPrefix(label.Arg, "//") platforms[label.Name] = strings.TrimPrefix(label.Arg, "//")
case SchemeHost: case SchemeHost:
platforms[label.Name] = "-self-hosted" platforms[label.Name] = SelfHostedPlatform
default: default:
// unreachable: Parse only produces host or docker schemas // unreachable: Parse only produces host or docker schemas
continue continue
@@ -80,18 +84,7 @@ func (l Labels) PickPlatform(runsOn []string) string {
return v return v
} }
} }
return ""
// TODO: support multiple labels
// like:
// ["ubuntu-22.04"] => "ubuntu:22.04"
// ["with-gpu"] => "linux:with-gpu"
// ["ubuntu-22.04", "with-gpu"] => "ubuntu:22.04_with-gpu"
// return default.
// So the runner receives a task with a label that the runner doesn't have,
// it happens when the user have edited the label of the runner in the web UI.
// TODO: it may be not correct, what if the runner is used as host mode only?
return "docker.gitea.com/runner-images:ubuntu-latest"
} }
func (l Labels) Names() []string { func (l Labels) Names() []string {
+5 -5
View File
@@ -123,10 +123,10 @@ func TestPickPlatform(t *testing.T) {
want string want string
}{ }{
{"docker strips leading slashes", []string{"ubuntu"}, "node:18"}, {"docker strips leading slashes", []string{"ubuntu"}, "node:18"},
{"host maps to self-hosted marker", []string{"self-hosted"}, "-self-hosted"}, {"host maps to self-hosted marker", []string{"self-hosted"}, SelfHostedPlatform},
{"first match wins", []string{"self-hosted", "ubuntu"}, "-self-hosted"}, {"first match wins", []string{"self-hosted", "ubuntu"}, SelfHostedPlatform},
{"unknown falls back to default", []string{"windows"}, "docker.gitea.com/runner-images:ubuntu-latest"}, {"unknown label picks nothing", []string{"windows"}, ""},
{"no runsOn falls back to default", nil, "docker.gitea.com/runner-images:ubuntu-latest"}, {"no runsOn picks nothing", nil, ""},
} }
for _, tt := range tests { for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) { t.Run(tt.name, func(t *testing.T) {
@@ -167,5 +167,5 @@ func TestOpaqueLabelRoundTrip(t *testing.T) {
require.Equal(t, ls, again) require.Equal(t, ls, again)
require.Equal(t, []string{raw}, again.Names()) require.Equal(t, []string{raw}, again.Names())
require.False(t, again.RequireDocker()) require.False(t, again.RequireDocker())
require.Equal(t, "-self-hosted", again.PickPlatform([]string{raw})) require.Equal(t, SelfHostedPlatform, again.PickPlatform([]string{raw}))
} }
+120
View File
@@ -0,0 +1,120 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package report
import (
"fmt"
"os"
"path/filepath"
"strings"
"time"
"gitea.com/gitea/runner/internal/pkg/config"
log "github.com/sirupsen/logrus"
)
const (
jobLogNameLayout = "20060102-150405"
jobLogTimestamp = "2006-01-02T15:04:05.000Z"
)
// jobLog is this task's copy of the rows sent to Gitea. A nil *jobLog is a no-op, and every
// caller holds Reporter.stateMu, so it needs no lock.
type jobLog struct {
file *os.File
size int64
max int64
stopped bool // the cap was reached or a write failed, only the trailer still follows
closed bool
}
// openJobLog returns nil when the logs are off or cannot be created: a copy must never fail a job.
func openJobLog(cfg config.LogJob, taskID int64, started time.Time) *jobLog {
if cfg.Dir == "" {
return nil
}
if err := os.MkdirAll(cfg.Dir, 0o700); err != nil { // repository output, readable by this user only
log.Warnf("cannot create job log directory %s: %v", cfg.Dir, err)
return nil
}
pruneJobLogs(cfg.Dir, cfg.Retention, started)
name := fmt.Sprintf("%s-task-%d.log", started.UTC().Format(jobLogNameLayout), taskID)
file, err := os.OpenFile(filepath.Join(cfg.Dir, name), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
if err != nil {
log.Warnf("cannot create job log: %v", err)
return nil
}
log.Infof("writing the log of task %d to %s", taskID, file.Name())
return &jobLog{file: file, max: int64(cfg.MaxSize)}
}
func (j *jobLog) write(t time.Time, content string) {
if j == nil || j.stopped || j.closed {
return
}
line := t.UTC().Format(jobLogTimestamp) + " " + content
if j.max > 0 && j.size+int64(len(line))+1 > j.max {
j.stopped = true
j.line(runnerLine(fmt.Sprintf("truncated: log.job.max_size of %d bytes reached", j.max)))
return
}
j.line(line)
}
func (j *jobLog) close(trailer string) {
if j == nil || j.closed {
return
}
j.closed = true
j.line(runnerLine(trailer)) // past the cap on purpose: no trailer means the runner died mid-job
if err := j.file.Close(); err != nil {
log.Warnf("cannot close %s: %v", j.file.Name(), err)
}
}
// line writes unbuffered, so a killed runner keeps what it had written. Only the runner's own
// lines can carry a newline, a row reaching Gitea cannot (see DEVELOPMENT.md).
func (j *jobLog) line(content string) {
n, err := j.file.WriteString(strings.ReplaceAll(content, "\n", `\n`) + "\n")
j.size += int64(n)
if err != nil {
j.stopped = true // reported once, a failing write is a full disk and retrying floods the log
log.Warnf("cannot write %s: %v", j.file.Name(), err)
}
}
func runnerLine(content string) string {
return time.Now().UTC().Format(jobLogTimestamp) + " [runner] " + content
}
// pruneJobLogs removes the logs older than retention. The age comes from the name, not the
// mtime, which a reader or a backup tool can move.
func pruneJobLogs(root string, retention time.Duration, now time.Time) {
if retention <= 0 {
return
}
entries, err := os.ReadDir(root)
if err != nil {
log.Warnf("cannot list job log directory %s: %v", root, err)
return
}
cutoff := now.Add(-retention)
for _, entry := range entries {
stamp, _, isTaskLog := strings.Cut(entry.Name(), "-task-")
if !isTaskLog || entry.IsDir() || !strings.HasSuffix(entry.Name(), ".log") {
continue
}
if started, err := time.Parse(jobLogNameLayout, stamp); err != nil || !started.Before(cutoff) {
continue
}
name := filepath.Join(root, entry.Name())
if err := os.Remove(name); err != nil {
log.Warnf("cannot remove expired job log %s: %v", name, err)
}
}
}
+110
View File
@@ -0,0 +1,110 @@
// Copyright 2026 The Gitea Authors. All rights reserved.
// SPDX-License-Identifier: MIT
package report
import (
"context"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.com/gitea/runner/internal/pkg/client/mocks"
"gitea.com/gitea/runner/internal/pkg/config"
connect_go "connectrpc.com/connect"
runnerv1 "gitea.dev/actionslib/runner/v1"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/structpb"
)
var testStart = time.Date(2026, 8, 14, 9, 12, 3, 0, time.UTC)
func readJobLog(t *testing.T, joblog *jobLog) string {
t.Helper()
content, err := os.ReadFile(joblog.file.Name())
require.NoError(t, err)
return string(content)
}
func TestJobLog_MirrorsUploadedRows(t *testing.T) {
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil
})
client.On("UpdateTask", mock.Anything, mock.Anything).Return(connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil)
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.Log.Job.Dir = t.TempDir()
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
task := &runnerv1.Task{Id: 41, Context: &structpb.Struct{}, Secrets: map[string]string{"TOKEN": "s3cret-value"}}
reporter := NewReporter(ctx, cancel, client, task, cfg)
require.NotNil(t, reporter.jobLog)
reporter.RunDaemon()
reporter.ResetSteps(1)
fire := func(message string) {
require.NoError(t, reporter.Fire(&log.Entry{
Message: message,
Level: log.InfoLevel,
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
}
fire("the token is s3cret-value")
fire("::add-mask::dyn4mic-value")
fire("and dyn4mic-value too")
fire("::debug::suppressed unless ACTIONS_STEP_DEBUG")
require.NoError(t, reporter.Close(""))
job := readJobLog(t, reporter.jobLog)
assert.Contains(t, job, "the token is ***")
assert.Contains(t, job, "and *** too")
assert.NotContains(t, job, "s3cret-value")
assert.NotContains(t, job, "dyn4mic-value")
assert.NotContains(t, job, "add-mask", "the row carrying the secret never reaches the log")
assert.NotContains(t, job, "suppressed unless", "the file holds only what was sent")
assert.NotRegexp(t, `(?m)^\S+Z ?$`, job, "the empty row Gitea needs is not job output")
assert.Contains(t, job, "[runner] task 41 finished: failure")
}
func TestJobLog_MaxSize(t *testing.T) {
joblog := openJobLog(config.LogJob{Dir: t.TempDir(), MaxSize: 200}, 1, testStart)
require.NotNil(t, joblog)
for range 10 {
joblog.write(testStart, strings.Repeat("x", 40))
}
joblog.close("task 1 finished: success")
joblog.write(testStart, "after the close") // a container goroutine can outlive the step
job := readJobLog(t, joblog)
assert.Equal(t, 1, strings.Count(job, "log.job.max_size"), "the cap is reported once")
assert.NotContains(t, job, "after the close")
assert.Contains(t, job, "[runner] task 1 finished: success", "the trailer is written past the cap")
}
func TestPruneJobLogs(t *testing.T) {
root := t.TempDir()
expired := filepath.Join(root, "20200101-000000-task-1.log")
fresh := filepath.Join(root, testStart.Format(jobLogNameLayout)+"-task-2.log")
unrelated := filepath.Join(root, "20200101-000000-task-3.txt")
for _, name := range []string{expired, fresh, unrelated} {
require.NoError(t, os.WriteFile(name, []byte("log"), 0o600))
}
pruneJobLogs(root, 0, testStart)
assert.FileExists(t, expired, "retention 0 keeps every log")
pruneJobLogs(root, 24*time.Hour, testStart)
assert.NoFileExists(t, expired)
assert.FileExists(t, fresh)
assert.FileExists(t, unrelated)
}
+95 -45
View File
@@ -26,6 +26,9 @@ import (
"google.golang.org/protobuf/types/known/timestamppb" "google.golang.org/protobuf/types/known/timestamppb"
) )
// errOutputsNotSent travels the same return path as transport failures but is not one.
var errOutputsNotSent = errors.New("there are still outputs that have not been sent")
// Size limits for the outputs reported to the server. // Size limits for the outputs reported to the server.
const ( const (
maxOutputKeyLen = 255 maxOutputKeyLen = 255
@@ -56,8 +59,13 @@ type Reporter struct {
// so the gauge skips no-op Set calls when the buffer size is unchanged. // so the gauge skips no-op Set calls when the buffer size is unchanged.
lastLogBufferRows int lastLogBufferRows int
state *runnerv1.TaskState state *runnerv1.TaskState
stateChanged bool stateChanged bool
// reportFailing keeps an outage to one log line at each end.
reportFailing map[string]bool
// serverResult is what the server decided, e.g. the zombie reaper failing
// the task. Guarded by stateMu.
serverResult runnerv1.Result
stateMu sync.RWMutex stateMu sync.RWMutex
outputsMu sync.Mutex outputsMu sync.Mutex
outputs map[string]jobOutput outputs map[string]jobOutput
@@ -77,6 +85,8 @@ type Reporter struct {
// closeTimeout bounds each RPC attempt in the final flush, on a context // closeTimeout bounds each RPC attempt in the final flush, on a context
// detached from r.ctx so a server cancel can't abort the acknowledgement. // detached from r.ctx so a server cancel can't abort the acknowledgement.
closeTimeout time.Duration closeTimeout time.Duration
// daemonWait bounds how long Close waits for the daemon loop to acknowledge.
daemonWait time.Duration
// Event notification channels (non-blocking, buffered 1) // Event notification channels (non-blocking, buffered 1)
logNotify chan struct{} // signal: new log rows arrived logNotify chan struct{} // signal: new log rows arrived
@@ -84,6 +94,8 @@ type Reporter struct {
debugOutputEnabled bool debugOutputEnabled bool
stopCommandEndToken string stopCommandEndToken string
jobLog *jobLog // this task's rows on the runner's own disk, nil when log.job.dir is unset
} }
// extraMasks are values known before the job starts that are not among its secrets, such as // extraMasks are values known before the job starts that are not among its secrets, such as
@@ -119,10 +131,14 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C
state: &runnerv1.TaskState{ state: &runnerv1.TaskState{
Id: task.Id, Id: task.Id,
}, },
reportFailing: map[string]bool{},
daemon: make(chan struct{}), daemon: make(chan struct{}),
heartbeatStop: make(chan struct{}), heartbeatStop: make(chan struct{}),
jobLog: openJobLog(cfg.Log.Job, task.Id, time.Now()),
} }
rv.daemonWait = 6 * rv.effectiveCloseTimeout()
if task.Secrets["ACTIONS_STEP_DEBUG"] == "true" { if task.Secrets["ACTIONS_STEP_DEBUG"] == "true" {
rv.debugOutputEnabled = true rv.debugOutputEnabled = true
} }
@@ -151,11 +167,14 @@ func (r *Reporter) Levels() []log.Level {
return log.AllLevels return log.AllLevels
} }
func appendIfNotNil[T any](s []*T, v *T) []*T { // appendLogRow buffers a row for the uploader and mirrors it into job.log. A nil row is one
if v != nil { // the command handler dropped, such as ::add-mask::. Caller holds stateMu.
return append(s, v) func (r *Reporter) appendLogRow(row *runnerv1.LogRow) {
if row == nil {
return
} }
return s r.logRows = append(r.logRows, row)
r.jobLog.write(row.Time.AsTime(), row.Content)
} }
// isJobStepEntry is used to not report composite step results incorrectly as step result // isJobStepEntry is used to not report composite step results incorrectly as step result
@@ -232,7 +251,7 @@ func (r *Reporter) Fire(entry *log.Entry) error {
} }
} }
if r.shouldAppendLogRow(entry) { if r.shouldAppendLogRow(entry) {
r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry)) r.appendLogRow(r.parseLogRow(entry))
} }
r.unlockAndNotify(urgentState) r.unlockAndNotify(urgentState)
return nil return nil
@@ -246,7 +265,7 @@ func (r *Reporter) Fire(entry *log.Entry) error {
} }
if step == nil { if step == nil {
if r.shouldAppendLogRow(entry) { if r.shouldAppendLogRow(entry) {
r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry)) r.appendLogRow(r.parseLogRow(entry))
} }
r.unlockAndNotify(false) r.unlockAndNotify(false)
return nil return nil
@@ -272,11 +291,11 @@ func (r *Reporter) Fire(entry *log.Entry) error {
step.LogIndex = int64(r.logOffset + len(r.logRows)) step.LogIndex = int64(r.logOffset + len(r.logRows))
} }
step.LogLength++ step.LogLength++
r.logRows = append(r.logRows, row) r.appendLogRow(row)
} }
} }
} else if r.shouldAppendLogRow(entry) { } else if r.shouldAppendLogRow(entry) {
r.logRows = appendIfNotNil(r.logRows, r.parseLogRow(entry)) r.appendLogRow(r.parseLogRow(entry))
} }
if v, ok := entry.Data["stepResult"]; ok && isJobStepEntry(entry) { if v, ok := entry.Data["stepResult"]; ok && isJobStepEntry(entry) {
if stepResult, ok := r.parseResult(v); ok { if stepResult, ok := r.parseResult(v); ok {
@@ -293,6 +312,21 @@ func (r *Reporter) Fire(entry *log.Entry) error {
return nil return nil
} }
// Only the daemon loop calls this, so reportFailing needs no lock.
func (r *Reporter) noteReport(method string, err error) {
if errors.Is(err, errOutputsNotSent) {
err = nil // the RPC itself succeeded
}
switch {
case err != nil && !r.reportFailing[method]:
r.reportFailing[method] = true
log.Warnf("%s error: %v, retrying until reconnected", method, err)
case err == nil && r.reportFailing[method]:
delete(r.reportFailing, method)
log.Infof("%s reconnected", method)
}
}
func (r *Reporter) RunDaemon() { func (r *Reporter) RunDaemon() {
go r.runDaemonLoop() go r.runDaemonLoop()
} }
@@ -319,6 +353,8 @@ func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) {
} }
func (r *Reporter) runDaemonLoop() { func (r *Reporter) runDaemonLoop() {
defer close(r.daemon)
logTicker := time.NewTicker(r.logReportInterval) logTicker := time.NewTicker(r.logReportInterval)
stateTicker := time.NewTicker(r.stateReportInterval) stateTicker := time.NewTicker(r.stateReportInterval)
@@ -337,11 +373,11 @@ func (r *Reporter) runDaemonLoop() {
for { for {
select { select {
case <-logTicker.C: case <-logTicker.C:
_ = r.ReportLog(false) r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer) r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
case <-stateTicker.C: case <-stateTicker.C:
_ = r.ReportState(false) r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
case <-r.logNotify: case <-r.logNotify:
r.stateMu.RLock() r.stateMu.RLock()
@@ -349,7 +385,7 @@ func (r *Reporter) runDaemonLoop() {
r.stateMu.RUnlock() r.stateMu.RUnlock()
if n >= r.logBatchSize { if n >= r.logBatchSize {
_ = r.ReportLog(false) r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer) r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
} else if !maxLatencyActive && n > 0 { } else if !maxLatencyActive && n > 0 {
maxLatencyTimer.Reset(r.logReportMaxLatency) maxLatencyTimer.Reset(r.logReportMaxLatency)
@@ -358,25 +394,23 @@ func (r *Reporter) runDaemonLoop() {
case <-r.stateNotify: case <-r.stateNotify:
// Step transition or job result — flush both immediately for frontend UX. // Step transition or job result — flush both immediately for frontend UX.
_ = r.ReportLog(false) r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
_ = r.ReportState(false) r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer) r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
case <-maxLatencyTimer.C: case <-maxLatencyTimer.C:
maxLatencyActive = false maxLatencyActive = false
_ = r.ReportLog(false) r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
case <-r.ctx.Done(): case <-r.ctx.Done():
// Stop heartbeating on cancel so Gitea sees the runner as offline // Stop heartbeating on cancel so Gitea sees the runner as offline
// during cleanup and won't assign an overlapping task. Close() still // during cleanup and won't assign an overlapping task. Close() still
// delivers the final flush on a detached context (flushFinal). // delivers the final flush on a detached context (flushFinal).
close(r.daemon)
return return
case <-r.heartbeatStop: case <-r.heartbeatStop:
// Stop heartbeating during post-task script execution. Close() still // Stop heartbeating during post-task script execution. Close() still
// delivers the final flush on a detached context (flushFinal). // delivers the final flush on a detached context (flushFinal).
close(r.daemon)
return return
} }
@@ -384,7 +418,6 @@ func (r *Reporter) runDaemonLoop() {
closed := r.closed closed := r.closed
r.stateMu.RUnlock() r.stateMu.RUnlock()
if closed { if closed {
close(r.daemon)
return return
} }
} }
@@ -401,7 +434,7 @@ func (r *Reporter) logf(format string, a ...any) {
if !r.duringSteps() { if !r.duringSteps() {
// Masked like any other row: these bypass parseLogRow, but a caller can still // Masked like any other row: these bypass parseLogRow, but a caller can still
// interpolate a secret, such as a configured URL carrying credentials. // interpolate a secret, such as a configured URL carrying credentials.
r.logRows = append(r.logRows, r.newLogRow(timestamppb.Now(), fmt.Sprintf(format, a...))) r.appendLogRow(r.newLogRow(timestamppb.Now(), fmt.Sprintf(format, a...)))
} }
} }
@@ -435,35 +468,30 @@ func (r *Reporter) Close(lastWords string) error {
r.stateMu.Lock() r.stateMu.Lock()
r.closed = true r.closed = true
if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED { if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED {
// When r.ctx has been cancelled (server returned RESULT_CANCELLED via // No result of its own, so say why it stopped.
// rpcCtx/ReportState, see line 590) the job is being torn down on the result, words := runnerv1.Result_RESULT_FAILURE, "Early termination"
// cancellation path: surface that explicitly instead of attributing it switch {
// to a generic failure. case r.serverResult != runnerv1.Result_RESULT_UNSPECIFIED:
cancelled := errors.Is(r.ctx.Err(), context.Canceled) result, words = r.serverResult, "Ended by the server"
case errors.Is(r.ctx.Err(), context.Canceled):
result, words = runnerv1.Result_RESULT_CANCELLED, "Cancelled"
}
if lastWords == "" { if lastWords == "" {
if cancelled { lastWords = words
lastWords = "Cancelled"
} else {
lastWords = "Early termination"
}
} }
for _, v := range r.state.Steps { for _, v := range r.state.Steps {
if v.Result == runnerv1.Result_RESULT_UNSPECIFIED { if v.Result == runnerv1.Result_RESULT_UNSPECIFIED {
v.Result = runnerv1.Result_RESULT_CANCELLED v.Result = runnerv1.Result_RESULT_CANCELLED
} }
} }
if cancelled { r.state.Result = result
r.state.Result = runnerv1.Result_RESULT_CANCELLED r.appendLogRow(&runnerv1.LogRow{
} else {
r.state.Result = runnerv1.Result_RESULT_FAILURE
}
r.logRows = append(r.logRows, &runnerv1.LogRow{
Time: timestamppb.Now(), Time: timestamppb.Now(),
Content: lastWords, Content: lastWords,
}) })
r.state.StoppedAt = timestamppb.Now() r.state.StoppedAt = timestamppb.Now()
} else if lastWords != "" { } else if lastWords != "" {
r.logRows = append(r.logRows, &runnerv1.LogRow{ r.appendLogRow(&runnerv1.LogRow{
Time: timestamppb.Now(), Time: timestamppb.Now(),
Content: lastWords, Content: lastWords,
}) })
@@ -476,9 +504,8 @@ func (r *Reporter) Close(lastWords string) error {
// Wait for Acknowledge // Wait for Acknowledge
select { select {
case <-r.daemon: case <-r.daemon:
case <-time.After(60 * time.Second): case <-time.After(r.daemonWait):
close(r.daemon) log.Errorf("No Response from RunDaemon for %s, continue best effort", r.daemonWait)
log.Error("No Response from RunDaemon for 60s, continue best effort")
} }
// Gitea's UpdateLog short-circuits on len(Rows)==0 before honoring NoMore, // Gitea's UpdateLog short-circuits on len(Rows)==0 before honoring NoMore,
@@ -489,6 +516,7 @@ func (r *Reporter) Close(lastWords string) error {
// supported branches, e.g. v1.28+. // supported branches, e.g. v1.28+.
r.stateMu.Lock() r.stateMu.Lock()
if len(r.logRows) == 0 { if len(r.logRows) == 0 {
// Not appendLogRow: the sentinel is not job output and has no place in job.log.
r.logRows = append(r.logRows, &runnerv1.LogRow{ r.logRows = append(r.logRows, &runnerv1.LogRow{
Time: timestamppb.Now(), Time: timestamppb.Now(),
Content: "", Content: "",
@@ -498,10 +526,21 @@ func (r *Reporter) Close(lastWords string) error {
// Separate budgets so a slow ReportLog can't starve the ReportState that // Separate budgets so a slow ReportLog can't starve the ReportState that
// carries the cancel acknowledgement. // carries the cancel acknowledgement.
return errors.Join( err := errors.Join(
r.flushFinal(func() error { return r.ReportLog(true) }), r.flushFinal(func() error { return r.ReportLog(true) }),
r.flushFinal(func() error { return r.ReportState(true) }), r.flushFinal(func() error { return r.ReportState(true) }),
) )
// After the flush so a failed handover is in the file too, under stateMu so a late entry cannot race.
r.stateMu.Lock()
trailer := fmt.Sprintf("task %d finished: %s", r.state.Id, metrics.ResultToStatusLabel(r.state.Result))
if err != nil {
trailer += fmt.Sprintf(", the final flush to Gitea failed: %v", err)
}
r.jobLog.close(r.mask(trailer))
r.stateMu.Unlock()
return err
} }
// flushFinal retries fn on a detached, bounded context so a cancelled r.ctx // flushFinal retries fn on a detached, bounded context so a cancelled r.ctx
@@ -569,8 +608,10 @@ func (r *Reporter) ReportLog(noMore bool) error {
} }
r.stateMu.Lock() r.stateMu.Lock()
r.logRows = r.logRows[ack-r.logOffset:]
submitted := r.logOffset + len(rows) submitted := r.logOffset + len(rows)
// A server can ack beyond what it was sent; clamp to stay within the buffer.
ack = min(ack, submitted)
r.logRows = r.logRows[ack-r.logOffset:]
r.logOffset = ack r.logOffset = ack
remaining := len(r.logRows) remaining := len(r.logRows)
r.stateMu.Unlock() r.stateMu.Unlock()
@@ -655,11 +696,16 @@ func (r *Reporter) ReportState(reportResult bool) error {
} }
r.outputsMu.Unlock() r.outputsMu.Unlock()
if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED { // A terminal result means the server is done with this task; keep running and
// the job holds a capacity slot until runner.timeout.
if state := resp.Msg.State; state != nil && state.Result != runnerv1.Result_RESULT_UNSPECIFIED {
r.stateMu.Lock()
r.serverResult = state.Result
r.stateMu.Unlock()
r.cancel() r.cancel()
} }
if len(noSent) > 0 { if len(noSent) > 0 {
return fmt.Errorf("there are still outputs that have not been sent: %v", noSent) return fmt.Errorf("%w: %v", errOutputsNotSent, noSent)
} }
return nil return nil
@@ -823,10 +869,14 @@ func (r *Reporter) parseLogRow(entry *log.Entry) *runnerv1.LogRow {
func (r *Reporter) newLogRow(t *timestamppb.Timestamp, content string) *runnerv1.LogRow { func (r *Reporter) newLogRow(t *timestamppb.Timestamp, content string) *runnerv1.LogRow {
return &runnerv1.LogRow{ return &runnerv1.LogRow{
Time: t, Time: t,
Content: strings.ToValidUTF8(r.logReplacer.Replace(content), "?"), Content: r.mask(content),
} }
} }
func (r *Reporter) mask(content string) string {
return strings.ToValidUTF8(r.logReplacer.Replace(content), "?")
}
func (r *Reporter) addMask(msg string) { func (r *Reporter) addMask(msg string) {
r.oldnew = runner.AppendSecretMasker(r.oldnew, msg) r.oldnew = runner.AppendSecretMasker(r.oldnew, msg)
r.logReplacer = strings.NewReplacer(r.oldnew...) r.logReplacer = strings.NewReplacer(r.oldnew...)
+139 -12
View File
@@ -12,6 +12,7 @@ import (
"net/url" "net/url"
"slices" "slices"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"testing" "testing"
"time" "time"
@@ -19,10 +20,12 @@ import (
"gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/act/runner"
"gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/client/mocks"
"gitea.com/gitea/runner/internal/pkg/config" "gitea.com/gitea/runner/internal/pkg/config"
"gitea.com/gitea/runner/internal/pkg/metrics"
connect_go "connectrpc.com/connect" connect_go "connectrpc.com/connect"
runnerv1 "gitea.dev/actionslib/runner/v1" runnerv1 "gitea.dev/actionslib/runner/v1"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
logrustest "github.com/sirupsen/logrus/hooks/test"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock" "github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@@ -1038,18 +1041,13 @@ func TestReporter_StopHeartbeats(t *testing.T) {
"Close() must still send a final UpdateTask after StopHeartbeats") "Close() must still send a final UpdateTask after StopHeartbeats")
} }
func TestAppendIfNotNil(t *testing.T) { func TestAppendLogRow(t *testing.T) {
var s []*int r := &Reporter{}
s = appendIfNotNil(s, nil) row := &runnerv1.LogRow{Time: timestamppb.Now(), Content: "hello"}
assert.Empty(t, s) r.appendLogRow(nil)
r.appendLogRow(row)
v := 7 r.appendLogRow(nil)
s = appendIfNotNil(s, &v) assert.Equal(t, []*runnerv1.LogRow{row}, r.logRows)
require.Len(t, s, 1)
assert.Equal(t, &v, s[0])
s = appendIfNotNil(s, nil)
require.Len(t, s, 1)
} }
func TestReporter_Levels(t *testing.T) { func TestReporter_Levels(t *testing.T) {
@@ -1169,3 +1167,132 @@ func TestReporter_masksEncodedSecrets(t *testing.T) {
assert.NotContains(t, row.Content, base64.StdEncoding.EncodeToString([]byte(secret))) assert.NotContains(t, row.Content, base64.StdEncoding.EncodeToString([]byte(secret)))
} }
} }
// A server that acknowledges more rows than were sent must not take the runner
// process down with it.
func TestReporter_AckIndexBeyondBuffer(t *testing.T) {
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: 1000}), nil
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(1)
require.NoError(t, reporter.Fire(&log.Entry{
Message: "hello",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: log.InfoLevel,
}))
require.NoError(t, reporter.ReportLog(false))
}
// Close has to survive giving up on a daemon still parked in an RPC.
func TestReporter_CloseWithStuckDaemon(t *testing.T) {
release := make(chan struct{})
entered := make(chan struct{})
var once sync.Once
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
once.Do(func() {
close(entered)
<-release
})
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
}), nil
})
client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
}).Maybe()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Millisecond
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.daemonWait = time.Millisecond
reporter.RunDaemon()
reporter.ResetSteps(1)
require.NoError(t, reporter.Fire(&log.Entry{
Message: "hello",
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
Level: log.InfoLevel,
}))
select {
case <-entered:
case <-time.After(5 * time.Second):
t.Fatal("daemon never reached UpdateLog")
}
go func() {
time.Sleep(10 * reporter.daemonWait)
close(release)
}()
require.NoError(t, reporter.Close(""))
}
// The zombie reaper marks a task failed, and cancelling the context is how the
// runner stops, so the two must not be conflated into "cancelled".
func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) {
var lastState *runnerv1.TaskState
client := mocks.NewClient(t)
client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
lastState = req.Msg.State
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{
State: &runnerv1.TaskState{Result: runnerv1.Result_RESULT_FAILURE},
}), nil
})
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
}), nil
}).Maybe()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
close(reporter.daemon) // no daemon loop to acknowledge
require.NoError(t, reporter.ReportState(false))
require.ErrorIs(t, ctx.Err(), context.Canceled)
require.NoError(t, reporter.Close(""))
require.NotNil(t, lastState)
assert.Equal(t, runnerv1.Result_RESULT_FAILURE, lastState.Result)
}
func TestReporter_NoteReport(t *testing.T) {
hook := logrustest.NewGlobal()
defer hook.Reset()
reporter := &Reporter{reportFailing: map[string]bool{}}
// A pending output is not a transport failure and must not read as one.
reporter.noteReport(metrics.LabelMethodUpdateTask, fmt.Errorf("wrapped: %w", errOutputsNotSent))
assert.Empty(t, hook.AllEntries())
// An outage logs once at each end, not every report interval for hours.
reporter.noteReport(metrics.LabelMethodUpdateTask, errors.New("connection refused"))
reporter.noteReport(metrics.LabelMethodUpdateTask, errors.New("connection refused"))
require.Len(t, hook.AllEntries(), 1)
assert.Contains(t, hook.LastEntry().Message, "connection refused")
reporter.noteReport(metrics.LabelMethodUpdateTask, nil)
require.Len(t, hook.AllEntries(), 2)
assert.Contains(t, hook.LastEntry().Message, "reconnected")
}
+4 -8
View File
@@ -6,13 +6,9 @@
# Cloudflare R2 bucket, using curl's built-in AWS SigV4 signer (R2 is # Cloudflare R2 bucket, using curl's built-in AWS SigV4 signer (R2 is
# S3-API compatible). # S3-API compatible).
# #
# This is the R2 half of the release process's parallel S3+R2 upload # It is invoked once per release artifact via a goreleaser
# period: goreleaser's `blobs:` pipe still uploads every release # `publishers:` entry, and is the only object storage upload of the
# artifact to AWS S3, and this script is invoked once per artifact # release process.
# (via a goreleaser `publishers:` entry) to mirror the same artifact
# into R2. Once the migration away from S3 is complete, the `blobs:`
# block and the AWS_* secrets can be dropped without touching this
# script.
# #
# Usage: # Usage:
# upload-r2.sh <local-file> <remote-key> # upload-r2.sh <local-file> <remote-key>
@@ -24,7 +20,7 @@
# preflight step in CI: goreleaser custom publishers run as the very # preflight step in CI: goreleaser custom publishers run as the very
# last step of the publish pipeline, so without a preflight check a # last step of the publish pipeline, so without a preflight check a
# missing R2_* secret would only be discovered after the Gitea release # missing R2_* secret would only be discovered after the Gitea release
# has already been created and every artifact already uploaded to S3. # has already been created.
# #
# Required environment variables: # Required environment variables:
# R2_ENDPOINT Base URL of the R2 endpoint, e.g. # R2_ENDPOINT Base URL of the R2 endpoint, e.g.