mirror of
https://gitea.com/gitea/runner.git
synced 2026-08-26 05:47:45 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c0a07cfb72 | |||
| 2fa5fe7121 | |||
| 825c6af07c |
@@ -20,10 +20,11 @@ jobs:
|
|||||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
|
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
|
||||||
with:
|
with:
|
||||||
fetch-depth: 0
|
fetch-depth: 0
|
||||||
# Custom publishers (the R2 upload below) run as the very last
|
# Custom publishers (the R2 mirror 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. Fail here instead, before anything
|
# has already been created and every artifact already uploaded
|
||||||
# is built or published, if the R2 secrets are missing.
|
# to S3. Fail here instead, before anything is built or
|
||||||
|
# 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,6 +43,11 @@ 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 }}
|
||||||
|
|||||||
@@ -12,10 +12,11 @@ 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 upload below) run as the very last
|
# Custom publishers (the R2 mirror 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. Fail here instead, before anything
|
# has already been created and every artifact already uploaded
|
||||||
# is built or published, if the R2 secrets are missing.
|
# to S3. Fail here instead, before anything is built or
|
||||||
|
# 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:
|
||||||
@@ -41,6 +42,11 @@ 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 }}
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ jobs:
|
|||||||
check-latest: true
|
check-latest: true
|
||||||
- name: prepare anonymous docker config
|
- name: prepare anonymous docker config
|
||||||
run: mkdir -p "$DOCKER_CONFIG" && echo '{}' > "$DOCKER_CONFIG/config.json"
|
run: mkdir -p "$DOCKER_CONFIG" && echo '{}' > "$DOCKER_CONFIG/config.json"
|
||||||
# Pre-pull act/runner's two largest base images so a slow pull can't dominate `make test`;
|
# Pre-pull internal/act/runner's two largest base images so a slow pull can't dominate `make test`;
|
||||||
# the rest (alpine/ubuntu) pull on demand, absorbed by the make-test -timeout. The host
|
# the rest (alpine/ubuntu) pull on demand, absorbed by the make-test -timeout. The host
|
||||||
# daemon retains them between runs, so this is usually a fast manifest re-check.
|
# daemon retains them between runs, so this is usually a fast manifest re-check.
|
||||||
- name: pre-pull test images
|
- name: pre-pull test images
|
||||||
|
|||||||
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
/gitea-runner
|
/gitea-runner
|
||||||
.env
|
.env
|
||||||
!/act/runner/testdata/secrets/.env
|
!/internal/act/runner/testdata/secrets/.env
|
||||||
.runner
|
.runner
|
||||||
.runner.lock
|
.runner.lock
|
||||||
coverage.txt
|
coverage.txt
|
||||||
|
|||||||
+19
-7
@@ -83,12 +83,24 @@ 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
|
||||||
|
|
||||||
# Uploads every release artifact to Cloudflare R2. The `blobs:` pipe
|
blobs:
|
||||||
# isn't usable here since it authenticates from the global AWS_* env
|
-
|
||||||
# with no per-entry credentials; `publishers:` supports per-entry
|
provider: s3
|
||||||
# `env:` instead, so it's used to invoke scripts/upload-r2.sh once per
|
bucket: "{{ .Env.S3_BUCKET }}"
|
||||||
# artifact. Custom publishers inherit almost nothing from the
|
region: "{{ .Env.S3_REGION }}"
|
||||||
# environment, hence the explicit R2_* forwarding below.
|
directory: "gitea-runner/{{.Version}}"
|
||||||
|
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`
|
||||||
@@ -113,7 +125,7 @@ publishers:
|
|||||||
- R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }}
|
- R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }}
|
||||||
|
|
||||||
archives:
|
archives:
|
||||||
- formats: [binary]
|
- format: binary
|
||||||
name_template: "{{ .Binary }}"
|
name_template: "{{ .Binary }}"
|
||||||
allow_different_binary_count: true
|
allow_different_binary_count: true
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -17,7 +17,7 @@ RUN make clean && make build
|
|||||||
### DIND VARIANT
|
### DIND VARIANT
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
FROM docker:29.7.1-dind AS dind
|
FROM docker:29.6.2-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.7.1-dind-rootless AS dind-rootless
|
FROM docker:29.6.2-dind-rootless AS dind-rootless
|
||||||
|
|
||||||
ARG VERSION=dev
|
ARG VERSION=dev
|
||||||
|
|
||||||
|
|||||||
@@ -158,32 +158,6 @@ 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.
|
||||||
@@ -299,12 +273,6 @@ 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:
|
||||||
@@ -345,8 +313,6 @@ 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.
|
||||||
|
|||||||
@@ -11,8 +11,6 @@ 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)
|
||||||
|
|||||||
@@ -56,7 +56,6 @@ 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: {}
|
||||||
|
|||||||
@@ -1,38 +0,0 @@
|
|||||||
// 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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -56,7 +56,6 @@ 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,7 +33,6 @@ 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: {}
|
||||||
|
|||||||
@@ -12,9 +12,8 @@ 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.7.1+incompatible
|
github.com/docker/cli v29.6.2+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
|
||||||
@@ -23,7 +22,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.3.2
|
github.com/moby/go-archive v0.2.1
|
||||||
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
|
||||||
@@ -62,6 +61,7 @@ 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
|
||||||
|
|||||||
@@ -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.7.1+incompatible h1:ILZpP6B7fedIr6ANy824QkDp1WMJuouIq0O2SrBkB2w=
|
github.com/docker/cli v29.6.2+incompatible h1:/bjePvcbbFTnRrMfWJBY7AjfICdsiLVgHn6LwTVOcqw=
|
||||||
github.com/docker/cli v29.7.1+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
|
github.com/docker/cli v29.6.2+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.3.2 h1:x893kC3zRygv2C+k4Y9kMxYRPLCj4XEJB0srbAP06Hw=
|
github.com/moby/go-archive v0.2.1 h1:fAa0wUS/ikZKyx7o/1fhUYmhZ7RgpthdeoDhJvunTLc=
|
||||||
github.com/moby/go-archive v0.3.2/go.mod h1:Npdv43fFqlhZW7Xo8fbm3ZMYFvAGNviUPqX21VERbcE=
|
github.com/moby/go-archive v0.2.1/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=
|
||||||
|
|||||||
@@ -5,7 +5,6 @@
|
|||||||
package artifactcache
|
package artifactcache
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"cmp"
|
|
||||||
"context"
|
"context"
|
||||||
"crypto/hmac"
|
"crypto/hmac"
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
@@ -21,15 +20,13 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"regexp"
|
"regexp"
|
||||||
"slices"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/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"
|
||||||
@@ -62,9 +59,6 @@ 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
|
||||||
@@ -106,38 +100,19 @@ 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.
|
// StartHandler opens the on-disk cache store and starts the HTTP server.
|
||||||
type Options struct {
|
//
|
||||||
Dir string
|
// internalSecret, when non-empty, enables a control-plane API at
|
||||||
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
|
// /_internal/{register,revoke} that lets a remote runner pre-register the
|
||||||
// per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. 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
|
// embedded in-process handler leaves it empty and registers tokens via the
|
||||||
// in-process RegisterJob method directly.
|
// in-process RegisterJob method directly.
|
||||||
InternalSecret string
|
func StartHandler(dir, outboundIP string, port uint16, internalSecret string, logger logrus.FieldLogger) (*Handler, error) {
|
||||||
|
|
||||||
Policy Policy
|
|
||||||
Logger logrus.FieldLogger
|
|
||||||
}
|
|
||||||
|
|
||||||
// StartHandler opens the on-disk cache store and starts the HTTP server.
|
|
||||||
func StartHandler(opts Options) (*Handler, error) {
|
|
||||||
dir, logger := opts.Dir, opts.Logger
|
|
||||||
h := &Handler{
|
h := &Handler{
|
||||||
creds: make(map[string]*credEntry),
|
creds: make(map[string]*credEntry),
|
||||||
internalSecret: opts.InternalSecret,
|
internalSecret: internalSecret,
|
||||||
policy: opts.Policy.withDefaults(),
|
|
||||||
freeDisk: disk.FreeBytes,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if logger == nil {
|
if logger == nil {
|
||||||
@@ -167,8 +142,8 @@ func StartHandler(opts Options) (*Handler, error) {
|
|||||||
}
|
}
|
||||||
h.storage = storage
|
h.storage = storage
|
||||||
|
|
||||||
if opts.OutboundIP != "" {
|
if outboundIP != "" {
|
||||||
h.outboundIP = opts.OutboundIP
|
h.outboundIP = 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 {
|
||||||
@@ -207,7 +182,7 @@ func StartHandler(opts Options) (*Handler, error) {
|
|||||||
// 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", opts.Port))
|
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -237,13 +212,6 @@ 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
|
||||||
@@ -391,7 +359,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(cred, cache.ID, time.Now().Add(artifactURLTTL)),
|
"archiveLocation": h.signedArtifactURL(cache.ID, time.Now().Add(artifactURLTTL)),
|
||||||
"cacheKey": cache.Key,
|
"cacheKey": cache.Key,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -412,9 +380,6 @@ 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
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -552,22 +517,13 @@ 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()
|
||||||
if err := db.Update(cache.ID, cache); err != nil {
|
return db.Update(cache.ID, cache)
|
||||||
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
|
||||||
@@ -685,7 +641,7 @@ func (h *Handler) ResultsURL(cred JobCredential) string {
|
|||||||
if h == nil || cred.Results == "" {
|
if h == nil || cred.Results == "" {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
return h.baseURL(cred)
|
return h.ExternalURL()
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
|
func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
|
||||||
@@ -744,16 +700,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(cred JobCredential, path, purpose string, cacheID uint64, exp time.Time) string {
|
func (h *Handler) signedURL(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.baseURL(cred), path, cacheID, q.Encode())
|
return fmt.Sprintf("%s%s/%d?%s", h.ExternalURL(), path, cacheID, q.Encode())
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) signedArtifactURL(cred JobCredential, cacheID uint64, exp time.Time) string {
|
func (h *Handler) signedArtifactURL(cacheID uint64, exp time.Time) string {
|
||||||
return h.signedURL(cred, apiPath+"/artifacts", "", cacheID, exp)
|
return h.signedURL(apiPath+"/artifacts", "", cacheID, exp)
|
||||||
}
|
}
|
||||||
|
|
||||||
// if not found, return (nil, nil) instead of an error.
|
// if not found, return (nil, nil) instead of an error.
|
||||||
@@ -855,43 +811,12 @@ func (h *Handler) touchCache(id uint64, requireIncomplete bool) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const (
|
const (
|
||||||
miB = 1024 * 1024
|
keepUsed = 30 * 24 * time.Hour
|
||||||
|
keepUnused = 7 * 24 * time.Hour
|
||||||
defaultSweepInterval = time.Hour
|
keepTemp = 5 * time.Minute
|
||||||
|
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
|
||||||
@@ -901,7 +826,7 @@ func (h *Handler) gcCache() {
|
|||||||
}
|
}
|
||||||
defer h.gcing.Store(false)
|
defer h.gcing.Store(false)
|
||||||
|
|
||||||
if time.Since(h.gcAt) < h.policy.SweepInterval {
|
if time.Since(h.gcAt) < time.Hour {
|
||||||
h.logger.Debugf("skip gc: %v", h.gcAt.String())
|
h.logger.Debugf("skip gc: %v", h.gcAt.String())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -914,60 +839,71 @@ func (h *Handler) gcCache() {
|
|||||||
}
|
}
|
||||||
defer db.Close()
|
defer db.Close()
|
||||||
|
|
||||||
h.evictIncomplete(db)
|
// Remove the caches which are not completed for a while, they are most likely to be broken.
|
||||||
h.evictExpired(db)
|
|
||||||
h.evictSuperseded(db)
|
|
||||||
h.evictOversized(db)
|
|
||||||
h.evictForFreeSpace(db)
|
|
||||||
}
|
|
||||||
|
|
||||||
// evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never
|
|
||||||
// accounted for.
|
|
||||||
func (h *Handler) evictForFreeSpace(db *bolthold.Store) {
|
|
||||||
free, err := h.freeDisk(h.dir)
|
|
||||||
if err != nil {
|
|
||||||
h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full
|
|
||||||
return
|
|
||||||
}
|
|
||||||
if free >= uint64(h.policy.MinFreeDisk) {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
caches := h.completedByUse(db)
|
|
||||||
total, shortfall := totalSize(caches), h.policy.MinFreeDisk-int64(free)
|
|
||||||
if total <= shortfall {
|
|
||||||
// Say so, or shedding everything and still being short reads as the backstop working.
|
|
||||||
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.evictTo(db, caches, total-shortfall, "the cache volume")
|
|
||||||
}
|
|
||||||
|
|
||||||
// evictIncomplete removes uploads that stopped part way, which are most likely broken.
|
|
||||||
func (h *Handler) evictIncomplete(db *bolthold.Store) {
|
|
||||||
h.sweep(db, bolthold.
|
|
||||||
Where("UsedAt").Lt(time.Now().Add(-uploadStallTimeout).Unix()).
|
|
||||||
And("Complete").Eq(false).
|
|
||||||
Index("UsedAt"))
|
|
||||||
}
|
|
||||||
|
|
||||||
func (h *Handler) evictExpired(db *bolthold.Store) {
|
|
||||||
if h.policy.Retention <= 0 {
|
|
||||||
return
|
|
||||||
}
|
|
||||||
// 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"))
|
|
||||||
}
|
|
||||||
|
|
||||||
// evictSuperseded removes entries a newer one with the same key and version replaced. The
|
|
||||||
// aggregation includes Repo so two repos sharing a (key, version) do not evict each other.
|
|
||||||
func (h *Handler) evictSuperseded(db *bolthold.Store) {
|
|
||||||
results, err := db.FindAggregate(&Cache{}, bolthold.Where("Complete").Eq(true).Index("Complete"), "Repo", "Key", "Version")
|
|
||||||
if err != nil {
|
|
||||||
h.logger.Warnf("find aggregate caches: %v", err)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
var caches []*Cache
|
var caches []*Cache
|
||||||
|
if err := db.Find(&caches, bolthold.
|
||||||
|
Where("UsedAt").Lt(time.Now().Add(-keepTemp).Unix()).
|
||||||
|
And("Complete").Eq(false),
|
||||||
|
); err != nil {
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Remove the old caches which have not been used recently.
|
||||||
|
caches = caches[:0]
|
||||||
|
if err := db.Find(&caches, bolthold.
|
||||||
|
Where("UsedAt").Lt(time.Now().Add(-keepUnused).Unix()),
|
||||||
|
); err != nil {
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Remove the old caches which are too old.
|
||||||
|
caches = caches[:0]
|
||||||
|
if err := db.Find(&caches, bolthold.
|
||||||
|
Where("CreatedAt").Lt(time.Now().Add(-keepUsed).Unix()),
|
||||||
|
); err != nil {
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Remove the old caches with the same key and version within the same
|
||||||
|
// repository, keep the latest one. Aggregation must include Repo so two
|
||||||
|
// repos that happen to share a (key, version) do not evict each other —
|
||||||
|
// otherwise per-repo scoping holds for reads but one repo can age
|
||||||
|
// another out after keepOld.
|
||||||
|
// 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)
|
||||||
|
} else {
|
||||||
for _, result := range results {
|
for _, result := range results {
|
||||||
if result.Count() <= 1 {
|
if result.Count() <= 1 {
|
||||||
continue
|
continue
|
||||||
@@ -976,144 +912,20 @@ func (h *Handler) evictSuperseded(db *bolthold.Store) {
|
|||||||
caches = caches[:0]
|
caches = caches[:0]
|
||||||
result.Reduction(&caches)
|
result.Reduction(&caches)
|
||||||
for _, cache := range caches[:len(caches)-1] {
|
for _, cache := range caches[:len(caches)-1] {
|
||||||
if inUse(cache) {
|
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
|
continue
|
||||||
}
|
}
|
||||||
h.deleteCache(db, cache)
|
h.storage.Remove(cache.ID)
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// 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 {
|
if err := db.Delete(cache.ID, cache); err != nil {
|
||||||
h.logger.Warnf("delete cache: %v", err)
|
h.logger.Warnf("delete cache: %v", err)
|
||||||
return false
|
continue
|
||||||
}
|
}
|
||||||
h.logger.Infof("deleted cache: %+v", cache)
|
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) {
|
||||||
@@ -8,7 +8,6 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -42,19 +41,16 @@ 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(JobCredential{}, uint64(id), time.Now().Add(artifactURLTTL))
|
return h.signedArtifactURL(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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
||||||
|
|
||||||
@@ -660,7 +656,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) {
|
func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) { //nolint:unparam // pre-existing issue from nektos/act
|
||||||
var id uint64
|
var id uint64
|
||||||
{
|
{
|
||||||
body, err := json.Marshal(&Request{
|
body, err := json.Marshal(&Request{
|
||||||
@@ -726,7 +722,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(Options{Dir: dir, Policy: Policy{Retention: testRetention}})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
@@ -756,8 +752,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(-(inUseGrace + time.Second)).Unix(),
|
UsedAt: now.Add(-(keepTemp + time.Second)).Unix(),
|
||||||
CreatedAt: now.Add(-(inUseGrace + time.Hour)).Unix(),
|
CreatedAt: now.Add(-(keepTemp + time.Hour)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: false,
|
||||||
},
|
},
|
||||||
@@ -767,21 +763,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(-(testRetention + time.Second)).Unix(),
|
UsedAt: now.Add(-(keepUnused + time.Second)).Unix(),
|
||||||
CreatedAt: now.Add(-(testRetention + time.Hour)).Unix(),
|
CreatedAt: now.Add(-(keepUnused + time.Hour)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: false,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
// should be kept, since age alone does not retire an entry that is still used.
|
// should be removed, since it's used but too old.
|
||||||
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(-365 * 24 * time.Hour).Unix(),
|
CreatedAt: now.Add(-(keepUsed + time.Second)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: true,
|
Kept: false,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
// 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.
|
||||||
@@ -789,7 +785,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(-(inUseGrace - time.Minute)).Unix(),
|
UsedAt: now.Add(-(keepOld - time.Minute)).Unix(),
|
||||||
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: true,
|
Kept: true,
|
||||||
@@ -800,7 +796,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(-(inUseGrace + time.Second)).Unix(),
|
UsedAt: now.Add(-(keepOld + time.Second)).Unix(),
|
||||||
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: false,
|
||||||
@@ -833,265 +829,11 @@ 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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1124,7 +866,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1144,7 +886,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1175,7 +917,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
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"})
|
||||||
@@ -1241,7 +983,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
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})
|
||||||
@@ -1256,7 +998,7 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
t.Run("tampered signature", func(t *testing.T) {
|
t.Run("tampered signature", func(t *testing.T) {
|
||||||
good := signArtifactURL(handler, 1)
|
good := handler.signedArtifactURL(1, time.Now().Add(artifactURLTTL))
|
||||||
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)
|
||||||
@@ -1265,7 +1007,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(JobCredential{}, 1, time.Now().Add(-time.Second))
|
expired := handler.signedArtifactURL(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()
|
||||||
@@ -1274,10 +1016,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(Options{Dir: dir2})
|
other, err := StartHandler(dir2, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer other.Close()
|
defer other.Close()
|
||||||
otherURL := signArtifactURL(other, 1)
|
otherURL := other.signedArtifactURL(1, time.Now().Add(artifactURLTTL))
|
||||||
// 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)
|
||||||
@@ -1296,13 +1038,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(Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
first, err := StartHandler(dir, "127.0.0.1", 0, "", nil)
|
||||||
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(Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
second, err := StartHandler(dir, "127.0.0.1", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer second.Close()
|
defer second.Close()
|
||||||
|
|
||||||
@@ -1314,7 +1056,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
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})
|
||||||
@@ -1354,7 +1096,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1383,10 +1125,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
|
||||||
// inUseGrace window, GC must keep the one from each repo.
|
// keepOld 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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
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"})
|
||||||
@@ -1400,7 +1142,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(-inUseGrace - time.Minute).Unix()
|
stale := time.Now().Add(-keepOld - 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))
|
||||||
@@ -1437,7 +1179,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(Options{Dir: dir})
|
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1455,7 +1197,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(Options{Dir: dir, InternalSecret: secret})
|
handler, err := StartHandler(dir, "", 0, secret, nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -77,7 +77,6 @@ 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
|
||||||
}
|
}
|
||||||
@@ -98,7 +97,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(cred, blobPath, blobUploadPurpose, cache.ID, time.Now().Add(blobUploadURLTTL)),
|
"signed_upload_url": h.signedURL(blobPath, blobUploadPurpose, cache.ID, time.Now().Add(blobUploadURLTTL)),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -169,7 +168,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(cred, cache.ID, time.Now().Add(artifactURLTTL)),
|
"signed_download_url": h.signedArtifactURL(cache.ID, time.Now().Add(artifactURLTTL)),
|
||||||
"matched_key": cache.Key,
|
"matched_key": cache.Key,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
+14
-16
@@ -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,6 +66,16 @@ 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) {
|
||||||
@@ -87,7 +97,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 := newTestHandler(t, Policy{})
|
handler := startTestHandler(t)
|
||||||
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)
|
||||||
@@ -116,7 +126,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 := newTestHandler(t, Policy{})
|
handler := startTestHandler(t)
|
||||||
|
|
||||||
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)
|
||||||
@@ -153,7 +163,7 @@ func TestCacheServiceV2BlockUpload(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestCacheServiceV2Lookups(t *testing.T) {
|
func TestCacheServiceV2Lookups(t *testing.T) {
|
||||||
handler := newTestHandler(t, Policy{})
|
handler := startTestHandler(t)
|
||||||
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"])
|
||||||
|
|
||||||
@@ -188,18 +198,6 @@ 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,
|
||||||
@@ -24,7 +24,7 @@ func TestFrontResultsService(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer gitea.Close()
|
defer gitea.Close()
|
||||||
|
|
||||||
handler, err := StartHandler(Options{Dir: t.TempDir(), OutboundIP: "127.0.0.1"})
|
handler, err := StartHandler(t.TempDir(), "127.0.0.1", 0, "", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
const token = "forward-token"
|
const token = "forward-token"
|
||||||
@@ -143,13 +143,9 @@ func (s *Storage) Serve(w http.ResponseWriter, r *http.Request, id uint64) {
|
|||||||
http.ServeFile(w, r, name)
|
http.ServeFile(w, r, name)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Remove deletes an entry's blob and any staged parts. It reports failure so the caller can
|
func (s *Storage) Remove(id uint64) {
|
||||||
// keep the entry and retry, rather than dropping the only reference to bytes on disk.
|
_ = os.Remove(s.filename(id))
|
||||||
func (s *Storage) Remove(id uint64) error {
|
_ = os.RemoveAll(s.tempDir(id))
|
||||||
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 {
|
||||||
@@ -17,7 +17,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/julienschmidt/httprouter"
|
"github.com/julienschmidt/httprouter"
|
||||||
)
|
)
|
||||||
@@ -15,7 +15,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
"gitea.com/gitea/runner/internal/pkg/lock"
|
"gitea.com/gitea/runner/internal/pkg/lock"
|
||||||
|
|
||||||
"github.com/go-git/go-git/v5"
|
"github.com/go-git/go-git/v5"
|
||||||
@@ -345,16 +345,6 @@ 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 {
|
||||||
@@ -395,7 +385,7 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if !isOfflineMode {
|
if !isOfflineMode {
|
||||||
err = r.FetchContext(ctx, &fetchOptions)
|
err = r.Fetch(&fetchOptions)
|
||||||
if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
|
if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -464,17 +454,18 @@ 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.PullContext(ctx, &pullOptions); err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
|
if err = w.Pull(&pullOptions); err != nil && 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 = " (offline mode)"
|
||||||
}
|
}
|
||||||
|
|
||||||
logger.Debugf("Cloned %s to %s%s", input.URL, input.Dir, reusedMsg)
|
if reused {
|
||||||
|
logger.Debugf("Reused %s at %s%s", input.URL, input.Dir, reusedMsg)
|
||||||
|
} else {
|
||||||
|
logger.Debugf("Cloned %s to %s", input.URL, input.Dir)
|
||||||
|
}
|
||||||
|
|
||||||
if hash.String() != input.Ref && refType == "branch" {
|
if hash.String() != input.Ref && refType == "branch" {
|
||||||
logger.Debugf("Provided ref is not a sha. Updating branch ref after pull")
|
logger.Debugf("Provided ref is not a sha. Updating branch ref after pull")
|
||||||
@@ -6,24 +6,18 @@ 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/internal/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"
|
||||||
@@ -379,22 +373,28 @@ func TestGitCloneExecutorOfflineMode(t *testing.T) {
|
|||||||
|
|
||||||
// Prime the cache with an online clone of main.
|
// Prime the cache with an online clone of main.
|
||||||
cacheDir := t.TempDir()
|
cacheDir := t.TempDir()
|
||||||
|
logger, hook := logrustest.NewNullLogger()
|
||||||
|
logger.SetLevel(log.DebugLevel)
|
||||||
|
ctx := common.WithLogger(context.Background(), logger.WithField("job", "j1"))
|
||||||
require.NoError(t, NewGitCloneExecutor(NewGitCloneExecutorInput{
|
require.NoError(t, NewGitCloneExecutor(NewGitCloneExecutorInput{
|
||||||
URL: remoteDir,
|
URL: remoteDir,
|
||||||
Ref: "main",
|
Ref: "main",
|
||||||
Dir: cacheDir,
|
Dir: cacheDir,
|
||||||
})(context.Background()))
|
})(ctx))
|
||||||
|
assert.Contains(t, logMessages(hook), "Cloned "+remoteDir+" to "+cacheDir)
|
||||||
|
|
||||||
t.Run("cached branch resolves without fetching", func(t *testing.T) {
|
t.Run("cached branch resolves without fetching", func(t *testing.T) {
|
||||||
// Offline reuse of a cached branch must succeed even though ResolveRevision(input.Ref)
|
// Offline reuse of a cached branch must succeed even though ResolveRevision(input.Ref)
|
||||||
// finds no local refs/heads/<ref>.
|
// finds no local refs/heads/<ref>.
|
||||||
|
hook.Reset()
|
||||||
err := NewGitCloneExecutor(NewGitCloneExecutorInput{
|
err := NewGitCloneExecutor(NewGitCloneExecutorInput{
|
||||||
URL: remoteDir,
|
URL: remoteDir,
|
||||||
Ref: "main",
|
Ref: "main",
|
||||||
Dir: cacheDir,
|
Dir: cacheDir,
|
||||||
OfflineMode: true,
|
OfflineMode: true,
|
||||||
})(context.Background())
|
})(ctx)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
assert.Contains(t, logMessages(hook), "Reused "+remoteDir+" at "+cacheDir+" (offline mode)")
|
||||||
|
|
||||||
out, err := exec.Command("git", "-C", cacheDir, "log", "--oneline", "-1", "--format=%s").Output()
|
out, err := exec.Command("git", "-C", cacheDir, "log", "--oneline", "-1", "--format=%s").Output()
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -451,6 +451,14 @@ func TestGitCloneExecutorQuietDemotesCloneLine(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func logMessages(hook *logrustest.Hook) []string {
|
||||||
|
messages := []string{}
|
||||||
|
for _, entry := range hook.AllEntries() {
|
||||||
|
messages = append(messages, entry.Message)
|
||||||
|
}
|
||||||
|
return messages
|
||||||
|
}
|
||||||
|
|
||||||
func TestGitCloneExecutorShallow(t *testing.T) {
|
func TestGitCloneExecutorShallow(t *testing.T) {
|
||||||
// Build a local "remote" with several commits on main plus a tag, so a full clone would pull noticeably more history than a shallow one.
|
// Build a local "remote" with several commits on main plus a tag, so a full clone would pull noticeably more history than a shallow one.
|
||||||
remoteDir := t.TempDir()
|
remoteDir := t.TempDir()
|
||||||
@@ -616,55 +624,3 @@ 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))
|
|
||||||
}
|
|
||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/docker/go-connections/nat"
|
"github.com/docker/go-connections/nat"
|
||||||
"github.com/moby/moby/api/types/container"
|
"github.com/moby/moby/api/types/container"
|
||||||
@@ -9,7 +9,7 @@ package container
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/distribution/reference"
|
"github.com/distribution/reference"
|
||||||
"github.com/docker/cli/cli/config"
|
"github.com/docker/cli/cli/config"
|
||||||
@@ -12,7 +12,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/moby/go-archive"
|
"github.com/moby/go-archive"
|
||||||
"github.com/moby/go-archive/compression"
|
"github.com/moby/go-archive/compression"
|
||||||
@@ -13,7 +13,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/moby/moby/client"
|
"github.com/moby/moby/client"
|
||||||
)
|
)
|
||||||
@@ -11,7 +11,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/distribution/reference"
|
"github.com/distribution/reference"
|
||||||
"github.com/moby/moby/api/pkg/authconfig"
|
"github.com/moby/moby/api/pkg/authconfig"
|
||||||
@@ -22,8 +22,8 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
"gitea.com/gitea/runner/act/filecollector"
|
"gitea.com/gitea/runner/internal/act/filecollector"
|
||||||
|
|
||||||
"dario.cat/mergo"
|
"dario.cat/mergo"
|
||||||
cerrdefs "github.com/containerd/errdefs"
|
cerrdefs "github.com/containerd/errdefs"
|
||||||
@@ -88,14 +88,14 @@ func (cr *containerReference) connectToNetwork(name string, aliases []string) co
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// supportsContainerImagePlatform reports whether the Docker server API version
|
// supportsContainerImagePlatform returns true if the underlying Docker server
|
||||||
// is 1.41 and beyond
|
// API version is 1.41 and beyond
|
||||||
func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) (bool, error) {
|
func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) bool {
|
||||||
ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{})
|
ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Errorf("get docker API version: %w", err)
|
common.Logger(ctx).Panicf("Failed to get Docker API Version: %s", err)
|
||||||
}
|
}
|
||||||
return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41"), nil
|
return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41")
|
||||||
}
|
}
|
||||||
|
|
||||||
func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor {
|
func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor {
|
||||||
@@ -682,17 +682,11 @@ func (cr *containerReference) create(capAdd, capDrop []string) common.Executor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
var platSpecs *specs.Platform
|
var platSpecs *specs.Platform
|
||||||
if cr.input.Platform != "" {
|
if cr.input.Platform != "" && supportsContainerImagePlatform(ctx, cr.cli) {
|
||||||
// Dropping the platform silently would build for the host arch.
|
platSpecs, err = parsePlatform(cr.input.Platform)
|
||||||
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{
|
||||||
@@ -19,7 +19,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
cerrdefs "github.com/containerd/errdefs"
|
cerrdefs "github.com/containerd/errdefs"
|
||||||
"github.com/moby/moby/api/pkg/stdcopy"
|
"github.com/moby/moby/api/pkg/stdcopy"
|
||||||
@@ -79,11 +79,6 @@ 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)
|
||||||
@@ -935,15 +930,3 @@ 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)
|
|
||||||
}
|
|
||||||
@@ -12,7 +12,7 @@ import (
|
|||||||
"runtime"
|
"runtime"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/moby/moby/api/types/system"
|
"github.com/moby/moby/api/types/system"
|
||||||
)
|
)
|
||||||
@@ -9,7 +9,7 @@ package container
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/moby/moby/client"
|
"github.com/moby/moby/client"
|
||||||
)
|
)
|
||||||
@@ -21,9 +21,9 @@ import (
|
|||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
"gitea.com/gitea/runner/act/filecollector"
|
"gitea.com/gitea/runner/internal/act/filecollector"
|
||||||
"gitea.com/gitea/runner/act/lookpath"
|
"gitea.com/gitea/runner/internal/act/lookpath"
|
||||||
"gitea.com/gitea/runner/internal/pkg/process"
|
"gitea.com/gitea/runner/internal/pkg/process"
|
||||||
|
|
||||||
"github.com/go-git/go-billy/v5/helper/polyfill"
|
"github.com/go-git/go-billy/v5/helper/polyfill"
|
||||||
+1
-1
@@ -17,7 +17,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"github.com/sirupsen/logrus"
|
"github.com/sirupsen/logrus"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -12,7 +12,7 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"golang.org/x/text/encoding/unicode"
|
"golang.org/x/text/encoding/unicode"
|
||||||
"golang.org/x/text/transform"
|
"golang.org/x/text/transform"
|
||||||
@@ -13,8 +13,8 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
"gitea.com/gitea/runner/act/common/git"
|
"gitea.com/gitea/runner/internal/act/common/git"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
)
|
)
|
||||||
@@ -20,9 +20,9 @@ import (
|
|||||||
"runtime"
|
"runtime"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
"gitea.com/gitea/runner/act/common/git"
|
"gitea.com/gitea/runner/internal/act/common/git"
|
||||||
"gitea.com/gitea/runner/act/container"
|
"gitea.com/gitea/runner/internal/act/container"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
"github.com/kballard/go-shellquote"
|
"github.com/kballard/go-shellquote"
|
||||||
@@ -293,7 +293,9 @@ func removeGitIgnore(ctx context.Context, directory string) error {
|
|||||||
// same `act-dockeraction:latest` image on a shared docker daemon. A subsequent
|
// same `act-dockeraction:latest` image on a shared docker daemon. A subsequent
|
||||||
// repository would then silently run the image built for an earlier one.
|
// repository would then silently run the image built for an earlier one.
|
||||||
// Including the repository keeps the tag stable for caching within a repository
|
// Including the repository keeps the tag stable for caching within a repository
|
||||||
// while preventing cross-repository collisions.
|
// while preventing cross-repository collisions. A remote action needs the same
|
||||||
|
// treatment, because its actionName is the shared checkout of its repository and
|
||||||
|
// ref plus the action's path inside it.
|
||||||
// See https://gitea.com/gitea/runner/issues/1039.
|
// See https://gitea.com/gitea/runner/issues/1039.
|
||||||
func dockerActionImageTag(repository, actionName string, localAction bool) string {
|
func dockerActionImageTag(repository, actionName string, localAction bool) string {
|
||||||
name := actionName
|
name := actionName
|
||||||
@@ -302,11 +304,10 @@ func dockerActionImageTag(repository, actionName string, localAction bool) strin
|
|||||||
}
|
}
|
||||||
// The human-readable name is sanitized by collapsing every non-alphanumeric character to "-".
|
// The human-readable name is sanitized by collapsing every non-alphanumeric character to "-".
|
||||||
sanitized := regexp.MustCompile("[^a-zA-Z0-9]").ReplaceAllString(name, "-")
|
sanitized := regexp.MustCompile("[^a-zA-Z0-9]").ReplaceAllString(name, "-")
|
||||||
if localAction {
|
// Sanitizing is lossy, so a short hash of the raw repository and action path is appended, keeping
|
||||||
// For local actions a short hash of the raw repository and action path is appended so the tag stays unique per repository.
|
// the tag unique per repository and per action inside it.
|
||||||
sum := sha256.Sum256([]byte(repository + "\x00" + actionName))
|
sum := sha256.Sum256([]byte(repository + "\x00" + actionName))
|
||||||
sanitized += "-" + hex.EncodeToString(sum[:])[:12]
|
sanitized += "-" + hex.EncodeToString(sum[:])[:12]
|
||||||
}
|
|
||||||
// "-dockeraction" ensures that "./", "./test " won't get converted to "act-:latest", "act-test-:latest" which are invalid docker image names
|
// "-dockeraction" ensures that "./", "./test " won't get converted to "act-:latest", "act-test-:latest" which are invalid docker image names
|
||||||
image := fmt.Sprintf("%s-dockeraction:%s", sanitized, "latest")
|
image := fmt.Sprintf("%s-dockeraction:%s", sanitized, "latest")
|
||||||
image = "act-" + strings.TrimLeft(image, "-")
|
image = "act-" + strings.TrimLeft(image, "-")
|
||||||
@@ -17,7 +17,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -31,7 +31,7 @@ func runGit(t *testing.T, dir string, args ...string) {
|
|||||||
}
|
}
|
||||||
cmd := exec.Command("git", args...)
|
cmd := exec.Command("git", args...)
|
||||||
// Fixed identity and host-config isolation so commits succeed offline regardless of the
|
// Fixed identity and host-config isolation so commits succeed offline regardless of the
|
||||||
// host's git config (mirrors gitCmd in act/common/git).
|
// host's git config (mirrors gitCmd in internal/act/common/git).
|
||||||
cmd.Env = append(os.Environ(),
|
cmd.Env = append(os.Environ(),
|
||||||
"GIT_AUTHOR_NAME=test", "GIT_AUTHOR_EMAIL=test@example.com",
|
"GIT_AUTHOR_NAME=test", "GIT_AUTHOR_EMAIL=test@example.com",
|
||||||
"GIT_COMMITTER_NAME=test", "GIT_COMMITTER_EMAIL=test@example.com",
|
"GIT_COMMITTER_NAME=test", "GIT_COMMITTER_EMAIL=test@example.com",
|
||||||
@@ -12,7 +12,7 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
)
|
)
|
||||||
@@ -13,9 +13,9 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
"gitea.com/gitea/runner/act/common/git"
|
"gitea.com/gitea/runner/internal/act/common/git"
|
||||||
"gitea.com/gitea/runner/act/container"
|
"gitea.com/gitea/runner/internal/act/container"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -496,11 +496,11 @@ func TestExecAsDockerHoldsCloneLockForRemoteUncached(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestDockerActionImageTag(t *testing.T) {
|
func TestDockerActionImageTag(t *testing.T) {
|
||||||
// Remote actions already carry a unique, ref-scoped actionName (the uses
|
// A remote action's actionName is the checkout of its repository and ref plus its path inside it,
|
||||||
// hash), so the tag must be left untouched for backwards compatibility.
|
// and paths that sanitize alike share the readable prefix, so siblings must stay apart.
|
||||||
assert.Equal(t,
|
assert.NotEqual(t,
|
||||||
"act-abc123-dockeraction:latest",
|
dockerActionImageTag("owner/repo", "abc123/a-b", false),
|
||||||
dockerActionImageTag("owner/repo", "abc123", false),
|
dockerActionImageTag("owner/repo", "abc123/a_b", false),
|
||||||
)
|
)
|
||||||
|
|
||||||
// Local actions keep a human-readable, repository-namespaced prefix and gain a short hash suffix that makes the tag unique per (repository, actionName).
|
// Local actions keep a human-readable, repository-namespaced prefix and gain a short hash suffix that makes the tag unique per (repository, actionName).
|
||||||
@@ -8,7 +8,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/exprparser"
|
"gitea.dev/actionslib/pkg/exprparser"
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/internal/act/common"
|
||||||
)
|
)
|
||||||
|
|
||||||
var commandPatternGA *regexp.Regexp
|
var commandPatternGA *regexp.Regexp
|
||||||
Some files were not shown because too many files have changed in this diff Show More
Reference in New Issue
Block a user