mirror of
https://gitea.com/gitea/runner.git
synced 2026-08-28 06:47:45 +00:00
Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| bc8161c673 | |||
| 3f70822458 | |||
| be90c01468 | |||
| 6c6a878403 | |||
| f97680a68d | |||
| dbd9a892f8 | |||
| e178c03adc |
@@ -20,11 +20,10 @@ jobs:
|
|||||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
|
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
|
||||||
with:
|
with:
|
||||||
fetch-depth: 0
|
fetch-depth: 0
|
||||||
# Custom publishers (the R2 mirror below) run as the very last
|
# Custom publishers (the R2 upload below) run as the very last
|
||||||
# step of goreleaser's publish pipeline, after the Gitea release
|
# step of goreleaser's publish pipeline, after the Gitea release
|
||||||
# has already been created and every artifact already uploaded
|
# has already been created. Fail here instead, before anything
|
||||||
# to S3. Fail here instead, before anything is built or
|
# is built or published, if the R2 secrets are missing.
|
||||||
# published, if the R2 secrets are missing.
|
|
||||||
- name: check R2 configuration
|
- name: check R2 configuration
|
||||||
run: sh scripts/upload-r2.sh --check-config
|
run: sh scripts/upload-r2.sh --check-config
|
||||||
env:
|
env:
|
||||||
@@ -43,11 +42,6 @@ jobs:
|
|||||||
args: release --nightly
|
args: release --nightly
|
||||||
env:
|
env:
|
||||||
GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
|
GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
|
||||||
AWS_REGION: ${{ secrets.AWS_REGION }}
|
|
||||||
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_KEY_ID }}
|
|
||||||
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
|
|
||||||
S3_REGION: ${{ secrets.AWS_REGION }}
|
|
||||||
S3_BUCKET: ${{ secrets.AWS_BUCKET }}
|
|
||||||
R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
|
R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
|
||||||
R2_BUCKET: ${{ secrets.R2_BUCKET }}
|
R2_BUCKET: ${{ secrets.R2_BUCKET }}
|
||||||
R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
|
R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
|
||||||
|
|||||||
@@ -12,11 +12,10 @@ jobs:
|
|||||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
|
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7
|
||||||
with:
|
with:
|
||||||
fetch-depth: 0 # all history for all branches and tags
|
fetch-depth: 0 # all history for all branches and tags
|
||||||
# Custom publishers (the R2 mirror below) run as the very last
|
# Custom publishers (the R2 upload below) run as the very last
|
||||||
# step of goreleaser's publish pipeline, after the Gitea release
|
# step of goreleaser's publish pipeline, after the Gitea release
|
||||||
# has already been created and every artifact already uploaded
|
# has already been created. Fail here instead, before anything
|
||||||
# to S3. Fail here instead, before anything is built or
|
# is built or published, if the R2 secrets are missing.
|
||||||
# published, if the R2 secrets are missing.
|
|
||||||
- name: check R2 configuration
|
- name: check R2 configuration
|
||||||
run: sh scripts/upload-r2.sh --check-config
|
run: sh scripts/upload-r2.sh --check-config
|
||||||
env:
|
env:
|
||||||
@@ -42,11 +41,6 @@ jobs:
|
|||||||
args: release
|
args: release
|
||||||
env:
|
env:
|
||||||
GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
|
GORELEASER_KEY: ${{ secrets.GORELEASER_KEY }}
|
||||||
AWS_REGION: ${{ secrets.AWS_REGION }}
|
|
||||||
AWS_ACCESS_KEY_ID: ${{ secrets.AWS_KEY_ID }}
|
|
||||||
AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
|
|
||||||
S3_REGION: ${{ secrets.AWS_REGION }}
|
|
||||||
S3_BUCKET: ${{ secrets.AWS_BUCKET }}
|
|
||||||
R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
|
R2_ENDPOINT: ${{ secrets.R2_ENDPOINT }}
|
||||||
R2_BUCKET: ${{ secrets.R2_BUCKET }}
|
R2_BUCKET: ${{ secrets.R2_BUCKET }}
|
||||||
R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
|
R2_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
|
||||||
|
|||||||
+7
-19
@@ -83,24 +83,12 @@ builds:
|
|||||||
- cmd: sh .goreleaser.checksum.sh {{ .Path }}
|
- cmd: sh .goreleaser.checksum.sh {{ .Path }}
|
||||||
- cmd: sh .goreleaser.checksum.sh {{ .Path }}.xz
|
- cmd: sh .goreleaser.checksum.sh {{ .Path }}.xz
|
||||||
|
|
||||||
blobs:
|
# Uploads every release artifact to Cloudflare R2. The `blobs:` pipe
|
||||||
-
|
# isn't usable here since it authenticates from the global AWS_* env
|
||||||
provider: s3
|
# with no per-entry credentials; `publishers:` supports per-entry
|
||||||
bucket: "{{ .Env.S3_BUCKET }}"
|
# `env:` instead, so it's used to invoke scripts/upload-r2.sh once per
|
||||||
region: "{{ .Env.S3_REGION }}"
|
# artifact. Custom publishers inherit almost nothing from the
|
||||||
directory: "gitea-runner/{{.Version}}"
|
# environment, hence the explicit R2_* forwarding below.
|
||||||
extra_files:
|
|
||||||
- glob: ./**.xz
|
|
||||||
- glob: ./**.sha256
|
|
||||||
|
|
||||||
# Mirrors the S3 `blobs:` upload above into Cloudflare R2 during the
|
|
||||||
# parallel S3+R2 period (S3 will be removed once migration completes).
|
|
||||||
# A second `blobs:` entry is impossible here since the blob pipe
|
|
||||||
# authenticates from the global AWS_* env with no per-entry
|
|
||||||
# credentials; `publishers:` supports per-entry `env:` instead, so
|
|
||||||
# it's used to invoke scripts/upload-r2.sh once per artifact. Custom
|
|
||||||
# publishers inherit almost nothing from the environment, hence the
|
|
||||||
# explicit R2_* forwarding below.
|
|
||||||
#
|
#
|
||||||
# This publisher fires 109 times for 73 distinct keys because
|
# This publisher fires 109 times for 73 distinct keys because
|
||||||
# goreleaser's release pipe already registers `release.extra_files`
|
# goreleaser's release pipe already registers `release.extra_files`
|
||||||
@@ -125,7 +113,7 @@ publishers:
|
|||||||
- R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }}
|
- R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }}
|
||||||
|
|
||||||
archives:
|
archives:
|
||||||
- format: binary
|
- formats: [binary]
|
||||||
name_template: "{{ .Binary }}"
|
name_template: "{{ .Binary }}"
|
||||||
allow_different_binary_count: true
|
allow_different_binary_count: true
|
||||||
|
|
||||||
|
|||||||
+2
-2
@@ -17,7 +17,7 @@ RUN make clean && make build
|
|||||||
### DIND VARIANT
|
### DIND VARIANT
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
FROM docker:29.6.2-dind AS dind
|
FROM docker:29.7.1-dind AS dind
|
||||||
|
|
||||||
ARG VERSION=dev
|
ARG VERSION=dev
|
||||||
|
|
||||||
@@ -37,7 +37,7 @@ ENTRYPOINT ["s6-svscan","/etc/s6"]
|
|||||||
### DIND-ROOTLESS VARIANT
|
### DIND-ROOTLESS VARIANT
|
||||||
#
|
#
|
||||||
#
|
#
|
||||||
FROM docker:29.6.2-dind-rootless AS dind-rootless
|
FROM docker:29.7.1-dind-rootless AS dind-rootless
|
||||||
|
|
||||||
ARG VERSION=dev
|
ARG VERSION=dev
|
||||||
|
|
||||||
|
|||||||
@@ -158,6 +158,32 @@ An edit keeps the comments and the key order of the file. Indentation becomes tw
|
|||||||
|
|
||||||
`config get`, `set`, `add` and `remove` use `config.yaml` (or `config.yml`) from the working directory, then from the directory of the binary, and print their choice to stderr. `config init` writes `config.yaml` in the working directory, and refuses to overwrite an existing config without `--force`. Pass `-c` for another path.
|
`config get`, `set`, `add` and `remove` use `config.yaml` (or `config.yml`) from the working directory, then from the directory of the binary, and print their choice to stderr. `config init` writes `config.yaml` in the working directory, and refuses to overwrite an existing config without `--force`. Pass `-c` for another path.
|
||||||
|
|
||||||
|
#### Tool cache
|
||||||
|
|
||||||
|
Setup actions like `setup-go` install tools into `RUNNER_TOOL_CACHE`, which is `/opt/hostedtoolcache` inside a job. `runner.tool_cache_mode` selects what backs it:
|
||||||
|
|
||||||
|
| Mode | Tool cache | Trade-off |
|
||||||
|
| --- | --- | --- |
|
||||||
|
| `none` (default) | Per job, provided by the job image | A version the image lacks is downloaded in every job |
|
||||||
|
| `shared` | One volume reused by every job | Two jobs writing the same tool version at once corrupt it, so use it only with `runner.capacity: 1` |
|
||||||
|
|
||||||
|
With `none`, tools must come from the job image. Install them into `/opt/hostedtoolcache/<tool>/<version>/<arch>`, with an empty `<arch>.complete` file next to the directory:
|
||||||
|
|
||||||
|
```dockerfile
|
||||||
|
RUN GO=$(curl -fsSL 'https://go.dev/dl/?mode=json' | grep -oP '"version": "\Kgo1\.26\.[0-9]*' | head -1); \
|
||||||
|
DIR="/opt/hostedtoolcache/go/${GO#go}/x64" && \
|
||||||
|
mkdir -p "$(dirname "$DIR")" && \
|
||||||
|
curl -fsSL "https://dl.google.com/go/${GO}.linux-amd64.tar.gz" | tar -xz -C /tmp && \
|
||||||
|
mv /tmp/go "$DIR" && \
|
||||||
|
touch "${DIR}.complete"
|
||||||
|
```
|
||||||
|
|
||||||
|
A workflow requesting a minor version, `go-version: "1.26"`, resolves to the newest matching version in the cache, so a patch update in the image still hits it.
|
||||||
|
|
||||||
|
Of the [runner images](https://gitea.com/gitea/runner-images), the `-full` flavour is the one that ships tools in this layout.
|
||||||
|
|
||||||
|
`gitea-runner exec` reads no config file and takes `--tool-cache-mode` instead, defaulting to `none`.
|
||||||
|
|
||||||
#### Environment variables
|
#### Environment variables
|
||||||
|
|
||||||
Earlier releases let a few environment variables (`GITEA_DEBUG`, `GITEA_TRACE`, `GITEA_RUNNER_CAPACITY`, `GITEA_RUNNER_FILE`, `GITEA_RUNNER_ENVIRON`, `GITEA_RUNNER_ENV_FILE`) override parts of the config. They are gone, use the YAML file for all settings. The Docker images still read their own variables, such as `RUNNER_STATE_FILE`, see [scripts/run.sh](scripts/run.sh) and the container documentation below.
|
Earlier releases let a few environment variables (`GITEA_DEBUG`, `GITEA_TRACE`, `GITEA_RUNNER_CAPACITY`, `GITEA_RUNNER_FILE`, `GITEA_RUNNER_ENVIRON`, `GITEA_RUNNER_ENV_FILE`) override parts of the config. They are gone, use the YAML file for all settings. The Docker images still read their own variables, such as `RUNNER_STATE_FILE`, see [scripts/run.sh](scripts/run.sh) and the container documentation below.
|
||||||
@@ -273,6 +299,12 @@ A password in a proxy URL is hidden in job logs. Any step can still read it, bec
|
|||||||
|
|
||||||
Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default.
|
Each runner starts its own cache server automatically. Cache entries are local to that runner — runners do not share a cache by default.
|
||||||
|
|
||||||
|
**Eviction**
|
||||||
|
|
||||||
|
An entry nothing has read or written for `retention` is removed, and a repository past `repo_size_limit` loses its least recently accessed entries until it fits; `size_limit` caps the whole cache the same way. Age alone never retires an entry still in use, and whatever these allow, the cache keeps free space above `health_check.min_free_disk_space_mb` when health checks are enabled.
|
||||||
|
|
||||||
|
These apply where the cache server runs, so on a shared server they belong in *its* config, not the runners'. See `retention`, `repo_size_limit`, `size_limit` and `sweep_interval` in [config.example.yaml](internal/pkg/config/config.example.yaml) for units and defaults.
|
||||||
|
|
||||||
**Cache service v2**
|
**Cache service v2**
|
||||||
|
|
||||||
`actions/cache@v4.2` and later can use the *cache service v2* API. The runner serves it from the same store as v1, on by default, and it works with `external_server`. Turn it off with:
|
`actions/cache@v4.2` and later can use the *cache service v2* API. The runner serves it from the same store as v1, on by default, and it works with `external_server`. Turn it off with:
|
||||||
@@ -313,6 +345,8 @@ Run one dedicated `gitea-runner cache-server` that all runners point at.
|
|||||||
# external_secret_file: /path/to/secret # secret can also be passed via a file
|
# external_secret_file: /path/to/secret # secret can also be passed via a file
|
||||||
```
|
```
|
||||||
|
|
||||||
|
Jobs reach the cache server at `external_server`, so when a reverse proxy fronts the server, point `external_server` at the proxy. The cache server itself needs no extra configuration.
|
||||||
|
|
||||||
Alternatively, mount the same NFS/CIFS share on every runner and point `cache.dir` at it — simpler, but with weaker isolation between repositories.
|
Alternatively, mount the same NFS/CIFS share on every runner and point `cache.dir` at it — simpler, but with weaker isolation between repositories.
|
||||||
|
|
||||||
**S3 / MinIO** — mount object storage as a FUSE filesystem (e.g. [s3fs](https://github.com/s3fs-fuse/s3fs-fuse) or [goofys](https://github.com/kahing/goofys)) and set `cache.dir` to the mount point.
|
**S3 / MinIO** — mount object storage as a FUSE filesystem (e.g. [s3fs](https://github.com/s3fs-fuse/s3fs-fuse) or [goofys](https://github.com/kahing/goofys)) and set `cache.dir` to the mount point.
|
||||||
|
|||||||
+277
-89
@@ -5,6 +5,7 @@
|
|||||||
package artifactcache
|
package artifactcache
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"cmp"
|
||||||
"context"
|
"context"
|
||||||
"crypto/hmac"
|
"crypto/hmac"
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
@@ -20,6 +21,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"regexp"
|
"regexp"
|
||||||
|
"slices"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -27,6 +29,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/act/common"
|
||||||
|
"gitea.com/gitea/runner/internal/pkg/disk"
|
||||||
|
|
||||||
"github.com/julienschmidt/httprouter"
|
"github.com/julienschmidt/httprouter"
|
||||||
"github.com/sirupsen/logrus"
|
"github.com/sirupsen/logrus"
|
||||||
@@ -59,6 +62,9 @@ type JobCredential struct {
|
|||||||
// remote runner registers with.
|
// remote runner registers with.
|
||||||
Results string `json:"results"`
|
Results string `json:"results"`
|
||||||
InsecureTLS bool `json:"insecure_tls"`
|
InsecureTLS bool `json:"insecure_tls"`
|
||||||
|
|
||||||
|
// PublicURL is this server as a reverse proxy makes the job reach it, not the listen address.
|
||||||
|
PublicURL string `json:"public_url"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// credEntry holds a registered job's credential along with an active
|
// credEntry holds a registered job's credential along with an active
|
||||||
@@ -100,19 +106,38 @@ type Handler struct {
|
|||||||
|
|
||||||
credMu sync.RWMutex
|
credMu sync.RWMutex
|
||||||
creds map[string]*credEntry
|
creds map[string]*credEntry
|
||||||
|
|
||||||
|
policy Policy
|
||||||
|
|
||||||
|
// freeDisk is a field so tests can drive evictForFreeSpace without a full volume.
|
||||||
|
freeDisk func(string) (uint64, error)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Options configures a cache server started by StartHandler; the zero value is usable.
|
||||||
|
type Options struct {
|
||||||
|
Dir string
|
||||||
|
OutboundIP string
|
||||||
|
Port uint16
|
||||||
|
|
||||||
|
// InternalSecret, when non-empty, enables a control-plane API at
|
||||||
|
// /_internal/{register,revoke} that lets a remote runner pre-register the
|
||||||
|
// per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. The
|
||||||
|
// embedded in-process handler leaves it empty and registers tokens via the
|
||||||
|
// in-process RegisterJob method directly.
|
||||||
|
InternalSecret string
|
||||||
|
|
||||||
|
Policy Policy
|
||||||
|
Logger logrus.FieldLogger
|
||||||
}
|
}
|
||||||
|
|
||||||
// StartHandler opens the on-disk cache store and starts the HTTP server.
|
// StartHandler opens the on-disk cache store and starts the HTTP server.
|
||||||
//
|
func StartHandler(opts Options) (*Handler, error) {
|
||||||
// internalSecret, when non-empty, enables a control-plane API at
|
dir, logger := opts.Dir, opts.Logger
|
||||||
// /_internal/{register,revoke} that lets a remote runner pre-register the
|
|
||||||
// per-job ACTIONS_RUNTIME_TOKENs it expects this server to honor. The
|
|
||||||
// embedded in-process handler leaves it empty and registers tokens via the
|
|
||||||
// in-process RegisterJob method directly.
|
|
||||||
func StartHandler(dir, outboundIP string, port uint16, internalSecret string, logger logrus.FieldLogger) (*Handler, error) {
|
|
||||||
h := &Handler{
|
h := &Handler{
|
||||||
creds: make(map[string]*credEntry),
|
creds: make(map[string]*credEntry),
|
||||||
internalSecret: internalSecret,
|
internalSecret: opts.InternalSecret,
|
||||||
|
policy: opts.Policy.withDefaults(),
|
||||||
|
freeDisk: disk.FreeBytes,
|
||||||
}
|
}
|
||||||
|
|
||||||
if logger == nil {
|
if logger == nil {
|
||||||
@@ -142,8 +167,8 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
|||||||
}
|
}
|
||||||
h.storage = storage
|
h.storage = storage
|
||||||
|
|
||||||
if outboundIP != "" {
|
if opts.OutboundIP != "" {
|
||||||
h.outboundIP = outboundIP
|
h.outboundIP = opts.OutboundIP
|
||||||
} else if ip := common.GetOutboundIP(); ip == nil {
|
} else if ip := common.GetOutboundIP(); ip == nil {
|
||||||
return nil, errors.New("unable to determine outbound IP address")
|
return nil, errors.New("unable to determine outbound IP address")
|
||||||
} else {
|
} else {
|
||||||
@@ -182,7 +207,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
|||||||
// can break Docker Desktop variants where the host's outbound IP is not
|
// can break Docker Desktop variants where the host's outbound IP is not
|
||||||
// routable from inside the container network. Authentication is enforced
|
// routable from inside the container network. Authentication is enforced
|
||||||
// by the bearer middleware and per-repo scoping, not by reachability.
|
// by the bearer middleware and per-repo scoping, not by reachability.
|
||||||
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", opts.Port))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -212,6 +237,13 @@ func (h *Handler) ExternalURL() string {
|
|||||||
return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
|
return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (h *Handler) baseURL(cred JobCredential) string {
|
||||||
|
if base := strings.TrimRight(cred.PublicURL, "/"); base != "" {
|
||||||
|
return base
|
||||||
|
}
|
||||||
|
return h.ExternalURL()
|
||||||
|
}
|
||||||
|
|
||||||
// RegisterJob makes token a valid bearer credential for cache requests from
|
// RegisterJob makes token a valid bearer credential for cache requests from
|
||||||
// the given repository and returns a function that removes it. The runner
|
// the given repository and returns a function that removes it. The runner
|
||||||
// calls this at job start and defers the returned func so that the credential
|
// calls this at job start and defers the returned func so that the credential
|
||||||
@@ -359,7 +391,7 @@ func (h *Handler) find(w http.ResponseWriter, r *http.Request, _ httprouter.Para
|
|||||||
}
|
}
|
||||||
h.responseJSON(w, r, 200, map[string]any{
|
h.responseJSON(w, r, 200, map[string]any{
|
||||||
"result": "hit",
|
"result": "hit",
|
||||||
"archiveLocation": h.signedArtifactURL(cache.ID, time.Now().Add(artifactURLTTL)),
|
"archiveLocation": h.signedArtifactURL(cred, cache.ID, time.Now().Add(artifactURLTTL)),
|
||||||
"cacheKey": cache.Key,
|
"cacheKey": cache.Key,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -380,6 +412,9 @@ func (h *Handler) lookupCache(db *bolthold.Store, repo string, keys []string, ve
|
|||||||
_ = db.Delete(cache.ID, cache)
|
_ = db.Delete(cache.ID, cache)
|
||||||
return nil, nil //nolint:nilnil // absence is not an error here
|
return nil, nil //nolint:nilnil // absence is not an error here
|
||||||
}
|
}
|
||||||
|
// Handing out a download URL counts as access, or eviction could drop the entry between
|
||||||
|
// this call and the GET that follows it.
|
||||||
|
h.touch(db, cache)
|
||||||
return cache, nil
|
return cache, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -517,13 +552,22 @@ func (h *Handler) commitCache(cache *Cache) error {
|
|||||||
// write real size back to cache, it may be different from the current value when the request doesn't specify it.
|
// write real size back to cache, it may be different from the current value when the request doesn't specify it.
|
||||||
cache.Size = written
|
cache.Size = written
|
||||||
cache.Complete = true
|
cache.Complete = true
|
||||||
|
cache.UsedAt = time.Now().Unix() // a just-written entry counts as accessed, so it cannot be its own eviction victim
|
||||||
|
|
||||||
db, err := h.openDB()
|
db, err := h.openDB()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
defer db.Close()
|
defer db.Close()
|
||||||
return db.Update(cache.ID, cache)
|
if err := db.Update(cache.ID, cache); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
// A commit is the only thing that grows the store, so the only thing that can push the
|
||||||
|
// volume under the floor.
|
||||||
|
h.evictRepo(db, cache.Repo)
|
||||||
|
h.evictTotal(db)
|
||||||
|
h.evictForFreeSpace(db)
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// GET /_apis/artifactcache/artifacts/:id
|
// GET /_apis/artifactcache/artifacts/:id
|
||||||
@@ -641,7 +685,7 @@ func (h *Handler) ResultsURL(cred JobCredential) string {
|
|||||||
if h == nil || cred.Results == "" {
|
if h == nil || cred.Results == "" {
|
||||||
return ""
|
return ""
|
||||||
}
|
}
|
||||||
return h.ExternalURL()
|
return h.baseURL(cred)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
|
func (h *Handler) internalRegister(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
|
||||||
@@ -700,16 +744,16 @@ func (h *Handler) computeSignature(purpose string, cacheID, exp int64) string {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// signedURL builds a URL under path that signedAuth accepts for the same purpose.
|
// signedURL builds a URL under path that signedAuth accepts for the same purpose.
|
||||||
func (h *Handler) signedURL(path, purpose string, cacheID uint64, exp time.Time) string {
|
func (h *Handler) signedURL(cred JobCredential, path, purpose string, cacheID uint64, exp time.Time) string {
|
||||||
expUnix := exp.Unix()
|
expUnix := exp.Unix()
|
||||||
q := url.Values{}
|
q := url.Values{}
|
||||||
q.Set("exp", strconv.FormatInt(expUnix, 10))
|
q.Set("exp", strconv.FormatInt(expUnix, 10))
|
||||||
q.Set("sig", h.computeSignature(purpose, int64(cacheID), expUnix))
|
q.Set("sig", h.computeSignature(purpose, int64(cacheID), expUnix))
|
||||||
return fmt.Sprintf("%s%s/%d?%s", h.ExternalURL(), path, cacheID, q.Encode())
|
return fmt.Sprintf("%s%s/%d?%s", h.baseURL(cred), path, cacheID, q.Encode())
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *Handler) signedArtifactURL(cacheID uint64, exp time.Time) string {
|
func (h *Handler) signedArtifactURL(cred JobCredential, cacheID uint64, exp time.Time) string {
|
||||||
return h.signedURL(apiPath+"/artifacts", "", cacheID, exp)
|
return h.signedURL(cred, apiPath+"/artifacts", "", cacheID, exp)
|
||||||
}
|
}
|
||||||
|
|
||||||
// if not found, return (nil, nil) instead of an error.
|
// if not found, return (nil, nil) instead of an error.
|
||||||
@@ -811,12 +855,43 @@ func (h *Handler) touchCache(id uint64, requireIncomplete bool) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const (
|
const (
|
||||||
keepUsed = 30 * 24 * time.Hour
|
miB = 1024 * 1024
|
||||||
keepUnused = 7 * 24 * time.Hour
|
|
||||||
keepTemp = 5 * time.Minute
|
defaultSweepInterval = time.Hour
|
||||||
keepOld = 5 * time.Minute
|
|
||||||
|
// inUseGrace matches artifactURLTTL so an entry outlives every signed URL still usable
|
||||||
|
// for it, and no sweep cuts off a download in progress.
|
||||||
|
inUseGrace = artifactURLTTL
|
||||||
|
|
||||||
|
// uploadStallTimeout is how long a reservation may sit without a chunk before it counts
|
||||||
|
// as abandoned. Widening it also widens the window for findExactCache to hand a finalize
|
||||||
|
// a stale reservation.
|
||||||
|
uploadStallTimeout = 5 * time.Minute
|
||||||
|
|
||||||
|
defaultMinFreeDisk = 1024 * miB
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// Policy bounds what the cache server keeps: a retention window counted from last access,
|
||||||
|
// and size limits that evict least recently accessed first. A zero limit is no limit.
|
||||||
|
type Policy struct {
|
||||||
|
Retention time.Duration // Retention removes entries nothing has read or written within this window. Zero keeps them regardless of age.
|
||||||
|
RepoSizeLimit int64 // RepoSizeLimit caps one repository's completed entries in bytes, evicting least recently accessed first.
|
||||||
|
SizeLimit int64 // SizeLimit caps every repository's completed entries together, in bytes.
|
||||||
|
SweepInterval time.Duration // SweepInterval is the minimum time between two eviction sweeps.
|
||||||
|
MinFreeDisk int64 // MinFreeDisk is volume headroom the cache will not eat into. Tracks the runner's health-check floor rather than taking a key of its own.
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p Policy) withDefaults() Policy {
|
||||||
|
// The limits default in config.LoadDefault, so a written 0 means off.
|
||||||
|
if p.MinFreeDisk <= 0 {
|
||||||
|
p.MinFreeDisk = defaultMinFreeDisk
|
||||||
|
}
|
||||||
|
if p.SweepInterval <= 0 {
|
||||||
|
p.SweepInterval = defaultSweepInterval
|
||||||
|
}
|
||||||
|
return p
|
||||||
|
}
|
||||||
|
|
||||||
func (h *Handler) gcCache() {
|
func (h *Handler) gcCache() {
|
||||||
if h.gcing.Load() {
|
if h.gcing.Load() {
|
||||||
return
|
return
|
||||||
@@ -826,7 +901,7 @@ func (h *Handler) gcCache() {
|
|||||||
}
|
}
|
||||||
defer h.gcing.Store(false)
|
defer h.gcing.Store(false)
|
||||||
|
|
||||||
if time.Since(h.gcAt) < time.Hour {
|
if time.Since(h.gcAt) < h.policy.SweepInterval {
|
||||||
h.logger.Debugf("skip gc: %v", h.gcAt.String())
|
h.logger.Debugf("skip gc: %v", h.gcAt.String())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -839,71 +914,60 @@ func (h *Handler) gcCache() {
|
|||||||
}
|
}
|
||||||
defer db.Close()
|
defer db.Close()
|
||||||
|
|
||||||
// Remove the caches which are not completed for a while, they are most likely to be broken.
|
h.evictIncomplete(db)
|
||||||
var caches []*Cache
|
h.evictExpired(db)
|
||||||
if err := db.Find(&caches, bolthold.
|
h.evictSuperseded(db)
|
||||||
Where("UsedAt").Lt(time.Now().Add(-keepTemp).Unix()).
|
h.evictOversized(db)
|
||||||
And("Complete").Eq(false),
|
h.evictForFreeSpace(db)
|
||||||
); err != nil {
|
}
|
||||||
h.logger.Warnf("find caches: %v", err)
|
|
||||||
} else {
|
// evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never
|
||||||
for _, cache := range caches {
|
// accounted for.
|
||||||
h.storage.Remove(cache.ID)
|
func (h *Handler) evictForFreeSpace(db *bolthold.Store) {
|
||||||
if err := db.Delete(cache.ID, cache); err != nil {
|
free, err := h.freeDisk(h.dir)
|
||||||
h.logger.Warnf("delete cache: %v", err)
|
if err != nil {
|
||||||
continue
|
h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full
|
||||||
}
|
return
|
||||||
h.logger.Infof("deleted cache: %+v", cache)
|
|
||||||
}
|
}
|
||||||
|
if free >= uint64(h.policy.MinFreeDisk) {
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Remove the old caches which have not been used recently.
|
caches := h.completedByUse(db)
|
||||||
caches = caches[:0]
|
total, shortfall := totalSize(caches), h.policy.MinFreeDisk-int64(free)
|
||||||
if err := db.Find(&caches, bolthold.
|
if total <= shortfall {
|
||||||
Where("UsedAt").Lt(time.Now().Add(-keepUnused).Unix()),
|
// Say so, or shedding everything and still being short reads as the backstop working.
|
||||||
); err != nil {
|
h.logger.Warnf("cache volume is %d MiB short of the free space floor with only %d MiB of cache on it; something else is filling it", shortfall/miB, total/miB)
|
||||||
h.logger.Warnf("find caches: %v", err)
|
|
||||||
} else {
|
|
||||||
for _, cache := range caches {
|
|
||||||
h.storage.Remove(cache.ID)
|
|
||||||
if err := db.Delete(cache.ID, cache); err != nil {
|
|
||||||
h.logger.Warnf("delete cache: %v", err)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
h.logger.Infof("deleted cache: %+v", cache)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
h.evictTo(db, caches, total-shortfall, "the cache volume")
|
||||||
|
}
|
||||||
|
|
||||||
// Remove the old caches which are too old.
|
// evictIncomplete removes uploads that stopped part way, which are most likely broken.
|
||||||
caches = caches[:0]
|
func (h *Handler) evictIncomplete(db *bolthold.Store) {
|
||||||
if err := db.Find(&caches, bolthold.
|
h.sweep(db, bolthold.
|
||||||
Where("CreatedAt").Lt(time.Now().Add(-keepUsed).Unix()),
|
Where("UsedAt").Lt(time.Now().Add(-uploadStallTimeout).Unix()).
|
||||||
); err != nil {
|
And("Complete").Eq(false).
|
||||||
h.logger.Warnf("find caches: %v", err)
|
Index("UsedAt"))
|
||||||
} else {
|
}
|
||||||
for _, cache := range caches {
|
|
||||||
h.storage.Remove(cache.ID)
|
|
||||||
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
|
func (h *Handler) evictExpired(db *bolthold.Store) {
|
||||||
// repository, keep the latest one. Aggregation must include Repo so two
|
if h.policy.Retention <= 0 {
|
||||||
// repos that happen to share a (key, version) do not evict each other —
|
return
|
||||||
// otherwise per-repo scoping holds for reads but one repo can age
|
}
|
||||||
// another out after keepOld.
|
// Never below inUseGrace, or a short retention would outrun a signed URL already issued.
|
||||||
// Also keep the olds which have been used recently for a while in case of the cache is still in use.
|
window := max(h.policy.Retention, inUseGrace)
|
||||||
if results, err := db.FindAggregate(
|
h.sweep(db, bolthold.Where("UsedAt").Lt(time.Now().Add(-window).Unix()).Index("UsedAt"))
|
||||||
&Cache{},
|
}
|
||||||
bolthold.Where("Complete").Eq(true),
|
|
||||||
"Repo", "Key", "Version",
|
// evictSuperseded removes entries a newer one with the same key and version replaced. The
|
||||||
); err != nil {
|
// 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)
|
h.logger.Warnf("find aggregate caches: %v", err)
|
||||||
} else {
|
return
|
||||||
|
}
|
||||||
|
var caches []*Cache
|
||||||
for _, result := range results {
|
for _, result := range results {
|
||||||
if result.Count() <= 1 {
|
if result.Count() <= 1 {
|
||||||
continue
|
continue
|
||||||
@@ -912,20 +976,144 @@ func (h *Handler) gcCache() {
|
|||||||
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 time.Since(time.Unix(cache.UsedAt, 0)) < keepOld {
|
if inUse(cache) {
|
||||||
// Keep it since it has been used recently, even if it's old.
|
|
||||||
// Or it could break downloading in process.
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
h.storage.Remove(cache.ID)
|
h.deleteCache(db, cache)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// evictOversized applies the per-repository limit, then the whole-store one. Only completed
|
||||||
|
// entries count, since only those carry a size measured at commit rather than claimed.
|
||||||
|
func (h *Handler) evictOversized(db *bolthold.Store) {
|
||||||
|
if h.policy.RepoSizeLimit > 0 {
|
||||||
|
byRepo := make(map[string][]*Cache)
|
||||||
|
for _, cache := range h.completedByUse(db) {
|
||||||
|
byRepo[cache.Repo] = append(byRepo[cache.Repo], cache)
|
||||||
|
}
|
||||||
|
for repo, caches := range byRepo {
|
||||||
|
h.evictTo(db, caches, h.policy.RepoSizeLimit, "repository "+repo)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
h.evictTotal(db)
|
||||||
|
}
|
||||||
|
|
||||||
|
// evictTotal caps the store as a whole. It re-queries because the per-repo pass may have
|
||||||
|
// deleted rows an earlier result still holds.
|
||||||
|
func (h *Handler) evictTotal(db *bolthold.Store) {
|
||||||
|
if h.policy.SizeLimit <= 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
h.evictTo(db, h.completedByUse(db), h.policy.SizeLimit, "the cache")
|
||||||
|
}
|
||||||
|
|
||||||
|
// evictRepo reclaims space when a commit pushes a repo over, rather than at the next sweep.
|
||||||
|
func (h *Handler) evictRepo(db *bolthold.Store, repo string) {
|
||||||
|
if h.policy.RepoSizeLimit <= 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
h.evictTo(db, h.cachesByUse(db, bolthold.Where("Repo").Eq(repo).And("Complete").Eq(true).Index("Repo")), h.policy.RepoSizeLimit, "repository "+repo)
|
||||||
|
}
|
||||||
|
|
||||||
|
// evictTo deletes until caches fit limit. caches must be ordered by UsedAt ascending.
|
||||||
|
func (h *Handler) evictTo(db *bolthold.Store, caches []*Cache, limit int64, scope string) {
|
||||||
|
// An entry bigger than the limit never fits, so it goes on its own account instead of
|
||||||
|
// dragging every neighbour out first and then following them next sweep.
|
||||||
|
fits := caches[:0]
|
||||||
|
for _, cache := range caches {
|
||||||
|
if cache.Size <= limit {
|
||||||
|
fits = append(fits, cache)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if !inUse(cache) {
|
||||||
|
h.logger.Warnf("cache %q is %d MiB on its own, over the limit for %s; dropping it", cache.Key, cache.Size/miB, scope)
|
||||||
|
h.deleteCache(db, cache)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
caches = fits
|
||||||
|
|
||||||
|
total := totalSize(caches)
|
||||||
|
var freed int64
|
||||||
|
for _, cache := range caches {
|
||||||
|
if total <= limit {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
if inUse(cache) || !h.deleteCache(db, cache) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
total -= cache.Size
|
||||||
|
freed += cache.Size
|
||||||
|
}
|
||||||
|
if freed > 0 {
|
||||||
|
h.logger.Warnf("evicted %d MiB from %s, least recently used first", freed/miB, scope)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// inUse reports whether an entry was read or written recently enough that removing it
|
||||||
|
// could break a download in progress.
|
||||||
|
func inUse(cache *Cache) bool {
|
||||||
|
return time.Since(time.Unix(cache.UsedAt, 0)) < inUseGrace
|
||||||
|
}
|
||||||
|
|
||||||
|
// touch stamps UsedAt through the caller's store, a bolt write on the read path. It cannot
|
||||||
|
// go through touchCache, which opens its own store and would block on the exclusive lock
|
||||||
|
// for as long as the caller holds one.
|
||||||
|
func (h *Handler) touch(db *bolthold.Store, cache *Cache) {
|
||||||
|
cache.UsedAt = time.Now().Unix()
|
||||||
|
if err := db.Update(cache.ID, cache); err != nil {
|
||||||
|
h.logger.Warnf("touch cache: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *Handler) sweep(db *bolthold.Store, query *bolthold.Query) {
|
||||||
|
for _, cache := range h.caches(db, query) {
|
||||||
|
h.deleteCache(db, cache)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *Handler) caches(db *bolthold.Store, query *bolthold.Query) []*Cache {
|
||||||
|
var caches []*Cache
|
||||||
|
if err := db.Find(&caches, query); err != nil {
|
||||||
|
h.logger.Warnf("find caches: %v", err)
|
||||||
|
}
|
||||||
|
return caches
|
||||||
|
}
|
||||||
|
|
||||||
|
// cachesByUse returns matches least recently accessed first, sorting here rather than with
|
||||||
|
// bolthold's SortBy, which reflects over every field it compares.
|
||||||
|
func (h *Handler) cachesByUse(db *bolthold.Store, query *bolthold.Query) []*Cache {
|
||||||
|
caches := h.caches(db, query)
|
||||||
|
slices.SortFunc(caches, func(a, b *Cache) int { return cmp.Compare(a.UsedAt, b.UsedAt) })
|
||||||
|
return caches
|
||||||
|
}
|
||||||
|
|
||||||
|
// completedByUse returns every entry the size limits count, least recently accessed first.
|
||||||
|
func (h *Handler) completedByUse(db *bolthold.Store) []*Cache {
|
||||||
|
return h.cachesByUse(db, bolthold.Where("Complete").Eq(true).Index("Complete"))
|
||||||
|
}
|
||||||
|
|
||||||
|
func totalSize(caches []*Cache) int64 {
|
||||||
|
var total int64
|
||||||
|
for _, cache := range caches {
|
||||||
|
total += cache.Size
|
||||||
|
}
|
||||||
|
return total
|
||||||
|
}
|
||||||
|
|
||||||
|
// deleteCache drops an entry and its bytes, reporting whether it went fully. The blob goes
|
||||||
|
// first, so a failed unlink leaves the row for the next sweep instead of orphaning bytes.
|
||||||
|
func (h *Handler) deleteCache(db *bolthold.Store, cache *Cache) bool {
|
||||||
|
if err := h.storage.Remove(cache.ID); err != nil {
|
||||||
|
h.logger.Warnf("remove cache blob: %v", err)
|
||||||
|
return false
|
||||||
|
}
|
||||||
if err := db.Delete(cache.ID, cache); err != nil {
|
if err := db.Delete(cache.ID, cache); err != nil {
|
||||||
h.logger.Warnf("delete cache: %v", err)
|
h.logger.Warnf("delete cache: %v", err)
|
||||||
continue
|
return false
|
||||||
}
|
}
|
||||||
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,6 +8,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"crypto/rand"
|
"crypto/rand"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -41,16 +42,19 @@ func (b *bearerTransport) RoundTrip(r *http.Request) (*http.Response, error) {
|
|||||||
|
|
||||||
var testClient = &http.Client{Transport: &bearerTransport{token: testToken}}
|
var testClient = &http.Client{Transport: &bearerTransport{token: testToken}}
|
||||||
|
|
||||||
|
// testRetention mirrors config.DefaultCacheRetention; Policy has no defaults of its own.
|
||||||
|
const testRetention = 7 * 24 * time.Hour
|
||||||
|
|
||||||
// signArtifactURL builds a signed download URL the same way the server does;
|
// signArtifactURL builds a signed download URL the same way the server does;
|
||||||
// tests use it to reach the get handler directly without going through a
|
// tests use it to reach the get handler directly without going through a
|
||||||
// find/cache-hit round trip.
|
// find/cache-hit round trip.
|
||||||
func signArtifactURL(h *Handler, id int64) string {
|
func signArtifactURL(h *Handler, id int64) string {
|
||||||
return h.signedArtifactURL(uint64(id), time.Now().Add(artifactURLTTL))
|
return h.signedArtifactURL(JobCredential{}, uint64(id), time.Now().Add(artifactURLTTL))
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestHandler(t *testing.T) {
|
func TestHandler(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
||||||
|
|
||||||
@@ -656,7 +660,7 @@ func backdateCache(t *testing.T, handler *Handler, key string, age time.Duration
|
|||||||
require.NoError(t, db.Update(caches[0].ID, caches[0]))
|
require.NoError(t, db.Update(caches[0].ID, caches[0]))
|
||||||
}
|
}
|
||||||
|
|
||||||
func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) { //nolint:unparam // pre-existing issue from nektos/act
|
func uploadCacheNormally(t *testing.T, base, key, version string, content []byte) {
|
||||||
var id uint64
|
var id uint64
|
||||||
{
|
{
|
||||||
body, err := json.Marshal(&Request{
|
body, err := json.Marshal(&Request{
|
||||||
@@ -722,7 +726,7 @@ func uploadCacheNormally(t *testing.T, base, key, version string, content []byte
|
|||||||
|
|
||||||
func TestHandler_gcCache(t *testing.T) {
|
func TestHandler_gcCache(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir, Policy: Policy{Retention: testRetention}})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
@@ -752,8 +756,8 @@ func TestHandler_gcCache(t *testing.T) {
|
|||||||
Key: "test_key_2",
|
Key: "test_key_2",
|
||||||
Version: "test_version",
|
Version: "test_version",
|
||||||
Complete: false,
|
Complete: false,
|
||||||
UsedAt: now.Add(-(keepTemp + time.Second)).Unix(),
|
UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(),
|
||||||
CreatedAt: now.Add(-(keepTemp + time.Hour)).Unix(),
|
CreatedAt: now.Add(-(inUseGrace + time.Hour)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: false,
|
||||||
},
|
},
|
||||||
@@ -763,21 +767,21 @@ func TestHandler_gcCache(t *testing.T) {
|
|||||||
Key: "test_key_3",
|
Key: "test_key_3",
|
||||||
Version: "test_version",
|
Version: "test_version",
|
||||||
Complete: true,
|
Complete: true,
|
||||||
UsedAt: now.Add(-(keepUnused + time.Second)).Unix(),
|
UsedAt: now.Add(-(testRetention + time.Second)).Unix(),
|
||||||
CreatedAt: now.Add(-(keepUnused + time.Hour)).Unix(),
|
CreatedAt: now.Add(-(testRetention + time.Hour)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: false,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
// should be removed, since it's used but too old.
|
// should be kept, since age alone does not retire an entry that is still used.
|
||||||
Cache: &Cache{
|
Cache: &Cache{
|
||||||
Key: "test_key_3",
|
Key: "test_key_3",
|
||||||
Version: "test_version",
|
Version: "test_version",
|
||||||
Complete: true,
|
Complete: true,
|
||||||
UsedAt: now.Unix(),
|
UsedAt: now.Unix(),
|
||||||
CreatedAt: now.Add(-(keepUsed + time.Second)).Unix(),
|
CreatedAt: now.Add(-365 * 24 * time.Hour).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: true,
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
// should be kept, since it has a newer edition but be used recently.
|
// should be kept, since it has a newer edition but be used recently.
|
||||||
@@ -785,7 +789,7 @@ func TestHandler_gcCache(t *testing.T) {
|
|||||||
Key: "test_key_1",
|
Key: "test_key_1",
|
||||||
Version: "test_version",
|
Version: "test_version",
|
||||||
Complete: true,
|
Complete: true,
|
||||||
UsedAt: now.Add(-(keepOld - time.Minute)).Unix(),
|
UsedAt: now.Add(-(inUseGrace - time.Minute)).Unix(),
|
||||||
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: true,
|
Kept: true,
|
||||||
@@ -796,7 +800,7 @@ func TestHandler_gcCache(t *testing.T) {
|
|||||||
Key: "test_key_1",
|
Key: "test_key_1",
|
||||||
Version: "test_version",
|
Version: "test_version",
|
||||||
Complete: true,
|
Complete: true,
|
||||||
UsedAt: now.Add(-(keepOld + time.Second)).Unix(),
|
UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(),
|
||||||
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
||||||
},
|
},
|
||||||
Kept: false,
|
Kept: false,
|
||||||
@@ -829,11 +833,265 @@ func TestHandler_gcCache(t *testing.T) {
|
|||||||
require.NoError(t, db.Close())
|
require.NoError(t, db.Close())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestHandler_evictPolicy covers the non-default policies; TestHandler_gcCache covers the
|
||||||
|
// defaults across every pass.
|
||||||
|
func TestHandler_evictPolicy(t *testing.T) {
|
||||||
|
now := time.Now()
|
||||||
|
stale := func(d time.Duration) int64 { return now.Add(-d).Unix() }
|
||||||
|
mib := func(n int64) int64 { return n * miB }
|
||||||
|
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
policy Policy
|
||||||
|
entries []*Cache
|
||||||
|
kept []string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "a zero retention keeps an entry nothing has touched",
|
||||||
|
policy: Policy{Retention: 0},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Key: "idle", UsedAt: stale(testRetention + time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"idle"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "evicts least recently accessed until the repository fits",
|
||||||
|
policy: Policy{RepoSizeLimit: mib(10)},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Repo: "o/a", Key: "oldest", Size: mib(4), UsedAt: stale(3 * time.Hour)},
|
||||||
|
{Repo: "o/a", Key: "middle", Size: mib(4), UsedAt: stale(2 * time.Hour)},
|
||||||
|
{Repo: "o/a", Key: "newest", Size: mib(4), UsedAt: stale(time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"middle", "newest"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "spares entries that may still be downloading",
|
||||||
|
policy: Policy{RepoSizeLimit: mib(10)},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Repo: "o/a", Key: "fresh_1", Size: mib(6), UsedAt: stale(time.Minute)},
|
||||||
|
{Repo: "o/a", Key: "fresh_2", Size: mib(6), UsedAt: stale(time.Minute)},
|
||||||
|
},
|
||||||
|
kept: []string{"fresh_1", "fresh_2"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "one repository over its limit leaves another alone",
|
||||||
|
policy: Policy{RepoSizeLimit: mib(10)},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Repo: "o/a", Key: "a_old", Size: mib(6), UsedAt: stale(3 * time.Hour)},
|
||||||
|
{Repo: "o/a", Key: "a_new", Size: mib(6), UsedAt: stale(time.Hour)},
|
||||||
|
{Repo: "o/b", Key: "b_old", Size: mib(6), UsedAt: stale(4 * time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"a_new", "b_old"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "the total limit evicts across repositories once each fits its own",
|
||||||
|
policy: Policy{RepoSizeLimit: mib(10), SizeLimit: mib(12)},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Repo: "o/a", Key: "a_old", Size: mib(8), UsedAt: stale(3 * time.Hour)},
|
||||||
|
{Repo: "o/b", Key: "b_new", Size: mib(8), UsedAt: stale(time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"b_new"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
// Retention below inUseGrace would otherwise drop an entry whose signed URL a job
|
||||||
|
// is still holding.
|
||||||
|
name: "a retention shorter than the grace still spares a just-served entry",
|
||||||
|
policy: Policy{Retention: time.Minute},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Key: "just_served", UsedAt: stale(2 * time.Minute)},
|
||||||
|
{Key: "idle", UsedAt: stale(time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"just_served"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "an entry over the limit goes without emptying the repository",
|
||||||
|
policy: Policy{RepoSizeLimit: mib(10)},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Repo: "o/a", Key: "keeps", Size: mib(4), UsedAt: stale(3 * time.Hour)},
|
||||||
|
{Repo: "o/a", Key: "huge", Size: mib(20), UsedAt: stale(2 * time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"keeps"},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "a zero limit keeps everything",
|
||||||
|
policy: Policy{RepoSizeLimit: 0},
|
||||||
|
entries: []*Cache{
|
||||||
|
{Repo: "o/a", Key: "huge_1", Size: mib(100), UsedAt: stale(3 * time.Hour)},
|
||||||
|
{Repo: "o/a", Key: "huge_2", Size: mib(100), UsedAt: stale(2 * time.Hour)},
|
||||||
|
},
|
||||||
|
kept: []string{"huge_1", "huge_2"},
|
||||||
|
},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
for _, e := range tc.entries {
|
||||||
|
e.Complete = true // only completed entries carry a measured size, so only they count
|
||||||
|
}
|
||||||
|
handler := newTestHandler(t, tc.policy, tc.entries...)
|
||||||
|
handler.gcAt = time.Time{} // ensure gcCache will not skip
|
||||||
|
handler.gcCache()
|
||||||
|
assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, tc.entries))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHandler_evictForFreeSpace proves the volume backstop sheds only what it must, and only
|
||||||
|
// when the disk is actually short.
|
||||||
|
func TestHandler_evictForFreeSpace(t *testing.T) {
|
||||||
|
free := func(n int64) func(string) (uint64, error) {
|
||||||
|
return func(string) (uint64, error) { return uint64(n), nil }
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
freeDisk func(string) (uint64, error)
|
||||||
|
kept []string
|
||||||
|
}{
|
||||||
|
{"ample free space evicts nothing", free(defaultMinFreeDisk), []string{"oldest", "middle", "newest"}},
|
||||||
|
{"a small shortfall sheds one entry", free(defaultMinFreeDisk - 4*miB), []string{"middle", "newest"}},
|
||||||
|
{"a shortfall the cache cannot cover sheds all of it", free(0), nil},
|
||||||
|
{
|
||||||
|
"an unreadable volume is treated as unavailable, not as full",
|
||||||
|
func(string) (uint64, error) { return 0, errors.New("unsupported") },
|
||||||
|
[]string{"oldest", "middle", "newest"},
|
||||||
|
},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
now := time.Now()
|
||||||
|
entries := []*Cache{
|
||||||
|
{Key: "oldest", Complete: true, Size: 4 * miB, UsedAt: now.Add(-3 * time.Hour).Unix()},
|
||||||
|
{Key: "middle", Complete: true, Size: 4 * miB, UsedAt: now.Add(-2 * time.Hour).Unix()},
|
||||||
|
{Key: "newest", Complete: true, Size: 4 * miB, UsedAt: now.Add(-time.Hour).Unix()},
|
||||||
|
}
|
||||||
|
handler := newTestHandler(t, Policy{}, entries...)
|
||||||
|
handler.freeDisk = tc.freeDisk
|
||||||
|
|
||||||
|
db, err := handler.openDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
handler.evictForFreeSpace(db)
|
||||||
|
require.NoError(t, db.Close())
|
||||||
|
|
||||||
|
assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, entries))
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHandler_SweepKeepsEntryWhenBlobSurvives proves a failed unlink leaves the row in place,
|
||||||
|
// so the next sweep retries rather than orphaning bytes no row points at and no limit counts.
|
||||||
|
func TestHandler_SweepKeepsEntryWhenBlobSurvives(t *testing.T) {
|
||||||
|
cache := &Cache{Key: "stuck", Complete: true, UsedAt: time.Now().Add(-(testRetention + time.Hour)).Unix()}
|
||||||
|
handler := newTestHandler(t, Policy{Retention: testRetention}, cache)
|
||||||
|
|
||||||
|
// A non-empty directory where the blob belongs makes os.Remove fail on every platform.
|
||||||
|
blob := handler.storage.filename(cache.ID)
|
||||||
|
require.NoError(t, os.MkdirAll(blob, 0o755))
|
||||||
|
require.NoError(t, os.WriteFile(filepath.Join(blob, "held"), []byte("x"), 0o600))
|
||||||
|
|
||||||
|
handler.gcAt = time.Time{}
|
||||||
|
handler.gcCache()
|
||||||
|
|
||||||
|
assert.Equal(t, []string{"stuck"}, keptKeys(t, handler, []*Cache{cache}), "the entry must outlive a blob that could not be removed")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHandler_FindProtectsFromEviction covers the window between a find handing out a signed
|
||||||
|
// download URL and the GET that redeems it: the entry promised to a job must not be the next
|
||||||
|
// eviction victim just because its last access predates the find.
|
||||||
|
func TestHandler_FindProtectsFromEviction(t *testing.T) {
|
||||||
|
// 12 MiB against a 10 MiB limit, so exactly one entry has to go.
|
||||||
|
wanted := &Cache{Repo: testRepo, Key: "wanted", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-3 * time.Hour).Unix()}
|
||||||
|
other := &Cache{Repo: testRepo, Key: "other", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-2 * time.Hour).Unix()}
|
||||||
|
newest := &Cache{Repo: testRepo, Key: "newest", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()}
|
||||||
|
handler := newTestHandler(t, Policy{RepoSizeLimit: 10 * miB}, wanted, other, newest)
|
||||||
|
writeBlob(t, handler, wanted.ID) // find only reports a hit when the blob is on disk
|
||||||
|
|
||||||
|
resp, err := testClient.Get(fmt.Sprintf("%s%s/cache?keys=wanted&version=v", handler.ExternalURL(), apiPath))
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer resp.Body.Close()
|
||||||
|
require.Equal(t, 200, resp.StatusCode)
|
||||||
|
|
||||||
|
// Evict directly: the request above kicked off an async gcCache, and writing gcAt here
|
||||||
|
// to drive gcCache would race its read.
|
||||||
|
db, err := handler.openDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer func() { require.NoError(t, db.Close()) }()
|
||||||
|
handler.evictOversized(db)
|
||||||
|
|
||||||
|
require.NoError(t, db.Get(wanted.ID, &Cache{}), "the entry just promised to a job must survive")
|
||||||
|
assert.ErrorIs(t, db.Get(other.ID, &Cache{}), bolthold.ErrNotFound, "the next least recently used goes instead")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestHandler_evictOnCommit proves a repository that goes over its limit gets space back at
|
||||||
|
// once, rather than waiting out the collection interval.
|
||||||
|
func TestHandler_evictOnCommit(t *testing.T) {
|
||||||
|
full := &Cache{Repo: testRepo, Key: "full", Version: "v", Complete: true, Size: 4 * miB, UsedAt: time.Now().Add(-time.Hour).Unix()}
|
||||||
|
handler := newTestHandler(t, Policy{RepoSizeLimit: 4 * miB}, full)
|
||||||
|
|
||||||
|
// StartHandler already stamped gcAt, so the periodic sweep stays rate-limited out and
|
||||||
|
// only the commit path can evict.
|
||||||
|
uploadCacheNormally(t, handler.ExternalURL()+apiPath, "new", "v", []byte("some content"))
|
||||||
|
|
||||||
|
assert.Empty(t, keptKeys(t, handler, []*Cache{full}))
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestHandler_gcCacheInterval(t *testing.T) {
|
||||||
|
cache := &Cache{Key: "temp", UsedAt: time.Now().Add(-time.Hour).Unix()}
|
||||||
|
// Half the default, so a sweep 45m ago is still inside the default but past this one.
|
||||||
|
handler := newTestHandler(t, Policy{SweepInterval: 30 * time.Minute}, cache)
|
||||||
|
|
||||||
|
handler.gcAt = time.Now().Add(-45 * time.Minute) // past the configured interval, still inside the default
|
||||||
|
handler.gcCache()
|
||||||
|
assert.Empty(t, keptKeys(t, handler, []*Cache{cache}))
|
||||||
|
}
|
||||||
|
|
||||||
|
// newTestHandler starts a handler with testToken registered, seeded with entries.
|
||||||
|
func newTestHandler(t *testing.T, policy Policy, entries ...*Cache) *Handler {
|
||||||
|
t.Helper()
|
||||||
|
handler, err := StartHandler(Options{
|
||||||
|
Dir: filepath.Join(t.TempDir(), "artifactcache"),
|
||||||
|
OutboundIP: "127.0.0.1",
|
||||||
|
Policy: policy,
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(func() { require.NoError(t, handler.Close()) })
|
||||||
|
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
||||||
|
|
||||||
|
db, err := handler.openDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
for _, e := range entries {
|
||||||
|
require.NoError(t, insertCache(db, e))
|
||||||
|
}
|
||||||
|
require.NoError(t, db.Close())
|
||||||
|
return handler
|
||||||
|
}
|
||||||
|
|
||||||
|
// keptKeys reports which of entries are still in the store.
|
||||||
|
func keptKeys(t *testing.T, handler *Handler, entries []*Cache) []string {
|
||||||
|
t.Helper()
|
||||||
|
db, err := handler.openDB()
|
||||||
|
require.NoError(t, err)
|
||||||
|
defer func() { require.NoError(t, db.Close()) }()
|
||||||
|
|
||||||
|
var kept []string
|
||||||
|
for _, e := range entries {
|
||||||
|
if err := db.Get(e.ID, &Cache{}); err == nil {
|
||||||
|
kept = append(kept, e.Key)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return kept
|
||||||
|
}
|
||||||
|
|
||||||
|
// writeBlob gives an entry the on-disk bytes that find and get require.
|
||||||
|
func writeBlob(t *testing.T, handler *Handler, id uint64) {
|
||||||
|
t.Helper()
|
||||||
|
require.NoError(t, handler.storage.Write(id, 0, strings.NewReader("a")))
|
||||||
|
_, err := handler.storage.Commit(id, 1)
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
|
||||||
// TestHandler_RejectsMissingBearer covers the advisory's root cause:
|
// TestHandler_RejectsMissingBearer covers the advisory's root cause:
|
||||||
// unauthenticated access to management endpoints is now refused with 401.
|
// unauthenticated access to management endpoints is now refused with 401.
|
||||||
func TestHandler_RejectsMissingBearer(t *testing.T) {
|
func TestHandler_RejectsMissingBearer(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -866,7 +1124,7 @@ func TestHandler_RejectsMissingBearer(t *testing.T) {
|
|||||||
// accepted after RegisterJob; stale/forged tokens cannot be replayed.
|
// accepted after RegisterJob; stale/forged tokens cannot be replayed.
|
||||||
func TestHandler_RejectsUnknownBearer(t *testing.T) {
|
func TestHandler_RejectsUnknownBearer(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -886,7 +1144,7 @@ func TestHandler_RejectsUnknownBearer(t *testing.T) {
|
|||||||
// working the moment the job ends instead of living for the runner's lifetime.
|
// working the moment the job ends instead of living for the runner's lifetime.
|
||||||
func TestHandler_UnregisterRevokes(t *testing.T) {
|
func TestHandler_UnregisterRevokes(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -917,7 +1175,7 @@ func TestHandler_UnregisterRevokes(t *testing.T) {
|
|||||||
// invisible to queries scoped to repoB.
|
// invisible to queries scoped to repoB.
|
||||||
func TestHandler_CrossRepoIsolation(t *testing.T) {
|
func TestHandler_CrossRepoIsolation(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
handler.RegisterJob("token-a", JobCredential{Repo: "owner/repoA"})
|
handler.RegisterJob("token-a", JobCredential{Repo: "owner/repoA"})
|
||||||
@@ -983,7 +1241,7 @@ func TestHandler_CrossRepoIsolation(t *testing.T) {
|
|||||||
// working after artifactURLTTL even if the bearer token is still registered.
|
// working after artifactURLTTL even if the bearer token is still registered.
|
||||||
func TestHandler_ArtifactSignature(t *testing.T) {
|
func TestHandler_ArtifactSignature(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
||||||
@@ -998,7 +1256,7 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
t.Run("tampered signature", func(t *testing.T) {
|
t.Run("tampered signature", func(t *testing.T) {
|
||||||
good := handler.signedArtifactURL(1, time.Now().Add(artifactURLTTL))
|
good := signArtifactURL(handler, 1)
|
||||||
bad := good[:len(good)-4] + "dead"
|
bad := good[:len(good)-4] + "dead"
|
||||||
resp, err := testClient.Get(bad)
|
resp, err := testClient.Get(bad)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -1007,7 +1265,7 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
|||||||
})
|
})
|
||||||
|
|
||||||
t.Run("expired signature", func(t *testing.T) {
|
t.Run("expired signature", func(t *testing.T) {
|
||||||
expired := handler.signedArtifactURL(1, time.Now().Add(-time.Second))
|
expired := handler.signedArtifactURL(JobCredential{}, 1, time.Now().Add(-time.Second))
|
||||||
resp, err := testClient.Get(expired)
|
resp, err := testClient.Get(expired)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
resp.Body.Close()
|
resp.Body.Close()
|
||||||
@@ -1016,10 +1274,10 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
|||||||
|
|
||||||
t.Run("signature from a different server", func(t *testing.T) {
|
t.Run("signature from a different server", func(t *testing.T) {
|
||||||
dir2 := filepath.Join(t.TempDir(), "artifactcache2")
|
dir2 := filepath.Join(t.TempDir(), "artifactcache2")
|
||||||
other, err := StartHandler(dir2, "", 0, "", nil)
|
other, err := StartHandler(Options{Dir: dir2})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer other.Close()
|
defer other.Close()
|
||||||
otherURL := other.signedArtifactURL(1, time.Now().Add(artifactURLTTL))
|
otherURL := signArtifactURL(other, 1)
|
||||||
// Rewrite the host so the request still lands on our handler, but
|
// Rewrite the host so the request still lands on our handler, but
|
||||||
// the signature was computed with a different secret.
|
// the signature was computed with a different secret.
|
||||||
parts := strings.SplitN(otherURL, apiPath, 2)
|
parts := strings.SplitN(otherURL, apiPath, 2)
|
||||||
@@ -1038,13 +1296,13 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
|||||||
func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
|
func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
|
|
||||||
first, err := StartHandler(dir, "127.0.0.1", 0, "", nil)
|
first, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
exp := time.Now().Add(artifactURLTTL).Unix()
|
exp := time.Now().Add(artifactURLTTL).Unix()
|
||||||
sig := first.computeSignature("", 42, exp)
|
sig := first.computeSignature("", 42, exp)
|
||||||
require.NoError(t, first.Close())
|
require.NoError(t, first.Close())
|
||||||
|
|
||||||
second, err := StartHandler(dir, "127.0.0.1", 0, "", nil)
|
second, err := StartHandler(Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer second.Close()
|
defer second.Close()
|
||||||
|
|
||||||
@@ -1056,7 +1314,7 @@ func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
|
|||||||
// the auth refactor.
|
// the auth refactor.
|
||||||
func TestHandler_ArtifactSignatureDownload(t *testing.T) {
|
func TestHandler_ArtifactSignatureDownload(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
||||||
@@ -1096,7 +1354,7 @@ func TestHandler_ArtifactSignatureDownload(t *testing.T) {
|
|||||||
// (restart mid-task, retry), which must not kill the live job's auth.
|
// (restart mid-task, retry), which must not kill the live job's auth.
|
||||||
func TestHandler_RegisterJob_RefCounted(t *testing.T) {
|
func TestHandler_RegisterJob_RefCounted(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1125,10 +1383,10 @@ func TestHandler_RegisterJob_RefCounted(t *testing.T) {
|
|||||||
|
|
||||||
// TestHandler_GC_PerRepoDedup ensures duplicate-pruning does not evict
|
// TestHandler_GC_PerRepoDedup ensures duplicate-pruning does not evict
|
||||||
// another repo's entry. Two repos reserve the same (key, version); after the
|
// another repo's entry. Two repos reserve the same (key, version); after the
|
||||||
// keepOld window, GC must keep the one from each repo.
|
// inUseGrace window, GC must keep the one from each repo.
|
||||||
func TestHandler_GC_PerRepoDedup(t *testing.T) {
|
func TestHandler_GC_PerRepoDedup(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
handler.RegisterJob("tok-a", JobCredential{Repo: "owner/repoA"})
|
handler.RegisterJob("tok-a", JobCredential{Repo: "owner/repoA"})
|
||||||
@@ -1142,7 +1400,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) {
|
|||||||
db, err := handler.openDB()
|
db, err := handler.openDB()
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
now := time.Now().Unix()
|
now := time.Now().Unix()
|
||||||
stale := time.Now().Add(-keepOld - time.Minute).Unix()
|
stale := time.Now().Add(-inUseGrace - time.Minute).Unix()
|
||||||
a := &Cache{Repo: "owner/repoA", Key: key, Version: version, Complete: true, CreatedAt: stale, UsedAt: stale, Size: 1}
|
a := &Cache{Repo: "owner/repoA", Key: key, Version: version, Complete: true, CreatedAt: stale, UsedAt: stale, Size: 1}
|
||||||
b := &Cache{Repo: "owner/repoB", Key: key, Version: version, Complete: true, CreatedAt: now, UsedAt: now, Size: 1}
|
b := &Cache{Repo: "owner/repoB", Key: key, Version: version, Complete: true, CreatedAt: now, UsedAt: now, Size: 1}
|
||||||
require.NoError(t, insertCache(db, a))
|
require.NoError(t, insertCache(db, a))
|
||||||
@@ -1179,7 +1437,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) {
|
|||||||
// register/revoke when the feature is off.
|
// register/revoke when the feature is off.
|
||||||
func TestHandler_InternalAPI_Disabled(t *testing.T) {
|
func TestHandler_InternalAPI_Disabled(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: dir})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -1197,7 +1455,7 @@ func TestHandler_InternalAPI_Disabled(t *testing.T) {
|
|||||||
func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) {
|
func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
const secret = "internal-secret"
|
const secret = "internal-secret"
|
||||||
handler, err := StartHandler(dir, "", 0, secret, nil)
|
handler, err := StartHandler(Options{Dir: dir, InternalSecret: secret})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
|
|||||||
@@ -77,6 +77,7 @@ func (h *Handler) v2CreateCacheEntry(w http.ResponseWriter, r *http.Request, _ h
|
|||||||
h.twirpError(w, r, twirpInternal, err)
|
h.twirpError(w, r, twirpInternal, err)
|
||||||
return
|
return
|
||||||
} else if existing != nil {
|
} else if existing != nil {
|
||||||
|
h.touch(db, existing) // the client skips the upload, so this is the only sign the entry is still in use
|
||||||
h.twirpNotOK(w, r)
|
h.twirpNotOK(w, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -97,7 +98,7 @@ func (h *Handler) v2CreateCacheEntry(w http.ResponseWriter, r *http.Request, _ h
|
|||||||
|
|
||||||
h.responseJSON(w, r, http.StatusOK, map[string]any{
|
h.responseJSON(w, r, http.StatusOK, map[string]any{
|
||||||
"ok": true,
|
"ok": true,
|
||||||
"signed_upload_url": h.signedURL(blobPath, blobUploadPurpose, cache.ID, time.Now().Add(blobUploadURLTTL)),
|
"signed_upload_url": h.signedURL(cred, blobPath, blobUploadPurpose, cache.ID, time.Now().Add(blobUploadURLTTL)),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -168,7 +169,7 @@ func (h *Handler) v2GetCacheEntryDownloadURL(w http.ResponseWriter, r *http.Requ
|
|||||||
|
|
||||||
h.responseJSON(w, r, http.StatusOK, map[string]any{
|
h.responseJSON(w, r, http.StatusOK, map[string]any{
|
||||||
"ok": true,
|
"ok": true,
|
||||||
"signed_download_url": h.signedArtifactURL(cache.ID, time.Now().Add(artifactURLTTL)),
|
"signed_download_url": h.signedArtifactURL(cred, cache.ID, time.Now().Add(artifactURLTTL)),
|
||||||
"matched_key": cache.Key,
|
"matched_key": cache.Key,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -10,8 +10,8 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"path/filepath"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -66,16 +66,6 @@ func getURL(t *testing.T, url string) []byte {
|
|||||||
return body
|
return body
|
||||||
}
|
}
|
||||||
|
|
||||||
func startTestHandler(t *testing.T) *Handler {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
handler, err := StartHandler(filepath.Join(t.TempDir(), "artifactcache"), "127.0.0.1", 0, "", nil)
|
|
||||||
require.NoError(t, err)
|
|
||||||
t.Cleanup(func() { _ = handler.Close() })
|
|
||||||
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
|
||||||
return handler
|
|
||||||
}
|
|
||||||
|
|
||||||
// saveV2 runs the reserve/upload/finalize sequence and returns the finalize response along
|
// saveV2 runs the reserve/upload/finalize sequence and returns the finalize response along
|
||||||
// with the upload URL it used.
|
// with the upload URL it used.
|
||||||
func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) (finalized map[string]any, uploadURL string) {
|
func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) (finalized map[string]any, uploadURL string) {
|
||||||
@@ -97,7 +87,7 @@ func saveV2(t *testing.T, handler *Handler, key, version string, content []byte)
|
|||||||
// URLs it is handed: unsigned requests are refused, an upload URL cannot be replayed to read
|
// URLs it is handed: unsigned requests are refused, an upload URL cannot be replayed to read
|
||||||
// or to replace a finalized entry.
|
// or to replace a finalized entry.
|
||||||
func TestCacheServiceV2RoundTrip(t *testing.T) {
|
func TestCacheServiceV2RoundTrip(t *testing.T) {
|
||||||
handler := startTestHandler(t)
|
handler := newTestHandler(t, Policy{})
|
||||||
content := []byte("the cached archive")
|
content := []byte("the cached archive")
|
||||||
|
|
||||||
unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath)
|
unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath)
|
||||||
@@ -126,7 +116,7 @@ func TestCacheServiceV2RoundTrip(t *testing.T) {
|
|||||||
// A large archive is staged as blocks and only put in order by the final block list, so
|
// A large archive is staged as blocks and only put in order by the final block list, so
|
||||||
// blocks that arrive out of order must still be assembled the way the client asked.
|
// blocks that arrive out of order must still be assembled the way the client asked.
|
||||||
func TestCacheServiceV2BlockUpload(t *testing.T) {
|
func TestCacheServiceV2BlockUpload(t *testing.T) {
|
||||||
handler := startTestHandler(t)
|
handler := newTestHandler(t, Policy{})
|
||||||
|
|
||||||
created := v2Call(t, handler, testClient, "CreateCacheEntry", map[string]any{"key": "blocks", "version": "v1"})
|
created := v2Call(t, handler, testClient, "CreateCacheEntry", map[string]any{"key": "blocks", "version": "v1"})
|
||||||
uploadURL, _ := created["signed_upload_url"].(string)
|
uploadURL, _ := created["signed_upload_url"].(string)
|
||||||
@@ -163,7 +153,7 @@ func TestCacheServiceV2BlockUpload(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestCacheServiceV2Lookups(t *testing.T) {
|
func TestCacheServiceV2Lookups(t *testing.T) {
|
||||||
handler := startTestHandler(t)
|
handler := newTestHandler(t, Policy{})
|
||||||
saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x"))
|
saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x"))
|
||||||
require.Equal(t, true, saved["ok"])
|
require.Equal(t, true, saved["ok"])
|
||||||
|
|
||||||
@@ -198,6 +188,18 @@ func TestCacheServiceV2Lookups(t *testing.T) {
|
|||||||
assert.NotEmpty(t, reserved["signed_upload_url"])
|
assert.NotEmpty(t, reserved["signed_upload_url"])
|
||||||
})
|
})
|
||||||
|
|
||||||
|
t.Run("a proxied job is handed the address its runner registered", func(t *testing.T) {
|
||||||
|
const proxy = "https://cache.example.invalid"
|
||||||
|
handler.RegisterJob("proxied", JobCredential{Repo: testRepo, PublicURL: proxy + "/"})
|
||||||
|
client := &http.Client{Transport: &bearerTransport{token: "proxied"}}
|
||||||
|
|
||||||
|
created := v2Call(t, handler, client, "CreateCacheEntry", map[string]any{"key": "proxied-key", "version": "v1"})
|
||||||
|
assert.True(t, strings.HasPrefix(created["signed_upload_url"].(string), proxy+blobPath+"/"))
|
||||||
|
|
||||||
|
got := v2Call(t, handler, client, "GetCacheEntryDownloadURL", map[string]any{"key": "deps-abc", "version": "v1"})
|
||||||
|
assert.True(t, strings.HasPrefix(got["signed_download_url"].(string), proxy+apiPath+"/artifacts/"))
|
||||||
|
})
|
||||||
|
|
||||||
t.Run("finalizing without a reservation is not ok", func(t *testing.T) {
|
t.Run("finalizing without a reservation is not ok", func(t *testing.T) {
|
||||||
got := v2Call(t, handler, testClient, "FinalizeCacheEntryUpload", map[string]any{
|
got := v2Call(t, handler, testClient, "FinalizeCacheEntryUpload", map[string]any{
|
||||||
"key": "never-reserved", "version": "v1", "size_bytes": 1,
|
"key": "never-reserved", "version": "v1", "size_bytes": 1,
|
||||||
|
|||||||
@@ -24,7 +24,7 @@ func TestFrontResultsService(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer gitea.Close()
|
defer gitea.Close()
|
||||||
|
|
||||||
handler, err := StartHandler(t.TempDir(), "127.0.0.1", 0, "", nil)
|
handler, err := StartHandler(Options{Dir: t.TempDir(), OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
const token = "forward-token"
|
const token = "forward-token"
|
||||||
|
|||||||
@@ -143,9 +143,13 @@ func (s *Storage) Serve(w http.ResponseWriter, r *http.Request, id uint64) {
|
|||||||
http.ServeFile(w, r, name)
|
http.ServeFile(w, r, name)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Storage) Remove(id uint64) {
|
// Remove deletes an entry's blob and any staged parts. It reports failure so the caller can
|
||||||
_ = os.Remove(s.filename(id))
|
// keep the entry and retry, rather than dropping the only reference to bytes on disk.
|
||||||
_ = os.RemoveAll(s.tempDir(id))
|
func (s *Storage) Remove(id uint64) error {
|
||||||
|
if err := os.Remove(s.filename(id)); err != nil && !os.IsNotExist(err) {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return os.RemoveAll(s.tempDir(id))
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *Storage) filename(id uint64) string {
|
func (s *Storage) filename(id uint64) string {
|
||||||
|
|||||||
+15
-2
@@ -345,6 +345,16 @@ func gitOptions(token string) (fetchOptions git.FetchOptions, pullOptions git.Pu
|
|||||||
return fetchOptions, pullOptions
|
return fetchOptions, pullOptions
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// staleRefreshErr reports why a failed refresh must abort: the resolve and
|
||||||
|
// checkout that follow are local and succeed on a cancelled context, which
|
||||||
|
// would hand back the cached revision as if it were fresh.
|
||||||
|
func staleRefreshErr(ctx context.Context, err error) error {
|
||||||
|
if err == nil || errors.Is(err, git.NoErrAlreadyUpToDate) {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
|
|
||||||
// NewGitCloneExecutor creates an executor to clone git repos
|
// NewGitCloneExecutor creates an executor to clone git repos
|
||||||
func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
|
func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
|
||||||
return func(ctx context.Context) error {
|
return func(ctx context.Context) error {
|
||||||
@@ -385,7 +395,7 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if !isOfflineMode {
|
if !isOfflineMode {
|
||||||
err = r.Fetch(&fetchOptions)
|
err = r.FetchContext(ctx, &fetchOptions)
|
||||||
if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
|
if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -454,9 +464,12 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor {
|
|||||||
switch {
|
switch {
|
||||||
case !isOfflineMode && !shallow:
|
case !isOfflineMode && !shallow:
|
||||||
// In shallow mode the depth-limited fetch above already advanced the ref.
|
// In shallow mode the depth-limited fetch above already advanced the ref.
|
||||||
if err = w.Pull(&pullOptions); err != nil && err != git.NoErrAlreadyUpToDate {
|
if err = w.PullContext(ctx, &pullOptions); err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) {
|
||||||
logger.Debugf("Unable to pull %s: %v", refName, err)
|
logger.Debugf("Unable to pull %s: %v", refName, err)
|
||||||
}
|
}
|
||||||
|
if err := staleRefreshErr(ctx, err); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
case isOfflineMode && reused:
|
case isOfflineMode && reused:
|
||||||
reusedMsg = " (reused in offline mode)"
|
reusedMsg = " (reused in offline mode)"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,18 +6,24 @@ package git
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
"os"
|
"os"
|
||||||
"os/exec"
|
"os/exec"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"syscall"
|
"syscall"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/act/common"
|
||||||
|
|
||||||
|
gogit "github.com/go-git/go-git/v5"
|
||||||
|
gogitconfig "github.com/go-git/go-git/v5/config"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
logrustest "github.com/sirupsen/logrus/hooks/test"
|
logrustest "github.com/sirupsen/logrus/hooks/test"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -610,3 +616,55 @@ func TestAcquireCloneLock(t *testing.T) {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// An unresponsive remote must not pin a job: the refresh has to be interruptible.
|
||||||
|
func TestNewGitCloneExecutorFetchHonoursContext(t *testing.T) {
|
||||||
|
block := make(chan struct{})
|
||||||
|
reached := make(chan struct{})
|
||||||
|
var once sync.Once
|
||||||
|
server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) {
|
||||||
|
once.Do(func() { close(reached) })
|
||||||
|
<-block
|
||||||
|
}))
|
||||||
|
t.Cleanup(func() {
|
||||||
|
close(block)
|
||||||
|
server.Close()
|
||||||
|
})
|
||||||
|
|
||||||
|
dir := filepath.Join(t.TempDir(), "cached-action")
|
||||||
|
repo, err := gogit.PlainInit(dir, false)
|
||||||
|
require.NoError(t, err)
|
||||||
|
_, err = repo.CreateRemote(&gogitconfig.RemoteConfig{Name: "origin", URLs: []string{server.URL}})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
done := make(chan error, 1)
|
||||||
|
go func() {
|
||||||
|
done <- NewGitCloneExecutor(NewGitCloneExecutorInput{URL: server.URL, Ref: "main", Dir: dir})(ctx)
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-reached:
|
||||||
|
case <-time.After(10 * time.Second):
|
||||||
|
t.Fatal("the executor never reached the remote")
|
||||||
|
}
|
||||||
|
cancel()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case err := <-done:
|
||||||
|
require.Error(t, err)
|
||||||
|
case <-time.After(10 * time.Second):
|
||||||
|
t.Fatal("fetch ignored context cancellation")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestStaleRefreshErr(t *testing.T) {
|
||||||
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
|
require.NoError(t, staleRefreshErr(ctx, errors.New("remote hung up")))
|
||||||
|
|
||||||
|
cancel()
|
||||||
|
require.ErrorIs(t, staleRefreshErr(ctx, errors.New("remote hung up")), context.Canceled)
|
||||||
|
require.NoError(t, staleRefreshErr(ctx, gogit.NoErrAlreadyUpToDate))
|
||||||
|
}
|
||||||
|
|||||||
@@ -88,14 +88,14 @@ func (cr *containerReference) connectToNetwork(name string, aliases []string) co
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// supportsContainerImagePlatform returns true if the underlying Docker server
|
// supportsContainerImagePlatform reports whether the Docker server API version
|
||||||
// API version is 1.41 and beyond
|
// is 1.41 and beyond
|
||||||
func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) bool {
|
func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) (bool, error) {
|
||||||
ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{})
|
ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
common.Logger(ctx).Panicf("Failed to get Docker API Version: %s", err)
|
return false, fmt.Errorf("get docker API version: %w", err)
|
||||||
}
|
}
|
||||||
return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41")
|
return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41"), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor {
|
func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor {
|
||||||
@@ -682,11 +682,17 @@ func (cr *containerReference) create(capAdd, capDrop []string) common.Executor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
var platSpecs *specs.Platform
|
var platSpecs *specs.Platform
|
||||||
if cr.input.Platform != "" && supportsContainerImagePlatform(ctx, cr.cli) {
|
if cr.input.Platform != "" {
|
||||||
platSpecs, err = parsePlatform(cr.input.Platform)
|
// Dropping the platform silently would build for the host arch.
|
||||||
|
supported, err := supportsContainerImagePlatform(ctx, cr.cli)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
if supported {
|
||||||
|
if platSpecs, err = parsePlatform(cr.input.Platform); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
hostConfig := &container.HostConfig{
|
hostConfig := &container.HostConfig{
|
||||||
|
|||||||
@@ -79,6 +79,11 @@ type mockDockerClient struct {
|
|||||||
mock.Mock
|
mock.Mock
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *mockDockerClient) ServerVersion(ctx context.Context, opts mobyclient.ServerVersionOptions) (mobyclient.ServerVersionResult, error) {
|
||||||
|
args := m.Called(ctx, opts)
|
||||||
|
return args.Get(0).(mobyclient.ServerVersionResult), args.Error(1)
|
||||||
|
}
|
||||||
|
|
||||||
func (m *mockDockerClient) ExecCreate(ctx context.Context, id string, opts mobyclient.ExecCreateOptions) (mobyclient.ExecCreateResult, error) {
|
func (m *mockDockerClient) ExecCreate(ctx context.Context, id string, opts mobyclient.ExecCreateOptions) (mobyclient.ExecCreateResult, error) {
|
||||||
args := m.Called(ctx, id, opts)
|
args := m.Called(ctx, id, opts)
|
||||||
return args.Get(0).(mobyclient.ExecCreateResult), args.Error(1)
|
return args.Get(0).(mobyclient.ExecCreateResult), args.Error(1)
|
||||||
@@ -930,3 +935,15 @@ func TestMergeContainerConfigsVolumesReplaceRunnerMounts(t *testing.T) {
|
|||||||
assert.Equal(t, []string{"/var/run/docker.sock:/var/run/docker.sock", "/host/tools:/opt/hostedtoolcache"}, hostConf.Binds)
|
assert.Equal(t, []string{"/var/run/docker.sock:/var/run/docker.sock", "/host/tools:/opt/hostedtoolcache"}, hostConf.Binds)
|
||||||
assert.Empty(t, hostConf.Mounts)
|
assert.Empty(t, hostConf.Mounts)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A dead daemon must fail the job, not panic through logrus and not silently
|
||||||
|
// drop the requested platform.
|
||||||
|
func TestSupportsContainerImagePlatformDaemonError(t *testing.T) {
|
||||||
|
cli := &mockDockerClient{}
|
||||||
|
cli.On("ServerVersion", mock.Anything, mock.Anything).
|
||||||
|
Return(mobyclient.ServerVersionResult{}, errors.New("cannot connect to the Docker daemon"))
|
||||||
|
|
||||||
|
supported, err := supportsContainerImagePlatform(t.Context(), cli)
|
||||||
|
require.ErrorContains(t, err, "cannot connect to the Docker daemon")
|
||||||
|
assert.False(t, supported)
|
||||||
|
}
|
||||||
|
|||||||
@@ -229,14 +229,19 @@ func (rc *RunContext) containerDaemonSocket() string {
|
|||||||
return rc.Config.ContainerDaemonSocket
|
return rc.Config.ContainerDaemonSocket
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const sharedToolCacheVolume = "act-toolcache" // mounted only when the tool cache is shared
|
||||||
|
|
||||||
// validVolumes returns the volumes allowed on this job's containers: the configured base
|
// validVolumes returns the volumes allowed on this job's containers: the configured base
|
||||||
// plus the volumes the runner mounts automatically. It derives a fresh slice every call and
|
// plus the volumes the runner mounts automatically. It derives a fresh slice every call and
|
||||||
// never mutates the shared Config (see containerDaemonSocket).
|
// never mutates the shared Config (see containerDaemonSocket).
|
||||||
func (rc *RunContext) validVolumes() []string {
|
func (rc *RunContext) validVolumes() []string {
|
||||||
name := rc.jobContainerName()
|
name := rc.jobContainerName()
|
||||||
volumes := slices.Clone(rc.Config.ValidVolumes)
|
volumes := slices.Clone(rc.Config.ValidVolumes)
|
||||||
|
if rc.Config.SharedToolCache {
|
||||||
|
volumes = append(volumes, sharedToolCacheVolume)
|
||||||
|
}
|
||||||
// TODO: add a new configuration to control whether the docker daemon can be mounted
|
// TODO: add a new configuration to control whether the docker daemon can be mounted
|
||||||
return append(volumes, "act-toolcache", name, name+"-env",
|
return append(volumes, name, name+"-env",
|
||||||
getDockerDaemonSocketMountPath(rc.containerDaemonSocket()))
|
getDockerDaemonSocketMountPath(rc.containerDaemonSocket()))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -309,8 +314,10 @@ func (rc *RunContext) GetBindsAndMounts() ([]string, map[string]string) {
|
|||||||
if daemonSocket := rc.containerDaemonSocket(); daemonSocket != "-" && !claimed["/var/run/docker.sock"] {
|
if daemonSocket := rc.containerDaemonSocket(); daemonSocket != "-" && !claimed["/var/run/docker.sock"] {
|
||||||
binds = append(binds, getDockerDaemonSocketMountPath(daemonSocket)+":/var/run/docker.sock")
|
binds = append(binds, getDockerDaemonSocketMountPath(daemonSocket)+":/var/run/docker.sock")
|
||||||
}
|
}
|
||||||
|
if rc.Config.SharedToolCache {
|
||||||
if toolCache := rc.toolCache(container.DefaultToolCache); !claimed[toolCache] {
|
if toolCache := rc.toolCache(container.DefaultToolCache); !claimed[toolCache] {
|
||||||
mounts["act-toolcache"] = toolCache
|
mounts[sharedToolCacheVolume] = toolCache
|
||||||
|
}
|
||||||
}
|
}
|
||||||
mounts[name+"-env"] = ext.GetActPath() // runner-internal, never overridable
|
mounts[name+"-env"] = ext.GetActPath() // runner-internal, never overridable
|
||||||
|
|
||||||
@@ -360,7 +367,11 @@ func (rc *RunContext) startHostEnvironment() common.Executor {
|
|||||||
if err := os.MkdirAll(runnerTmp, 0o777); err != nil {
|
if err := os.MkdirAll(runnerTmp, 0o777); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
toolCache := rc.toolCache(filepath.Join(cacheDir, "tool_cache"))
|
toolCacheParent := miscpath // per job, so cleanup removes it with the job
|
||||||
|
if rc.Config.SharedToolCache {
|
||||||
|
toolCacheParent = cacheDir
|
||||||
|
}
|
||||||
|
toolCache := rc.toolCache(filepath.Join(toolCacheParent, "tool_cache"))
|
||||||
if err := os.MkdirAll(toolCache, 0o777); err != nil {
|
if err := os.MkdirAll(toolCache, 0o777); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -503,6 +503,7 @@ func TestRunContext_GetBindsAndMounts(t *testing.T) {
|
|||||||
},
|
},
|
||||||
Config: &Config{
|
Config: &Config{
|
||||||
BindWorkdir: false,
|
BindWorkdir: false,
|
||||||
|
SharedToolCache: true, // so OverridesToolCache has a mount to displace
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
rc.Run.JobID = "job1"
|
rc.Run.JobID = "job1"
|
||||||
@@ -543,25 +544,44 @@ func TestRunContext_GetBindsAndMounts(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
t.Run("ToolCacheMount", func(t *testing.T) {
|
||||||
|
rc := &RunContext{
|
||||||
|
Name: "TestRCName",
|
||||||
|
Run: &model.Run{Workflow: &model.Workflow{Name: "TestWorkflowName"}},
|
||||||
|
Config: &Config{},
|
||||||
|
}
|
||||||
|
|
||||||
|
_, gotmount := rc.GetBindsAndMounts()
|
||||||
|
assert.NotContains(t, gotmount, sharedToolCacheVolume)
|
||||||
|
|
||||||
|
rc.Config.SharedToolCache = true
|
||||||
|
_, gotmount = rc.GetBindsAndMounts()
|
||||||
|
assert.Equal(t, container.DefaultToolCache, gotmount[sharedToolCacheVolume])
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRunContextValidVolumes(t *testing.T) {
|
func TestRunContextValidVolumes(t *testing.T) {
|
||||||
rc := &RunContext{
|
rc := &RunContext{
|
||||||
Name: "job",
|
Name: "job",
|
||||||
Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}},
|
Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}},
|
||||||
Config: &Config{ValidVolumes: []string{"my-vol", "/host/path"}},
|
Config: &Config{ValidVolumes: []string{"my-vol", "/host/path"}, SharedToolCache: true},
|
||||||
}
|
}
|
||||||
name := rc.jobContainerName()
|
name := rc.jobContainerName()
|
||||||
|
|
||||||
got := rc.validVolumes()
|
got := rc.validVolumes()
|
||||||
|
|
||||||
// the configured volumes plus the four the runner mounts automatically
|
// the configured volumes plus the ones the runner mounts automatically
|
||||||
assert.Subset(t, got, []string{"my-vol", "/host/path", "act-toolcache", name, name + "-env", "/var/run/docker.sock"})
|
assert.Subset(t, got, []string{"my-vol", "/host/path", sharedToolCacheVolume, name, name + "-env", "/var/run/docker.sock"})
|
||||||
|
|
||||||
// deriving the list must never mutate or grow the shared Config slice: parallel matrix
|
// deriving the list must never mutate or grow the shared Config slice: parallel matrix
|
||||||
// combinations share one *Config, and the previous in-place append was a data race.
|
// combinations share one *Config, and the previous in-place append was a data race.
|
||||||
assert.Equal(t, []string{"my-vol", "/host/path"}, rc.Config.ValidVolumes)
|
assert.Equal(t, []string{"my-vol", "/host/path"}, rc.Config.ValidVolumes)
|
||||||
assert.Len(t, rc.validVolumes(), len(got), "repeated calls must be stable, not accumulate")
|
assert.Len(t, rc.validVolumes(), len(got), "repeated calls must be stable, not accumulate")
|
||||||
|
|
||||||
|
// a job may mount it only while the runner does
|
||||||
|
rc.Config.SharedToolCache = false
|
||||||
|
assert.NotContains(t, rc.validVolumes(), sharedToolCacheVolume)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestCleanupJobResourcesCleansServicesWithoutJobContainer(t *testing.T) {
|
func TestCleanupJobResourcesCleansServicesWithoutJobContainer(t *testing.T) {
|
||||||
|
|||||||
@@ -91,6 +91,7 @@ type Config struct {
|
|||||||
PlatformPicker func(labels []string) string // platform picker, it will take precedence over Platforms if isn't nil
|
PlatformPicker func(labels []string) string // platform picker, it will take precedence over Platforms if isn't nil
|
||||||
JobLoggerLevel *log.Level // the level of job logger
|
JobLoggerLevel *log.Level // the level of job logger
|
||||||
ValidVolumes []string // only volumes (and bind mounts) in this slice can be mounted on the job container or service containers
|
ValidVolumes []string // only volumes (and bind mounts) in this slice can be mounted on the job container or service containers
|
||||||
|
SharedToolCache bool // one tool cache for all jobs instead of one per job
|
||||||
InsecureSkipTLS bool // whether to skip verifying TLS certificate of the Gitea instance
|
InsecureSkipTLS bool // whether to skip verifying TLS certificate of the Gitea instance
|
||||||
MaxParallel int // max parallel jobs to run across all workflows (0 = no limit, uses CPU count)
|
MaxParallel int // max parallel jobs to run across all workflows (0 = no limit, uses CPU count)
|
||||||
AllocatePTY bool // allocate a pseudo-TTY for each step's process
|
AllocatePTY bool // allocate a pseudo-TTY for each step's process
|
||||||
|
|||||||
@@ -144,7 +144,7 @@ func TestCacheServiceV2EndToEnd(t *testing.T) {
|
|||||||
restore := patchedAction(t, "actions/cache", actionsCacheRef, "dist/restore/index.js")
|
restore := patchedAction(t, "actions/cache", actionsCacheRef, "dist/restore/index.js")
|
||||||
save := patchedAction(t, "actions/cache", actionsCacheRef, "dist/save/index.js")
|
save := patchedAction(t, "actions/cache", actionsCacheRef, "dist/save/index.js")
|
||||||
|
|
||||||
handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil)
|
handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
t.Cleanup(func() { _ = handler.Close() })
|
t.Cleanup(func() { _ = handler.Close() })
|
||||||
const token, repo = "e2e-runtime-token", "testuser/testrepo"
|
const token, repo = "e2e-runtime-token", "testuser/testrepo"
|
||||||
@@ -291,7 +291,7 @@ func TestUploadArtifactThroughTheResultsService(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer gitea.Close()
|
defer gitea.Close()
|
||||||
|
|
||||||
handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil)
|
handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
t.Cleanup(func() { _ = handler.Close() })
|
t.Cleanup(func() { _ = handler.Close() })
|
||||||
// The artifact client decodes the runtime token for the run ids it puts in its requests, where
|
// The artifact client decodes the runtime token for the run ids it puts in its requests, where
|
||||||
@@ -345,7 +345,7 @@ func TestSetupActionFindsTheCacheService(t *testing.T) {
|
|||||||
|
|
||||||
setup := patchedAction(t, "actions/setup-node", "v7.0.0", "dist/setup/index.js")
|
setup := patchedAction(t, "actions/setup-node", "v7.0.0", "dist/setup/index.js")
|
||||||
|
|
||||||
handler, err := artifactcache.StartHandler(filepath.Join(t.TempDir(), "cache"), "127.0.0.1", 0, "", nil)
|
handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: filepath.Join(t.TempDir(), "cache"), OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
t.Cleanup(func() { _ = handler.Close() })
|
t.Cleanup(func() { _ = handler.Close() })
|
||||||
const token = "setup-runtime-token"
|
const token = "setup-runtime-token"
|
||||||
|
|||||||
@@ -11,6 +11,8 @@ Each example persists **two** things, and it is worth knowing which is which:
|
|||||||
- `/data` is the runner's working directory. It holds the `.runner` registration file and, optionally, the config file — so the runner re-attaches to the server instead of registering again.
|
- `/data` is the runner's working directory. It holds the `.runner` registration file and, optionally, the config file — so the runner re-attaches to the server instead of registering again.
|
||||||
- The Docker daemon's data root holds the images pulled for jobs (`/var/lib/docker` for the dind sidecar, `/home/rootless/.local/share/docker` for `dind-rootless`). It is *not* under `/data`. If you drop this volume, the examples still work, but the image cache is discarded whenever the pod is recreated and every job re-pulls its images.
|
- The Docker daemon's data root holds the images pulled for jobs (`/var/lib/docker` for the dind sidecar, `/home/rootless/.local/share/docker` for `dind-rootless`). It is *not* under `/data`. If you drop this volume, the examples still work, but the image cache is discarded whenever the pod is recreated and every job re-pulls its images.
|
||||||
|
|
||||||
|
- Kubernetes SIGKILLs a pod 30s after SIGTERM by default, long before a job finishes and reports its result, which leaves tasks the server can only reap as zombies. The manifests raise `terminationGracePeriodSeconds` to three hours, matching the systemd example and the `runner.timeout` job ceiling; set `runner.shutdown_timeout` below that so the runner drains jobs within the window rather than being killed mid-cleanup.
|
||||||
|
|
||||||
Files in this directory:
|
Files in this directory:
|
||||||
|
|
||||||
- [`dind-docker.yaml`](dind-docker.yaml)
|
- [`dind-docker.yaml`](dind-docker.yaml)
|
||||||
|
|||||||
@@ -56,6 +56,7 @@ spec:
|
|||||||
app: runner
|
app: runner
|
||||||
spec:
|
spec:
|
||||||
restartPolicy: Always
|
restartPolicy: Always
|
||||||
|
terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README
|
||||||
volumes:
|
volumes:
|
||||||
- name: docker-socket
|
- name: docker-socket
|
||||||
emptyDir: {}
|
emptyDir: {}
|
||||||
|
|||||||
@@ -0,0 +1,38 @@
|
|||||||
|
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||||
|
// SPDX-License-Identifier: MIT
|
||||||
|
|
||||||
|
package kubernetes_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"regexp"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
)
|
||||||
|
|
||||||
|
var gracePeriod = regexp.MustCompile(`terminationGracePeriodSeconds: (\d+)`)
|
||||||
|
|
||||||
|
// Without it Kubernetes SIGKILLs the pod 30s after SIGTERM, mid-job.
|
||||||
|
func TestManifestsSetTerminationGracePeriod(t *testing.T) {
|
||||||
|
files, err := filepath.Glob("*.yaml")
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.NotEmpty(t, files)
|
||||||
|
|
||||||
|
for _, file := range files {
|
||||||
|
content, err := os.ReadFile(file)
|
||||||
|
require.NoError(t, err)
|
||||||
|
if !strings.Contains(string(content), "containers:") {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
match := gracePeriod.FindStringSubmatch(string(content))
|
||||||
|
require.NotNil(t, match, file)
|
||||||
|
seconds, err := strconv.Atoi(match[1])
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.GreaterOrEqual(t, seconds, 3600, file)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -56,6 +56,7 @@ spec:
|
|||||||
app: runner
|
app: runner
|
||||||
spec:
|
spec:
|
||||||
restartPolicy: Always
|
restartPolicy: Always
|
||||||
|
terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README
|
||||||
volumes:
|
volumes:
|
||||||
- name: runner-data
|
- name: runner-data
|
||||||
persistentVolumeClaim:
|
persistentVolumeClaim:
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ spec:
|
|||||||
app: runner
|
app: runner
|
||||||
spec:
|
spec:
|
||||||
restartPolicy: Always
|
restartPolicy: Always
|
||||||
|
terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README
|
||||||
volumes:
|
volumes:
|
||||||
- name: docker-socket
|
- name: docker-socket
|
||||||
emptyDir: {}
|
emptyDir: {}
|
||||||
|
|||||||
@@ -12,8 +12,9 @@ require (
|
|||||||
github.com/containerd/errdefs v1.0.0
|
github.com/containerd/errdefs v1.0.0
|
||||||
github.com/creack/pty v1.1.24
|
github.com/creack/pty v1.1.24
|
||||||
github.com/distribution/reference v0.6.0
|
github.com/distribution/reference v0.6.0
|
||||||
github.com/docker/cli v29.6.2+incompatible
|
github.com/docker/cli v29.7.1+incompatible
|
||||||
github.com/docker/go-connections v0.8.1
|
github.com/docker/go-connections v0.8.1
|
||||||
|
github.com/docker/go-units v0.5.0
|
||||||
github.com/go-git/go-billy/v5 v5.9.1
|
github.com/go-git/go-billy/v5 v5.9.1
|
||||||
github.com/go-git/go-git/v5 v5.19.2
|
github.com/go-git/go-git/v5 v5.19.2
|
||||||
github.com/gobwas/glob v0.2.3
|
github.com/gobwas/glob v0.2.3
|
||||||
@@ -22,7 +23,7 @@ require (
|
|||||||
github.com/julienschmidt/httprouter v1.3.0
|
github.com/julienschmidt/httprouter v1.3.0
|
||||||
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
|
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
|
||||||
github.com/mattn/go-isatty v0.0.24
|
github.com/mattn/go-isatty v0.0.24
|
||||||
github.com/moby/go-archive v0.2.1
|
github.com/moby/go-archive v0.3.2
|
||||||
github.com/moby/moby/api v1.55.0
|
github.com/moby/moby/api v1.55.0
|
||||||
github.com/moby/moby/client v0.5.1
|
github.com/moby/moby/client v0.5.1
|
||||||
github.com/moby/patternmatcher v0.6.1
|
github.com/moby/patternmatcher v0.6.1
|
||||||
@@ -61,7 +62,6 @@ require (
|
|||||||
github.com/cyphar/filepath-securejoin v0.6.1 // indirect
|
github.com/cyphar/filepath-securejoin v0.6.1 // indirect
|
||||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||||
github.com/docker/docker-credential-helpers v0.9.6 // indirect
|
github.com/docker/docker-credential-helpers v0.9.6 // indirect
|
||||||
github.com/docker/go-units v0.5.0 // indirect
|
|
||||||
github.com/emirpasic/gods v1.18.1 // indirect
|
github.com/emirpasic/gods v1.18.1 // indirect
|
||||||
github.com/fatih/color v1.19.0 // indirect
|
github.com/fatih/color v1.19.0 // indirect
|
||||||
github.com/felixge/httpsnoop v1.0.4 // indirect
|
github.com/felixge/httpsnoop v1.0.4 // indirect
|
||||||
|
|||||||
@@ -45,8 +45,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
|
|||||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||||
github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
|
github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
|
||||||
github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
|
github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
|
||||||
github.com/docker/cli v29.6.2+incompatible h1:/bjePvcbbFTnRrMfWJBY7AjfICdsiLVgHn6LwTVOcqw=
|
github.com/docker/cli v29.7.1+incompatible h1:ILZpP6B7fedIr6ANy824QkDp1WMJuouIq0O2SrBkB2w=
|
||||||
github.com/docker/cli v29.6.2+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
|
github.com/docker/cli v29.7.1+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
|
||||||
github.com/docker/docker-credential-helpers v0.9.6 h1:cT2PbRPSlnMmNTfT2TDMXRyQ1KMWHG7xoTLBcn1ZNv0=
|
github.com/docker/docker-credential-helpers v0.9.6 h1:cT2PbRPSlnMmNTfT2TDMXRyQ1KMWHG7xoTLBcn1ZNv0=
|
||||||
github.com/docker/docker-credential-helpers v0.9.6/go.mod h1:v1S+hepowrQXITkEfw6o4+BMbGot02wiKpzWhGUZK6c=
|
github.com/docker/docker-credential-helpers v0.9.6/go.mod h1:v1S+hepowrQXITkEfw6o4+BMbGot02wiKpzWhGUZK6c=
|
||||||
github.com/docker/go-connections v0.8.1 h1:JibmG5hULs5qXSr/cp/w3Pw5fZuStt4MOHMUExb29/M=
|
github.com/docker/go-connections v0.8.1 h1:JibmG5hULs5qXSr/cp/w3Pw5fZuStt4MOHMUExb29/M=
|
||||||
@@ -123,8 +123,8 @@ github.com/mattn/go-shellwords v1.0.12 h1:M2zGm7EW6UQJvDeQxo4T51eKPurbeFbe8WtebG
|
|||||||
github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y=
|
github.com/mattn/go-shellwords v1.0.12/go.mod h1:EZzvwXDESEeg03EKmM+RmDnNOPKG4lLtQsUlTZDWQ8Y=
|
||||||
github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
|
github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
|
||||||
github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
|
github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
|
||||||
github.com/moby/go-archive v0.2.1 h1:fAa0wUS/ikZKyx7o/1fhUYmhZ7RgpthdeoDhJvunTLc=
|
github.com/moby/go-archive v0.3.2 h1:x893kC3zRygv2C+k4Y9kMxYRPLCj4XEJB0srbAP06Hw=
|
||||||
github.com/moby/go-archive v0.2.1/go.mod h1:Npdv43fFqlhZW7Xo8fbm3ZMYFvAGNviUPqX21VERbcE=
|
github.com/moby/go-archive v0.3.2/go.mod h1:Npdv43fFqlhZW7Xo8fbm3ZMYFvAGNviUPqX21VERbcE=
|
||||||
github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc=
|
github.com/moby/moby/api v1.55.0 h1:2/sexvQyqIWS8pRSCFddBfpW2qE7vR7FCL+vN8pxwMc=
|
||||||
github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs=
|
github.com/moby/moby/api v1.55.0/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs=
|
||||||
github.com/moby/moby/client v0.5.1 h1:tYNaJno4c0HXz12y5BiqEDy0rVTYkWzI26lGvnTMiJw=
|
github.com/moby/moby/client v0.5.1 h1:tYNaJno4c0HXz12y5BiqEDy0rVTYkWzI26lGvnTMiJw=
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import (
|
|||||||
"os/signal"
|
"os/signal"
|
||||||
|
|
||||||
"gitea.com/gitea/runner/act/artifactcache"
|
"gitea.com/gitea/runner/act/artifactcache"
|
||||||
|
"gitea.com/gitea/runner/internal/app/run"
|
||||||
"gitea.com/gitea/runner/internal/pkg/config"
|
"gitea.com/gitea/runner/internal/pkg/config"
|
||||||
|
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
@@ -52,13 +53,14 @@ func runCacheServer(configFile *string, cacheArgs *cacheServerArgs) func(cmd *co
|
|||||||
if secret == "" {
|
if secret == "" {
|
||||||
return errors.New("cache.external_secret (or cache.external_secret_file) must be set for cache-server; configure the same value on each runner that points at this server via cache.external_server")
|
return errors.New("cache.external_secret (or cache.external_secret_file) must be set for cache-server; configure the same value on each runner that points at this server via cache.external_server")
|
||||||
}
|
}
|
||||||
cacheHandler, err := artifactcache.StartHandler(
|
cacheHandler, err := artifactcache.StartHandler(artifactcache.Options{
|
||||||
dir,
|
Dir: dir,
|
||||||
host,
|
OutboundIP: host,
|
||||||
port,
|
Port: port,
|
||||||
secret,
|
InternalSecret: secret,
|
||||||
log.StandardLogger().WithField("module", "cache_request"),
|
Policy: run.CachePolicy(cfg),
|
||||||
)
|
Logger: log.StandardLogger().WithField("module", "cache_request"),
|
||||||
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -158,6 +158,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu
|
|||||||
cfg.Runner.Insecure,
|
cfg.Runner.Insecure,
|
||||||
reg.UUID,
|
reg.UUID,
|
||||||
reg.Token,
|
reg.Token,
|
||||||
|
config.RequestTimeout,
|
||||||
)
|
)
|
||||||
|
|
||||||
runner := run.NewRunner(cfg, reg, cli)
|
runner := run.NewRunner(cfg, reg, cli)
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
"maps"
|
"maps"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"slices"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
@@ -22,6 +23,7 @@ import (
|
|||||||
"gitea.com/gitea/runner/act/common"
|
"gitea.com/gitea/runner/act/common"
|
||||||
"gitea.com/gitea/runner/act/runner"
|
"gitea.com/gitea/runner/act/runner"
|
||||||
"gitea.com/gitea/runner/internal/app/run"
|
"gitea.com/gitea/runner/internal/app/run"
|
||||||
|
"gitea.com/gitea/runner/internal/pkg/config"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
"github.com/joho/godotenv"
|
"github.com/joho/godotenv"
|
||||||
@@ -67,6 +69,15 @@ type executeArgs struct {
|
|||||||
cacheHandler *artifactcache.Handler
|
cacheHandler *artifactcache.Handler
|
||||||
network string
|
network string
|
||||||
githubInstance string
|
githubInstance string
|
||||||
|
toolCacheMode string
|
||||||
|
}
|
||||||
|
|
||||||
|
// sharedToolCache reports whether mode mounts one tool cache for every job.
|
||||||
|
func sharedToolCache(mode string) (bool, error) {
|
||||||
|
if !slices.Contains(config.ToolCacheModes, mode) {
|
||||||
|
return false, fmt.Errorf("invalid --tool-cache-mode %q: must be one of %q", mode, config.ToolCacheModes)
|
||||||
|
}
|
||||||
|
return mode == config.ToolCacheModeShared, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// WorkflowsPath returns path to workflow file(s)
|
// WorkflowsPath returns path to workflow file(s)
|
||||||
@@ -374,7 +385,10 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
|||||||
}
|
}
|
||||||
|
|
||||||
// init a cache server
|
// init a cache server
|
||||||
handler, err := artifactcache.StartHandler("", "", 0, "", log.StandardLogger().WithField("module", "cache_request"))
|
handler, err := artifactcache.StartHandler(artifactcache.Options{
|
||||||
|
Policy: run.CachePolicy(&config.Config{Cache: config.DefaultCache()}),
|
||||||
|
Logger: log.StandardLogger().WithField("module", "cache_request"),
|
||||||
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -423,6 +437,11 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
|||||||
proxyEnv := run.JobProxyEnv(env, env["ACTIONS_CACHE_URL"], nil)
|
proxyEnv := run.JobProxyEnv(env, env["ACTIONS_CACHE_URL"], nil)
|
||||||
maps.Copy(env, proxyEnv)
|
maps.Copy(env, proxyEnv)
|
||||||
|
|
||||||
|
shared, err := sharedToolCache(execArgs.toolCacheMode)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
// run the plan
|
// run the plan
|
||||||
config := &runner.Config{
|
config := &runner.Config{
|
||||||
Workdir: execArgs.Workdir(),
|
Workdir: execArgs.Workdir(),
|
||||||
@@ -464,6 +483,7 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
|||||||
return execArgs.image
|
return execArgs.image
|
||||||
},
|
},
|
||||||
ValidVolumes: []string{"**"}, // All volumes are allowed for `exec` command
|
ValidVolumes: []string{"**"}, // All volumes are allowed for `exec` command
|
||||||
|
SharedToolCache: shared,
|
||||||
}
|
}
|
||||||
|
|
||||||
config.Env["ACT_EXEC"] = "true"
|
config.Env["ACT_EXEC"] = "true"
|
||||||
@@ -539,6 +559,7 @@ func loadExecCmd(ctx context.Context) *cobra.Command {
|
|||||||
execCmd.PersistentFlags().BoolVarP(&execArg.debug, "debug", "d", false, "enable debug log")
|
execCmd.PersistentFlags().BoolVarP(&execArg.debug, "debug", "d", false, "enable debug log")
|
||||||
execCmd.PersistentFlags().BoolVarP(&execArg.dryrun, "dryrun", "n", false, "dryrun mode")
|
execCmd.PersistentFlags().BoolVarP(&execArg.dryrun, "dryrun", "n", false, "dryrun mode")
|
||||||
execCmd.PersistentFlags().StringVarP(&execArg.image, "image", "i", "docker.gitea.com/runner-images:ubuntu-latest", "Docker image to use. Use \"-self-hosted\" to run directly on the host.")
|
execCmd.PersistentFlags().StringVarP(&execArg.image, "image", "i", "docker.gitea.com/runner-images:ubuntu-latest", "Docker image to use. Use \"-self-hosted\" to run directly on the host.")
|
||||||
|
execCmd.PersistentFlags().StringVarP(&execArg.toolCacheMode, "tool-cache-mode", "", config.ToolCacheModeNone, "What to mount at RUNNER_TOOL_CACHE: none, or shared to reuse one tool cache across runs")
|
||||||
execCmd.PersistentFlags().StringVarP(&execArg.network, "network", "", "", "Specify the network to which the container will connect")
|
execCmd.PersistentFlags().StringVarP(&execArg.network, "network", "", "", "Specify the network to which the container will connect")
|
||||||
execCmd.PersistentFlags().StringVarP(&execArg.githubInstance, "gitea-instance", "", "", "Gitea instance to use.")
|
execCmd.PersistentFlags().StringVarP(&execArg.githubInstance, "gitea-instance", "", "", "Gitea instance to use.")
|
||||||
|
|
||||||
|
|||||||
@@ -12,6 +12,8 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"gitea.com/gitea/runner/internal/pkg/config"
|
||||||
|
|
||||||
"gitea.dev/actionslib/pkg/model"
|
"gitea.dev/actionslib/pkg/model"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"go.yaml.in/yaml/v4"
|
"go.yaml.in/yaml/v4"
|
||||||
@@ -28,6 +30,19 @@ func TestExecuteArgsResolve(t *testing.T) {
|
|||||||
require.Equal(t, abs, args.resolve(abs))
|
require.Equal(t, abs, args.resolve(abs))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSharedToolCache(t *testing.T) {
|
||||||
|
shared, err := sharedToolCache(config.ToolCacheModeShared)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.True(t, shared)
|
||||||
|
|
||||||
|
shared, err = sharedToolCache(config.ToolCacheModeNone)
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.False(t, shared)
|
||||||
|
|
||||||
|
_, err = sharedToolCache("everyone")
|
||||||
|
require.ErrorContains(t, err, "tool-cache-mode")
|
||||||
|
}
|
||||||
|
|
||||||
func TestExecuteArgsPaths(t *testing.T) {
|
func TestExecuteArgsPaths(t *testing.T) {
|
||||||
workdir := t.TempDir()
|
workdir := t.TempDir()
|
||||||
args := &executeArgs{
|
args := &executeArgs{
|
||||||
|
|||||||
@@ -366,6 +366,7 @@ func doRegister(ctx context.Context, cfg *config.Config, inputs *registerInputs)
|
|||||||
cfg.Runner.Insecure,
|
cfg.Runner.Insecure,
|
||||||
"",
|
"",
|
||||||
"",
|
"",
|
||||||
|
config.RequestTimeout,
|
||||||
)
|
)
|
||||||
|
|
||||||
for {
|
for {
|
||||||
|
|||||||
@@ -68,6 +68,8 @@ type Poller struct {
|
|||||||
type workerState struct {
|
type workerState struct {
|
||||||
consecutiveEmpty int64
|
consecutiveEmpty int64
|
||||||
consecutiveErrors int64
|
consecutiveErrors int64
|
||||||
|
// fetchTimedOut suppresses repeats of the fetch timeout warning.
|
||||||
|
fetchTimedOut bool
|
||||||
// lastBackoff is the last interval reported to the PollBackoffSeconds gauge;
|
// lastBackoff is the last interval reported to the PollBackoffSeconds gauge;
|
||||||
// used to suppress redundant no-op Set calls when the backoff plateaus
|
// used to suppress redundant no-op Set calls when the backoff plateaus
|
||||||
// (e.g. at FetchIntervalMax).
|
// (e.g. at FetchIntervalMax).
|
||||||
@@ -346,16 +348,20 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
|
|||||||
TasksVersion: v,
|
TasksVersion: v,
|
||||||
}))
|
}))
|
||||||
|
|
||||||
// DeadlineExceeded is the designed idle path for a long-poll: the server
|
// Our own deadline proves nothing either way: today Gitea answers immediately,
|
||||||
// found no work within FetchTimeout. Treat it as an empty response and do
|
// so it means a slow server, and once it holds the request open it is an idle
|
||||||
// not record the duration — the timeout value would swamp the histogram.
|
// poll. Back off without claiming the server is healthy, warn once a streak so
|
||||||
|
// neither case floods, and keep it out of the latency histogram.
|
||||||
if errors.Is(err, context.DeadlineExceeded) {
|
if errors.Is(err, context.DeadlineExceeded) {
|
||||||
p.markHealthyPoll()
|
if !s.fetchTimedOut {
|
||||||
|
s.fetchTimedOut = true
|
||||||
|
log.Warnf("fetching a task timed out after %s, raise runner.fetch_timeout if this persists", p.cfg.Runner.FetchTimeout)
|
||||||
|
}
|
||||||
s.consecutiveEmpty++
|
s.consecutiveEmpty++
|
||||||
s.consecutiveErrors = 0 // timeout is a healthy idle response
|
|
||||||
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
|
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc()
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
|
s.fetchTimedOut = false
|
||||||
metrics.PollFetchDuration.Observe(time.Since(start).Seconds())
|
metrics.PollFetchDuration.Observe(time.Since(start).Seconds())
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -368,7 +374,13 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task,
|
|||||||
p.shutdownPolling()
|
p.shutdownPolling()
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
|
// Not a long poll, so a deadline can mean the server assigned a task
|
||||||
|
// this runner never received.
|
||||||
|
if errors.Is(err, context.DeadlineExceeded) {
|
||||||
|
log.WithError(err).Errorf("fetching a task timed out after %s", p.cfg.Runner.FetchTimeout)
|
||||||
|
} else {
|
||||||
log.WithError(err).Error("failed to fetch task")
|
log.WithError(err).Error("failed to fetch task")
|
||||||
|
}
|
||||||
p.lastPollFailed.Store(true)
|
p.lastPollFailed.Store(true)
|
||||||
s.consecutiveErrors++
|
s.consecutiveErrors++
|
||||||
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc()
|
metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc()
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import (
|
|||||||
|
|
||||||
connect_go "connectrpc.com/connect"
|
connect_go "connectrpc.com/connect"
|
||||||
runnerv1 "gitea.dev/actionslib/runner/v1"
|
runnerv1 "gitea.dev/actionslib/runner/v1"
|
||||||
|
"github.com/sirupsen/logrus/hooks/test"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/mock"
|
"github.com/stretchr/testify/mock"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
@@ -61,11 +62,7 @@ func TestPoller_WorkerStateCounters(t *testing.T) {
|
|||||||
// increments only the per-worker error counter, not the empty counter.
|
// increments only the per-worker error counter, not the empty counter.
|
||||||
func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
|
func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
|
||||||
client := mocks.NewClient(t)
|
client := mocks.NewClient(t)
|
||||||
client.On("FetchTask", mock.Anything, mock.Anything).Return(
|
client.On("FetchTask", mock.Anything, mock.Anything).Return(nil, errors.New("network unreachable"))
|
||||||
func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) {
|
|
||||||
return nil, errors.New("network unreachable")
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
cfg, err := config.LoadDefault("")
|
cfg, err := config.LoadDefault("")
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -78,6 +75,33 @@ func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) {
|
|||||||
assert.Equal(t, int64(0), s.consecutiveEmpty)
|
assert.Equal(t, int64(0), s.consecutiveEmpty)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A deadline is the idle path once the server holds the request open, and a
|
||||||
|
// symptom before then, so it must neither reset the error count nor assert that
|
||||||
|
// the server is reachable.
|
||||||
|
func TestPoller_FetchTimeoutIsNoSignal(t *testing.T) {
|
||||||
|
client := mocks.NewClient(t)
|
||||||
|
client.On("FetchTask", mock.Anything, mock.Anything).Return(nil, context.DeadlineExceeded)
|
||||||
|
|
||||||
|
cfg, err := config.LoadDefault("")
|
||||||
|
require.NoError(t, err)
|
||||||
|
p := &Poller{client: client, cfg: cfg}
|
||||||
|
p.lastPollFailed.Store(true)
|
||||||
|
|
||||||
|
hook := test.NewGlobal()
|
||||||
|
defer hook.Reset()
|
||||||
|
|
||||||
|
s := &workerState{consecutiveErrors: 2}
|
||||||
|
_, ok := p.fetchTask(context.Background(), s)
|
||||||
|
require.False(t, ok)
|
||||||
|
assert.Equal(t, int64(2), s.consecutiveErrors)
|
||||||
|
assert.Equal(t, int64(1), s.consecutiveEmpty)
|
||||||
|
assert.True(t, p.lastPollFailed.Load(), "a timeout must not clear a known failure")
|
||||||
|
|
||||||
|
// An idle runner against a server holding the request open must not flood.
|
||||||
|
_, _ = p.fetchTask(context.Background(), s)
|
||||||
|
assert.Len(t, hook.AllEntries(), 1)
|
||||||
|
}
|
||||||
|
|
||||||
// TestPoller_FetchUnauthenticatedStopsPolling verifies that an Unauthenticated
|
// TestPoller_FetchUnauthenticatedStopsPolling verifies that an Unauthenticated
|
||||||
// response marks the runner as unregistered and cancels the polling context so
|
// response marks the runner as unregistered and cancels the polling context so
|
||||||
// the daemon can exit instead of retrying forever.
|
// the daemon can exit instead of retrying forever.
|
||||||
|
|||||||
+46
-13
@@ -27,6 +27,7 @@ import (
|
|||||||
"gitea.com/gitea/runner/act/runner"
|
"gitea.com/gitea/runner/act/runner"
|
||||||
"gitea.com/gitea/runner/internal/pkg/client"
|
"gitea.com/gitea/runner/internal/pkg/client"
|
||||||
"gitea.com/gitea/runner/internal/pkg/config"
|
"gitea.com/gitea/runner/internal/pkg/config"
|
||||||
|
"gitea.com/gitea/runner/internal/pkg/disk"
|
||||||
"gitea.com/gitea/runner/internal/pkg/labels"
|
"gitea.com/gitea/runner/internal/pkg/labels"
|
||||||
"gitea.com/gitea/runner/internal/pkg/metrics"
|
"gitea.com/gitea/runner/internal/pkg/metrics"
|
||||||
"gitea.com/gitea/runner/internal/pkg/report"
|
"gitea.com/gitea/runner/internal/pkg/report"
|
||||||
@@ -89,17 +90,18 @@ func NewRunner(cfg *config.Config, reg *config.Registration, cli client.Client)
|
|||||||
var cacheHandler *artifactcache.Handler
|
var cacheHandler *artifactcache.Handler
|
||||||
if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled {
|
if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled {
|
||||||
if cfg.Cache.ExternalServer != "" {
|
if cfg.Cache.ExternalServer != "" {
|
||||||
|
warnIgnoredCachePolicy(cfg)
|
||||||
// The v1 client appends its path to this without a separator, so the slash is required.
|
// The v1 client appends its path to this without a separator, so the slash is required.
|
||||||
envs["ACTIONS_CACHE_URL"] = strings.TrimRight(cfg.Cache.ExternalServer, "/") + "/"
|
envs["ACTIONS_CACHE_URL"] = strings.TrimRight(cfg.Cache.ExternalServer, "/") + "/"
|
||||||
} else {
|
} else {
|
||||||
warnIgnoredCacheSecret(cfg)
|
warnIgnoredCacheSecret(cfg)
|
||||||
handler, err := artifactcache.StartHandler(
|
handler, err := artifactcache.StartHandler(artifactcache.Options{
|
||||||
cfg.Cache.Dir,
|
Dir: cfg.Cache.Dir,
|
||||||
cfg.Cache.Host,
|
OutboundIP: cfg.Cache.Host,
|
||||||
cfg.Cache.Port,
|
Port: cfg.Cache.Port,
|
||||||
"",
|
Policy: CachePolicy(cfg),
|
||||||
log.StandardLogger().WithField("module", "cache_request"),
|
Logger: log.StandardLogger().WithField("module", "cache_request"),
|
||||||
)
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Errorf("cannot init cache server, it will be disabled: %v", err)
|
log.Errorf("cannot init cache server, it will be disabled: %v", err)
|
||||||
// go on
|
// go on
|
||||||
@@ -157,8 +159,8 @@ func (r *Runner) OnIdle(ctx context.Context) {
|
|||||||
}
|
}
|
||||||
// Host mode: reclaim per-job scratch dirs left behind when HostEnvironment
|
// Host mode: reclaim per-job scratch dirs left behind when HostEnvironment
|
||||||
// cleanup timed out (e.g. a delete stalled by an AV/EDR filter driver). They
|
// cleanup timed out (e.g. a delete stalled by an AV/EDR filter driver). They
|
||||||
// sit under the host workdir parent alongside the shared tool_cache, which
|
// sit under the host workdir parent next to a shared tool_cache, which the name
|
||||||
// the name match leaves untouched. No-op when no host-mode job ever ran.
|
// match leaves untouched. No-op when no host-mode job ever ran.
|
||||||
if hostRoot := filepath.FromSlash(r.cfg.Host.WorkdirParent); hostRoot != "" {
|
if hostRoot := filepath.FromSlash(r.cfg.Host.WorkdirParent); hostRoot != "" {
|
||||||
r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir)
|
r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir)
|
||||||
}
|
}
|
||||||
@@ -217,7 +219,7 @@ func isTaskIDDir(name string) bool {
|
|||||||
// isHostScratchDir reports whether name is a per-job host-mode scratch dir:
|
// isHostScratchDir reports whether name is a per-job host-mode scratch dir:
|
||||||
// hex.EncodeToString of 8 random bytes, i.e. exactly 16 lowercase hex chars
|
// hex.EncodeToString of 8 random bytes, i.e. exactly 16 lowercase hex chars
|
||||||
// (see startHostEnvironment in act/runner/run_context.go). The narrow match
|
// (see startHostEnvironment in act/runner/run_context.go). The narrow match
|
||||||
// leaves the sibling shared "tool_cache" dir and any operator data untouched.
|
// leaves a sibling shared "tool_cache" dir and any operator data untouched.
|
||||||
func isHostScratchDir(name string) bool {
|
func isHostScratchDir(name string) bool {
|
||||||
if len(name) != 16 {
|
if len(name) != 16 {
|
||||||
return false
|
return false
|
||||||
@@ -520,6 +522,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report.
|
|||||||
JobCompletedHook: r.cfg.Runner.Hooks.JobCompleted,
|
JobCompletedHook: r.cfg.Runner.Hooks.JobCompleted,
|
||||||
Vars: task.Vars,
|
Vars: task.Vars,
|
||||||
ValidVolumes: r.cfg.Container.ValidVolumes,
|
ValidVolumes: r.cfg.Container.ValidVolumes,
|
||||||
|
SharedToolCache: r.cfg.Runner.ToolCacheMode == config.ToolCacheModeShared,
|
||||||
InsecureSkipTLS: r.cfg.Runner.Insecure,
|
InsecureSkipTLS: r.cfg.Runner.Insecure,
|
||||||
RunnerName: r.name,
|
RunnerName: r.name,
|
||||||
}
|
}
|
||||||
@@ -605,14 +608,15 @@ func (r *Runner) registerExternalCacheJob(token string, cred artifactcache.JobCr
|
|||||||
resultsURL := ""
|
resultsURL := ""
|
||||||
if body, err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, map[string]any{
|
if body, err := postInternalCache(base+"/_internal/register", r.cfg.Cache.ExternalSecret, map[string]any{
|
||||||
"token": token, "repo": cred.Repo, "results": cred.Results, "insecure_tls": cred.InsecureTLS,
|
"token": token, "repo": cred.Repo, "results": cred.Results, "insecure_tls": cred.InsecureTLS,
|
||||||
|
"public_url": base,
|
||||||
}); err != nil {
|
}); err != nil {
|
||||||
log.Warnf("cache external_server register failed (%s): %v", base, err)
|
log.Warnf("cache external_server register failed (%s): %v", base, err)
|
||||||
if reporter != nil {
|
if reporter != nil {
|
||||||
reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf(
|
reporter.Logf("::warning::%s", runner.EscapeCommandData(fmt.Sprintf(
|
||||||
"cache external_server register failed (%s): %v — cache requests from this job will be unauthenticated and likely return 401", base, err)))
|
"cache external_server register failed (%s): %v — cache requests from this job will be unauthenticated and likely return 401", base, err)))
|
||||||
}
|
}
|
||||||
} else {
|
} else if forwarded, _ := body["results_url"].(string); forwarded != "" {
|
||||||
resultsURL, _ = body["results_url"].(string) // absent from a server too old to forward
|
resultsURL = base // the answer only says it forwards, its own address need not be the job's
|
||||||
}
|
}
|
||||||
return func() {
|
return func() {
|
||||||
if _, err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret,
|
if _, err := postInternalCache(base+"/_internal/revoke", r.cfg.Cache.ExternalSecret,
|
||||||
@@ -692,7 +696,7 @@ func checkFreeDisk(cfg *config.Config) (bool, string) {
|
|||||||
root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/"))
|
root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/"))
|
||||||
}
|
}
|
||||||
root = nearestExistingPath(root)
|
root = nearestExistingPath(root)
|
||||||
available, err := freeDiskBytes(root)
|
available, err := disk.FreeBytes(root)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err)
|
return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err)
|
||||||
}
|
}
|
||||||
@@ -725,6 +729,35 @@ func (r *Runner) Declare(ctx context.Context, labels []string) (*connect.Respons
|
|||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// minFreeDisk keeps the cache from growing past the point where the runner stops taking
|
||||||
|
// work, but only when health checks are on, since the key is documented as opt-in.
|
||||||
|
func minFreeDisk(cfg *config.Config) int64 {
|
||||||
|
if !cfg.HealthCheck.Enabled {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
return cfg.HealthCheck.MinFreeDiskSpaceMB * 1024 * 1024
|
||||||
|
}
|
||||||
|
|
||||||
|
// CachePolicy maps the cache config onto the cache server's own type, in bytes not MiB.
|
||||||
|
func CachePolicy(cfg *config.Config) artifactcache.Policy {
|
||||||
|
return artifactcache.Policy{
|
||||||
|
Retention: cfg.Cache.Retention,
|
||||||
|
RepoSizeLimit: int64(cfg.Cache.RepoSizeLimit),
|
||||||
|
SizeLimit: int64(cfg.Cache.SizeLimit),
|
||||||
|
SweepInterval: cfg.Cache.SweepInterval,
|
||||||
|
MinFreeDisk: minFreeDisk(cfg),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// warnIgnoredCachePolicy flags eviction settings configured on a runner that points at an external cache server.
|
||||||
|
func warnIgnoredCachePolicy(cfg *config.Config) {
|
||||||
|
defaults := config.DefaultCache()
|
||||||
|
if cfg.Cache.Retention != defaults.Retention || cfg.Cache.RepoSizeLimit != defaults.RepoSizeLimit ||
|
||||||
|
cfg.Cache.SizeLimit != defaults.SizeLimit || cfg.Cache.SweepInterval != defaults.SweepInterval {
|
||||||
|
log.Warn("cache eviction settings are ignored when cache.external_server is set; configure them on that server instead")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// warnIgnoredCacheSecret flags an external cache server secret configured on a runner that uses the built-in cache server.
|
// warnIgnoredCacheSecret flags an external cache server secret configured on a runner that uses the built-in cache server.
|
||||||
func warnIgnoredCacheSecret(cfg *config.Config) {
|
func warnIgnoredCacheSecret(cfg *config.Config) {
|
||||||
if cfg.Cache.ExternalServer != "" {
|
if cfg.Cache.ExternalServer != "" {
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ func emptyCfg() *config.Config { return &config.Config{} }
|
|||||||
|
|
||||||
func TestRunner_registerCacheForTask(t *testing.T) {
|
func TestRunner_registerCacheForTask(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil)
|
handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -62,7 +62,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) {
|
|||||||
|
|
||||||
t.Run("empty token", func(t *testing.T) {
|
t.Run("empty token", func(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil)
|
handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -77,7 +77,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) {
|
|||||||
// /find, no auth on the signed archiveLocation download.
|
// /find, no auth on the signed archiveLocation download.
|
||||||
func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) {
|
func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||||
handler, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, "", nil)
|
handler, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.1"})
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer handler.Close()
|
defer handler.Close()
|
||||||
|
|
||||||
@@ -157,14 +157,15 @@ func decodeJSON(resp *http.Response, v any) error {
|
|||||||
|
|
||||||
// End-to-end against a remote cache-server: token unknown → 401, register →
|
// End-to-end against a remote cache-server: token unknown → 401, register →
|
||||||
// reserve/upload/commit/find/download all OK, revoke → 401 again. Registering also names the
|
// reserve/upload/commit/find/download all OK, revoke → 401 again. Registering also names the
|
||||||
// instance, and the server answering with its own address is what makes a shared cache server the
|
// instance and the address its jobs reach the server at, which makes a shared cache server the
|
||||||
// whole results service, as the built-in one is.
|
// whole results service, as the built-in one is.
|
||||||
func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
||||||
dir := filepath.Join(t.TempDir(), "remote-cache")
|
dir := filepath.Join(t.TempDir(), "remote-cache")
|
||||||
const secret = "shared-secret-for-tests"
|
const secret = "shared-secret-for-tests"
|
||||||
remote, err := artifactcache.StartHandler(dir, "127.0.0.1", 0, secret, nil)
|
remote, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.2", InternalSecret: secret}) // advertised, never dialled
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
defer remote.Close()
|
defer remote.Close()
|
||||||
|
external := strings.Replace(remote.ExternalURL(), "127.0.0.2", "127.0.0.1", 1)
|
||||||
gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
gitea := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||||
_, _ = io.WriteString(w, `{"ok":true}`)
|
_, _ = io.WriteString(w, `{"ok":true}`)
|
||||||
}))
|
}))
|
||||||
@@ -172,7 +173,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
|||||||
|
|
||||||
r := &Runner{
|
r := &Runner{
|
||||||
cfg: &config.Config{Cache: config.Cache{
|
cfg: &config.Config{Cache: config.Cache{
|
||||||
ExternalServer: remote.ExternalURL(),
|
ExternalServer: external,
|
||||||
ExternalSecret: secret,
|
ExternalSecret: secret,
|
||||||
}},
|
}},
|
||||||
envs: map[string]string{"ACTIONS_RESULTS_URL": gitea.URL},
|
envs: map[string]string{"ACTIONS_RESULTS_URL": gitea.URL},
|
||||||
@@ -180,7 +181,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
|||||||
|
|
||||||
token := "external-task-token"
|
token := "external-task-token"
|
||||||
repo := "owner/repoX"
|
repo := "owner/repoX"
|
||||||
base := remote.ExternalURL() + "/_apis/artifactcache"
|
base := external + "/_apis/artifactcache"
|
||||||
probe := func() int {
|
probe := func() int {
|
||||||
req, _ := http.NewRequest(http.MethodGet, base+"/cache?keys=k&version=v", nil)
|
req, _ := http.NewRequest(http.MethodGet, base+"/cache?keys=k&version=v", nil)
|
||||||
req.Header.Set("Authorization", "Bearer "+token)
|
req.Header.Set("Authorization", "Bearer "+token)
|
||||||
@@ -198,7 +199,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
|||||||
"token must be accepted after registerCacheForTask")
|
"token must be accepted after registerCacheForTask")
|
||||||
|
|
||||||
// The server took the results service over, so the artifact half reaches Gitea through it.
|
// The server took the results service over, so the artifact half reaches Gitea through it.
|
||||||
require.Equal(t, remote.ExternalURL(), resultsURL)
|
require.Equal(t, external, resultsURL)
|
||||||
artifact, err := http.NewRequestWithContext(t.Context(), http.MethodPost,
|
artifact, err := http.NewRequestWithContext(t.Context(), http.MethodPost,
|
||||||
resultsURL+"/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", nil)
|
resultsURL+"/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact", nil)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -248,7 +249,7 @@ func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
|||||||
ArchiveLocation string `json:"archiveLocation"`
|
ArchiveLocation string `json:"archiveLocation"`
|
||||||
}
|
}
|
||||||
require.NoError(t, decodeJSON(resp, &hit))
|
require.NoError(t, decodeJSON(resp, &hit))
|
||||||
require.NotEmpty(t, hit.ArchiveLocation)
|
require.True(t, strings.HasPrefix(hit.ArchiveLocation, external), hit.ArchiveLocation)
|
||||||
|
|
||||||
dl, err := http.Get(hit.ArchiveLocation)
|
dl, err := http.Get(hit.ArchiveLocation)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|||||||
@@ -5,8 +5,10 @@ package run
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"maps"
|
||||||
"os"
|
"os"
|
||||||
"runtime"
|
"runtime"
|
||||||
|
"slices"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
@@ -14,6 +16,7 @@ import (
|
|||||||
"gitea.com/gitea/runner/internal/pkg/ver"
|
"gitea.com/gitea/runner/internal/pkg/ver"
|
||||||
|
|
||||||
runnerv1 "gitea.dev/actionslib/runner/v1"
|
runnerv1 "gitea.dev/actionslib/runner/v1"
|
||||||
|
"google.golang.org/protobuf/types/known/structpb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// osReleasePath describes the host distribution on Linux; absent elsewhere, where the platform
|
// osReleasePath describes the host distribution on Linux; absent elsewhere, where the platform
|
||||||
@@ -46,12 +49,27 @@ func (r *Runner) setupLines(task *runnerv1.Task) []string {
|
|||||||
"Repository: "+fields["repository"].GetStringValue(),
|
"Repository: "+fields["repository"].GetStringValue(),
|
||||||
"Triggered by event: "+fields["event_name"].GetStringValue(),
|
"Triggered by event: "+fields["event_name"].GetStringValue(),
|
||||||
"::endgroup::",
|
"::endgroup::",
|
||||||
"::group::Operating System",
|
|
||||||
)
|
)
|
||||||
|
lines = append(lines, inputLines(fields)...)
|
||||||
|
lines = append(lines, "::group::Operating System")
|
||||||
lines = append(lines, osInfo()...)
|
lines = append(lines, osInfo()...)
|
||||||
return append(lines, "::endgroup::")
|
return append(lines, "::endgroup::")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// inputLines lists the inputs the run was triggered with, as workflow_dispatch and workflow_call
|
||||||
|
// carry them in the event payload. Empty when the event has none.
|
||||||
|
func inputLines(fields map[string]*structpb.Value) []string {
|
||||||
|
inputs := fields["event"].GetStructValue().GetFields()["inputs"].GetStructValue().GetFields()
|
||||||
|
if len(inputs) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
lines := []string{"::group::Inputs"}
|
||||||
|
for _, name := range slices.Sorted(maps.Keys(inputs)) {
|
||||||
|
lines = append(lines, fmt.Sprintf("%s: %v", name, inputs[name].AsInterface()))
|
||||||
|
}
|
||||||
|
return append(lines, "::endgroup::")
|
||||||
|
}
|
||||||
|
|
||||||
// osInfo describes the host the runner executes on.
|
// osInfo describes the host the runner executes on.
|
||||||
func osInfo() []string {
|
func osInfo() []string {
|
||||||
lines := make([]string, 0, 2)
|
lines := make([]string, 0, 2)
|
||||||
|
|||||||
@@ -56,6 +56,45 @@ func TestSetupLines(t *testing.T) {
|
|||||||
}, r.setupLines(&runnerv1.Task{Id: 268506, Context: taskCtx}))
|
}, r.setupLines(&runnerv1.Task{Id: 268506, Context: taskCtx}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSetupLinesInputs(t *testing.T) {
|
||||||
|
original := osReleasePath
|
||||||
|
osReleasePath = filepath.Join(t.TempDir(), "absent")
|
||||||
|
defer func() { osReleasePath = original }()
|
||||||
|
|
||||||
|
r := &Runner{name: "gitea-com-gitea-0003"}
|
||||||
|
taskCtx, err := structpb.NewStruct(map[string]any{
|
||||||
|
"job": "test",
|
||||||
|
"repository": "gitea/runner",
|
||||||
|
"event_name": "workflow_dispatch",
|
||||||
|
"event": map[string]any{
|
||||||
|
"inputs": map[string]any{
|
||||||
|
"with_default": "default",
|
||||||
|
"required": "required input",
|
||||||
|
"boolean": true,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
})
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
assert.Equal(t, []string{
|
||||||
|
"gitea-com-gitea-0003(version:" + ver.Version() + ")",
|
||||||
|
"::group::Runner Information",
|
||||||
|
"Task: 1",
|
||||||
|
"Job: test",
|
||||||
|
"Repository: gitea/runner",
|
||||||
|
"Triggered by event: workflow_dispatch",
|
||||||
|
"::endgroup::",
|
||||||
|
"::group::Inputs",
|
||||||
|
"boolean: true",
|
||||||
|
"required: required input",
|
||||||
|
"with_default: default",
|
||||||
|
"::endgroup::",
|
||||||
|
"::group::Operating System",
|
||||||
|
fmt.Sprintf("%s/%s", runtime.GOOS, runtime.GOARCH),
|
||||||
|
"::endgroup::",
|
||||||
|
}, r.setupLines(&runnerv1.Task{Id: 1, Context: taskCtx}))
|
||||||
|
}
|
||||||
|
|
||||||
func TestPrettyOSName(t *testing.T) {
|
func TestPrettyOSName(t *testing.T) {
|
||||||
tests := map[string]struct {
|
tests := map[string]struct {
|
||||||
osRelease string
|
osRelease string
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ import (
|
|||||||
"gitea.dev/actionslib/runner/v1/runnerv1connect"
|
"gitea.dev/actionslib/runner/v1/runnerv1connect"
|
||||||
)
|
)
|
||||||
|
|
||||||
func getHTTPClient(endpoint string, insecure bool) *http.Client {
|
func getHTTPClient(endpoint string, insecure bool, timeout time.Duration) *http.Client {
|
||||||
transport := &http.Transport{
|
transport := &http.Transport{
|
||||||
Proxy: http.ProxyFromEnvironment,
|
Proxy: http.ProxyFromEnvironment,
|
||||||
MaxIdleConns: 10,
|
MaxIdleConns: 10,
|
||||||
@@ -29,11 +29,13 @@ func getHTTPClient(endpoint string, insecure bool) *http.Client {
|
|||||||
InsecureSkipVerify: true,
|
InsecureSkipVerify: true,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return &http.Client{Transport: transport}
|
return &http.Client{Transport: transport, Timeout: timeout}
|
||||||
}
|
}
|
||||||
|
|
||||||
// New returns a new runner client.
|
// New returns a new runner client. timeout bounds every RPC: without it a
|
||||||
func New(endpoint string, insecure bool, uuid, token string, opts ...connect.ClientOption) *HTTPClient {
|
// stalled connection parks the reporter for the whole job context, so logs and
|
||||||
|
// heartbeats stop together and the task is reaped as a zombie.
|
||||||
|
func New(endpoint string, insecure bool, uuid, token string, timeout time.Duration, opts ...connect.ClientOption) *HTTPClient {
|
||||||
baseURL := strings.TrimRight(endpoint, "/") + "/api/actions"
|
baseURL := strings.TrimRight(endpoint, "/") + "/api/actions"
|
||||||
|
|
||||||
opts = append(opts, connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
|
opts = append(opts, connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc {
|
||||||
@@ -49,7 +51,7 @@ func New(endpoint string, insecure bool, uuid, token string, opts ...connect.Cli
|
|||||||
}
|
}
|
||||||
})))
|
})))
|
||||||
|
|
||||||
httpClient := getHTTPClient(endpoint, insecure)
|
httpClient := getHTTPClient(endpoint, insecure, timeout)
|
||||||
return &HTTPClient{
|
return &HTTPClient{
|
||||||
PingServiceClient: pingv1connect.NewPingServiceClient(
|
PingServiceClient: pingv1connect.NewPingServiceClient(
|
||||||
httpClient,
|
httpClient,
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"connectrpc.com/connect"
|
"connectrpc.com/connect"
|
||||||
pingv1 "gitea.dev/actionslib/ping/v1"
|
pingv1 "gitea.dev/actionslib/ping/v1"
|
||||||
@@ -17,7 +18,8 @@ import (
|
|||||||
func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) {
|
func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) {
|
||||||
t.Setenv("HTTP_PROXY", "http://proxy.example.com:8080")
|
t.Setenv("HTTP_PROXY", "http://proxy.example.com:8080")
|
||||||
|
|
||||||
client := getHTTPClient("http://gitea.example.com", false)
|
client := getHTTPClient("http://gitea.example.com", false, time.Minute)
|
||||||
|
require.Equal(t, time.Minute, client.Timeout)
|
||||||
transport, ok := client.Transport.(*http.Transport)
|
transport, ok := client.Transport.(*http.Transport)
|
||||||
require.True(t, ok)
|
require.True(t, ok)
|
||||||
|
|
||||||
@@ -32,7 +34,7 @@ func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) {
|
|||||||
|
|
||||||
func TestGetHTTPClientInsecureTLS(t *testing.T) {
|
func TestGetHTTPClientInsecureTLS(t *testing.T) {
|
||||||
// insecure only takes effect for https endpoints
|
// insecure only takes effect for https endpoints
|
||||||
httpsInsecure := getHTTPClient("https://gitea.example.com", true)
|
httpsInsecure := getHTTPClient("https://gitea.example.com", true, time.Minute)
|
||||||
transport, ok := httpsInsecure.Transport.(*http.Transport)
|
transport, ok := httpsInsecure.Transport.(*http.Transport)
|
||||||
require.True(t, ok)
|
require.True(t, ok)
|
||||||
require.NotNil(t, transport.TLSClientConfig)
|
require.NotNil(t, transport.TLSClientConfig)
|
||||||
@@ -47,7 +49,7 @@ func TestGetHTTPClientInsecureTLS(t *testing.T) {
|
|||||||
{"http insecure ignored", "http://gitea.example.com", true},
|
{"http insecure ignored", "http://gitea.example.com", true},
|
||||||
} {
|
} {
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
c := getHTTPClient(tc.endpoint, tc.insecure)
|
c := getHTTPClient(tc.endpoint, tc.insecure, time.Minute)
|
||||||
tr, ok := c.Transport.(*http.Transport)
|
tr, ok := c.Transport.(*http.Transport)
|
||||||
require.True(t, ok)
|
require.True(t, ok)
|
||||||
require.Nil(t, tr.TLSClientConfig)
|
require.Nil(t, tr.TLSClientConfig)
|
||||||
@@ -66,7 +68,7 @@ func TestNewSetsBaseURLAndHeaders(t *testing.T) {
|
|||||||
defer server.Close()
|
defer server.Close()
|
||||||
|
|
||||||
// trailing slash must be trimmed before "/api/actions" is appended
|
// trailing slash must be trimmed before "/api/actions" is appended
|
||||||
c := New(server.URL+"/", false, "the-uuid", "the-token")
|
c := New(server.URL+"/", false, "the-uuid", "the-token", time.Minute)
|
||||||
// Address returns the endpoint as supplied (untrimmed)
|
// Address returns the endpoint as supplied (untrimmed)
|
||||||
require.Equal(t, server.URL+"/", c.Address())
|
require.Equal(t, server.URL+"/", c.Address())
|
||||||
require.False(t, c.Insecure())
|
require.False(t, c.Insecure())
|
||||||
@@ -87,7 +89,7 @@ func TestNewOmitsEmptyHeaders(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer server.Close()
|
defer server.Close()
|
||||||
|
|
||||||
c := New(server.URL, false, "", "")
|
c := New(server.URL, false, "", "", time.Minute)
|
||||||
_, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"}))
|
_, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"}))
|
||||||
|
|
||||||
require.Empty(t, gotHeaders.Get(UUIDHeader))
|
require.Empty(t, gotHeaders.Get(UUIDHeader))
|
||||||
|
|||||||
@@ -34,6 +34,7 @@ runner:
|
|||||||
# Whether skip verifying the TLS certificate of the Gitea instance.
|
# Whether skip verifying the TLS certificate of the Gitea instance.
|
||||||
#insecure: false
|
#insecure: false
|
||||||
# The timeout for fetching the job from the Gitea instance.
|
# The timeout for fetching the job from the Gitea instance.
|
||||||
|
# Values above the 60s RPC timeout are capped to it.
|
||||||
#fetch_timeout: 5s
|
#fetch_timeout: 5s
|
||||||
# The interval for fetching the job from the Gitea instance.
|
# The interval for fetching the job from the Gitea instance.
|
||||||
#fetch_interval: 2s
|
#fetch_interval: 2s
|
||||||
@@ -93,6 +94,12 @@ runner:
|
|||||||
# terminal; tools like `docker build` emit redrawing progress frames into the captured log
|
# terminal; tools like `docker build` emit redrawing progress frames into the captured log
|
||||||
# when a TTY is present.
|
# when a TTY is present.
|
||||||
#allocate_pty: false
|
#allocate_pty: false
|
||||||
|
# What to mount at RUNNER_TOOL_CACHE (/opt/hostedtoolcache), where setup actions install tools:
|
||||||
|
# none: nothing. A docker job sees what its image ships there, a host job an empty dir, and
|
||||||
|
# either way what it installs is gone when the job ends.
|
||||||
|
# shared: one tool cache every job reuses. Two jobs writing the same tool version at once
|
||||||
|
# corrupt it, so use it only with capacity 1.
|
||||||
|
#tool_cache_mode: none
|
||||||
# Optional executable on the host, run once after each task's built-in cleanup
|
# Optional executable on the host, run once after each task's built-in cleanup
|
||||||
# (post-steps, container teardown, bind-workdir removal). Additive only.
|
# (post-steps, container teardown, bind-workdir removal). Additive only.
|
||||||
#
|
#
|
||||||
@@ -137,6 +144,7 @@ cache:
|
|||||||
#port: 0
|
#port: 0
|
||||||
# URL of a shared `gitea-runner cache-server` to use instead of starting a local one.
|
# URL of a shared `gitea-runner cache-server` to use instead of starting a local one.
|
||||||
# Set on every runner that should share a cache pool. A trailing slash is optional.
|
# Set on every runner that should share a cache pool. A trailing slash is optional.
|
||||||
|
# Jobs reach the server at this URL too, so set it to the reverse proxy when one fronts the server.
|
||||||
# Example: "http://cache-host:8088/"
|
# Example: "http://cache-host:8088/"
|
||||||
# Requires external_secret (below) to match the value on the cache-server.
|
# Requires external_secret (below) to match the value on the cache-server.
|
||||||
#external_server: ""
|
#external_server: ""
|
||||||
@@ -158,6 +166,20 @@ cache:
|
|||||||
# the action's own bundle, keeping the untouched copy beside it. The same edit lets the stock
|
# the action's own bundle, keeping the untouched copy beside it. The same edit lets the stock
|
||||||
# upload-artifact and download-artifact work here. A bundle that does not match is left alone.
|
# upload-artifact and download-artifact work here. A bundle that does not match is left alone.
|
||||||
#v2: true
|
#v2: true
|
||||||
|
# How the cache server discards entries, ignored when external_server is set since that
|
||||||
|
# server applies its own. Leave a setting out for its default; 0s or 0 turns the three
|
||||||
|
# limits off. Sizes accept 10GB, 512mb, 1TiB or a plain byte count, binary either way.
|
||||||
|
# Whatever these allow, the cache still sheds entries to keep free space on its volume
|
||||||
|
# above health_check.min_free_disk_space_mb when health checks are enabled.
|
||||||
|
# Remove entries nothing has read or written within this window. Only last access counts.
|
||||||
|
#retention: 168h
|
||||||
|
# Cap one repository, removing its least recently accessed entries until it fits. An entry
|
||||||
|
# larger than the limit is dropped rather than emptying the repository to make room.
|
||||||
|
#repo_size_limit: 10GB
|
||||||
|
# Cap the whole cache the same way. Off by default, since the free space floor bounds it.
|
||||||
|
#size_limit: 0
|
||||||
|
# Minimum time between two eviction sweeps. This one has no "off".
|
||||||
|
#sweep_interval: 1h
|
||||||
|
|
||||||
container:
|
container:
|
||||||
# Specifies the network to which the container will connect.
|
# Specifies the network to which the container will connect.
|
||||||
@@ -180,7 +202,7 @@ container:
|
|||||||
#privileged: false
|
#privileged: false
|
||||||
# Any other options to be used when the container is started, for example:
|
# Any other options to be used when the container is started, for example:
|
||||||
# options: --add-host=my.gitea.url:host-gateway
|
# options: --add-host=my.gitea.url:host-gateway
|
||||||
# A volume declared here replaces the one the runner mounts on the same container path, so the
|
# A volume declared here replaces the one the runner would mount on the same container path, so the
|
||||||
# tool cache can be kept on the host. Its source must also be allowed by valid_volumes below:
|
# tool cache can be kept on the host. Its source must also be allowed by valid_volumes below:
|
||||||
# options: --volume /host/toolcache:/opt/hostedtoolcache
|
# options: --volume /host/toolcache:/opt/hostedtoolcache
|
||||||
#options:
|
#options:
|
||||||
|
|||||||
@@ -10,14 +10,19 @@ import (
|
|||||||
"maps"
|
"maps"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/docker/go-units"
|
||||||
"github.com/joho/godotenv"
|
"github.com/joho/godotenv"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
"go.yaml.in/yaml/v4"
|
"go.yaml.in/yaml/v4"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// RequestTimeout bounds every RPC to Gitea, and with it runner.fetch_timeout.
|
||||||
|
const RequestTimeout = 60 * time.Second
|
||||||
|
|
||||||
// DefaultPostTaskScriptTimeout is the fallback cap on how long the post-task
|
// DefaultPostTaskScriptTimeout is the fallback cap on how long the post-task
|
||||||
// script may run when post_task_script is set without an explicit timeout. It is
|
// script may run when post_task_script is set without an explicit timeout. It is
|
||||||
// applied both at config load (for a configured script) and at the point of use
|
// applied both at config load (for a configured script) and at the point of use
|
||||||
@@ -59,6 +64,7 @@ type Runner struct {
|
|||||||
ActionShallowClone *bool `yaml:"action_shallow_clone"` // ActionShallowClone fetches only the requested ref of an action repository at depth 1 instead of cloning every branch's full history. It is a pointer to distinguish between false and not set; if not set, it defaults to true.
|
ActionShallowClone *bool `yaml:"action_shallow_clone"` // ActionShallowClone fetches only the requested ref of an action repository at depth 1 instead of cloning every branch's full history. It is a pointer to distinguish between false and not set; if not set, it defaults to true.
|
||||||
SetActEnv *bool `yaml:"set_act_env"` // SetActEnv controls whether the ACT=true environment variable is injected into jobs. It is a pointer to distinguish between false and not set; if not set, it defaults to true. Set it to false so workflows gated on `if: ${{ !env.ACT }}` behave like on GitHub.
|
SetActEnv *bool `yaml:"set_act_env"` // SetActEnv controls whether the ACT=true environment variable is injected into jobs. It is a pointer to distinguish between false and not set; if not set, it defaults to true. Set it to false so workflows gated on `if: ${{ !env.ACT }}` behave like on GitHub.
|
||||||
AllocatePTY bool `yaml:"allocate_pty"` // AllocatePTY allocates a pseudo-TTY for each step's process. Default is false, matching GitHub's actions/runner. Enable only for jobs that need an interactive terminal; tools like docker build emit redrawing progress frames into the captured log when a TTY is present. Applies to both host and docker backends.
|
AllocatePTY bool `yaml:"allocate_pty"` // AllocatePTY allocates a pseudo-TTY for each step's process. Default is false, matching GitHub's actions/runner. Enable only for jobs that need an interactive terminal; tools like docker build emit redrawing progress frames into the captured log when a TTY is present. Applies to both host and docker backends.
|
||||||
|
ToolCacheMode string `yaml:"tool_cache_mode"` // ToolCacheMode is what the runner mounts at RUNNER_TOOL_CACHE on both backends: ToolCacheModeNone or ToolCacheModeShared.
|
||||||
PostTaskScript string `yaml:"post_task_script"` // PostTaskScript is the path to an executable script run on the host after each task's cleanup completes. Empty disables the hook. On Windows use .exe/.bat/.cmd; PowerShell (.ps1) is not supported yet as the configured path.
|
PostTaskScript string `yaml:"post_task_script"` // PostTaskScript is the path to an executable script run on the host after each task's cleanup completes. Empty disables the hook. On Windows use .exe/.bat/.cmd; PowerShell (.ps1) is not supported yet as the configured path.
|
||||||
PostTaskScriptTimeout time.Duration `yaml:"post_task_script_timeout"` // PostTaskScriptTimeout caps how long the post-task script may run. Default is 5m when post_task_script is set.
|
PostTaskScriptTimeout time.Duration `yaml:"post_task_script_timeout"` // PostTaskScriptTimeout caps how long the post-task script may run. Default is 5m when post_task_script is set.
|
||||||
Hooks RunnerHooks `yaml:"hooks"` // Hooks are scripts run inside the job environment around the job's steps.
|
Hooks RunnerHooks `yaml:"hooks"` // Hooks are scripts run inside the job environment around the job's steps.
|
||||||
@@ -81,6 +87,47 @@ type Cache struct {
|
|||||||
ExternalSecretFile string `yaml:"external_secret_file"` // ExternalSecretFile is the path to a file holding the ExternalSecret value, so the secret can be mounted instead of stored in the config file. LoadDefault reads it into ExternalSecret; setting both is an error.
|
ExternalSecretFile string `yaml:"external_secret_file"` // ExternalSecretFile is the path to a file holding the ExternalSecret value, so the secret can be mounted instead of stored in the config file. LoadDefault reads it into ExternalSecret; setting both is an error.
|
||||||
OfflineMode bool `yaml:"offline_mode"` // OfflineMode reuses a cached action without fetching from the remote; a moved tag or branch stays at the cached commit until the cache entry is removed.
|
OfflineMode bool `yaml:"offline_mode"` // OfflineMode reuses a cached action without fetching from the remote; a moved tag or branch stays at the cached commit until the cache entry is removed.
|
||||||
V2 *bool `yaml:"v2"` // V2 serves the actions cache service v2 API to jobs, used by actions/cache@v4.2 and later, and edits the action bundles that would otherwise refuse it. Unset means enabled.
|
V2 *bool `yaml:"v2"` // V2 serves the actions cache service v2 API to jobs, used by actions/cache@v4.2 and later, and edits the action bundles that would otherwise refuse it. Unset means enabled.
|
||||||
|
|
||||||
|
// Eviction settings, ignored when ExternalServer is set since that server applies its own.
|
||||||
|
Retention time.Duration `yaml:"retention"` // Retention removes entries nothing has read or written within this window. Default 168h, 0 keeps them regardless of age.
|
||||||
|
RepoSizeLimit Size `yaml:"repo_size_limit"` // RepoSizeLimit caps one repository, evicting least recently accessed first. Default 10GB, 0 is no limit.
|
||||||
|
SizeLimit Size `yaml:"size_limit"` // SizeLimit caps the whole cache the same way. No limit by default.
|
||||||
|
SweepInterval time.Duration `yaml:"sweep_interval"` // SweepInterval is the minimum time between two eviction sweeps. Default 1h; a cadence has no "off".
|
||||||
|
}
|
||||||
|
|
||||||
|
// DefaultCache returns the cache eviction defaults, seeded before the file is read so a
|
||||||
|
// written 0 can mean off. SizeLimit stays zero: the free space floor bounds the whole cache.
|
||||||
|
func DefaultCache() Cache {
|
||||||
|
return Cache{
|
||||||
|
Retention: 7 * 24 * time.Hour,
|
||||||
|
RepoSizeLimit: 10 * 1024 * 1024 * 1024,
|
||||||
|
SweepInterval: time.Hour,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Size is a byte count written the way people say it: 10GB, 512mb, 1TiB, or a plain number
|
||||||
|
// of bytes. Units are binary and case-insensitive, so GB and GiB both mean 1024³.
|
||||||
|
type Size int64
|
||||||
|
|
||||||
|
func (s *Size) UnmarshalYAML(value *yaml.Node) error {
|
||||||
|
if value.Kind != yaml.ScalarNode {
|
||||||
|
return fmt.Errorf("line %d: size must be a scalar such as 10GB", value.Line)
|
||||||
|
}
|
||||||
|
size, err := parseSize(value.Value)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("line %d: %w", value.Line, err)
|
||||||
|
}
|
||||||
|
*s = size
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// parseSize reads a Size such as 10GB, 512mb, 1TiB or a plain byte count.
|
||||||
|
func parseSize(value string) (Size, error) {
|
||||||
|
bytes, err := units.RAMInBytes(strings.TrimSpace(value))
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("%q is not a size such as 10GB, 512MB or a plain byte count", value)
|
||||||
|
}
|
||||||
|
return Size(bytes), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Container represents the configuration for the container.
|
// Container represents the configuration for the container.
|
||||||
@@ -101,6 +148,14 @@ type Container struct {
|
|||||||
ServiceReadyTimeout time.Duration `yaml:"service_ready_timeout"` // ServiceReadyTimeout bounds how long a job waits for a service container that declares a healthcheck to report healthy. Negative disables waiting.
|
ServiceReadyTimeout time.Duration `yaml:"service_ready_timeout"` // ServiceReadyTimeout bounds how long a job waits for a service container that declares a healthcheck to report healthy. Negative disables waiting.
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Values of Runner.ToolCacheMode: the runner mounts no tool cache, or one that every job reuses.
|
||||||
|
const (
|
||||||
|
ToolCacheModeNone = "none"
|
||||||
|
ToolCacheModeShared = "shared"
|
||||||
|
)
|
||||||
|
|
||||||
|
var ToolCacheModes = []string{ToolCacheModeNone, ToolCacheModeShared}
|
||||||
|
|
||||||
type ContainerNetworkCreateOptions struct {
|
type ContainerNetworkCreateOptions struct {
|
||||||
EnableIPv4 *bool `yaml:"enable_ipv4"` // Enable or disable IPv4 for the network (true for docker by default)
|
EnableIPv4 *bool `yaml:"enable_ipv4"` // Enable or disable IPv4 for the network (true for docker by default)
|
||||||
EnableIPv6 *bool `yaml:"enable_ipv6"` // Enable or disable IPv6 for the network (false for docker by default)
|
EnableIPv6 *bool `yaml:"enable_ipv6"` // Enable or disable IPv6 for the network (false for docker by default)
|
||||||
@@ -142,7 +197,7 @@ type Config struct {
|
|||||||
// LoadDefault returns the default configuration.
|
// LoadDefault returns the default configuration.
|
||||||
// If file is not empty, it will be used to load the configuration.
|
// If file is not empty, it will be used to load the configuration.
|
||||||
func LoadDefault(file string) (*Config, error) {
|
func LoadDefault(file string) (*Config, error) {
|
||||||
cfg := &Config{}
|
cfg := &Config{Cache: DefaultCache()}
|
||||||
definedRunnerKeys := map[string]bool{}
|
definedRunnerKeys := map[string]bool{}
|
||||||
if file != "" {
|
if file != "" {
|
||||||
content, err := os.ReadFile(file)
|
content, err := os.ReadFile(file)
|
||||||
@@ -215,6 +270,12 @@ func LoadDefault(file string) (*Config, error) {
|
|||||||
if cfg.Container.WorkdirParent == "" {
|
if cfg.Container.WorkdirParent == "" {
|
||||||
cfg.Container.WorkdirParent = "workspace"
|
cfg.Container.WorkdirParent = "workspace"
|
||||||
}
|
}
|
||||||
|
if cfg.Runner.ToolCacheMode == "" {
|
||||||
|
cfg.Runner.ToolCacheMode = ToolCacheModeNone
|
||||||
|
}
|
||||||
|
if !slices.Contains(ToolCacheModes, cfg.Runner.ToolCacheMode) {
|
||||||
|
return nil, fmt.Errorf("invalid runner.tool_cache_mode %q: must be one of %q", cfg.Runner.ToolCacheMode, ToolCacheModes)
|
||||||
|
}
|
||||||
if cfg.Host.WorkdirParent == "" {
|
if cfg.Host.WorkdirParent == "" {
|
||||||
home, err := os.UserHomeDir()
|
home, err := os.UserHomeDir()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -272,6 +333,14 @@ func LoadDefault(file string) (*Config, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Validate and fix invalid config combinations to prevent confusing behavior.
|
// Validate and fix invalid config combinations to prevent confusing behavior.
|
||||||
|
if cfg.Runner.ToolCacheMode == ToolCacheModeShared && cfg.Runner.Capacity > 1 {
|
||||||
|
log.Warnf("runner.tool_cache_mode %q with capacity %d: two jobs writing the same tool version at once corrupt it",
|
||||||
|
ToolCacheModeShared, cfg.Runner.Capacity)
|
||||||
|
}
|
||||||
|
if cfg.Runner.FetchTimeout > RequestTimeout {
|
||||||
|
log.Warnf("fetch_timeout (%v) exceeds the RPC timeout (%v), capping it", cfg.Runner.FetchTimeout, RequestTimeout)
|
||||||
|
cfg.Runner.FetchTimeout = RequestTimeout
|
||||||
|
}
|
||||||
if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval {
|
if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval {
|
||||||
log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval",
|
log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval",
|
||||||
cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval)
|
cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval)
|
||||||
|
|||||||
@@ -42,6 +42,23 @@ cache:
|
|||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestLoadDefault_ToolCacheMode(t *testing.T) {
|
||||||
|
cfg, err := LoadDefault("")
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, ToolCacheModeNone, cfg.Runner.ToolCacheMode)
|
||||||
|
|
||||||
|
path := filepath.Join(t.TempDir(), "config.yaml")
|
||||||
|
require.NoError(t, os.WriteFile(path, []byte("runner:\n tool_cache_mode: shared\n"), 0o600))
|
||||||
|
cfg, err = LoadDefault(path)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, ToolCacheModeShared, cfg.Runner.ToolCacheMode)
|
||||||
|
|
||||||
|
require.NoError(t, os.WriteFile(path, []byte("runner:\n tool_cache_mode: everyone\n"), 0o600))
|
||||||
|
_, err = LoadDefault(path)
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.Contains(t, err.Error(), "tool_cache_mode")
|
||||||
|
}
|
||||||
|
|
||||||
func TestLoadDefault_DefaultsWorkdirCleanupAge(t *testing.T) {
|
func TestLoadDefault_DefaultsWorkdirCleanupAge(t *testing.T) {
|
||||||
cfg, err := LoadDefault("")
|
cfg, err := LoadDefault("")
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -168,6 +185,37 @@ runner:
|
|||||||
assert.Equal(t, 5*time.Minute, cfg.Runner.PostTaskScriptTimeout)
|
assert.Equal(t, 5*time.Minute, cfg.Runner.PostTaskScriptTimeout)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestLoadDefault_LoadsCacheEviction(t *testing.T) {
|
||||||
|
write := func(t *testing.T, body string) string {
|
||||||
|
t.Helper()
|
||||||
|
path := filepath.Join(t.TempDir(), "config.yaml")
|
||||||
|
require.NoError(t, os.WriteFile(path, []byte(body), 0o600))
|
||||||
|
return path
|
||||||
|
}
|
||||||
|
|
||||||
|
t.Run("sizes accept any spelling of the unit", func(t *testing.T) {
|
||||||
|
cfg, err := LoadDefault(write(t, "cache:\n retention: 336h\n repo_size_limit: 50gb\n size_limit: 1TiB\n sweep_interval: 15m\n"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, 336*time.Hour, cfg.Cache.Retention)
|
||||||
|
assert.Equal(t, Size(50*1024*1024*1024), cfg.Cache.RepoSizeLimit)
|
||||||
|
assert.Equal(t, Size(1024*1024*1024*1024), cfg.Cache.SizeLimit)
|
||||||
|
assert.Equal(t, 15*time.Minute, cfg.Cache.SweepInterval)
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("zero turns a limit off where an absent key keeps its default", func(t *testing.T) {
|
||||||
|
cfg, err := LoadDefault(write(t, "cache:\n repo_size_limit: 0\n"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Zero(t, cfg.Cache.RepoSizeLimit)
|
||||||
|
assert.Equal(t, DefaultCache().Retention, cfg.Cache.Retention, "an absent key still defaults")
|
||||||
|
})
|
||||||
|
|
||||||
|
t.Run("a bad size names the offending value", func(t *testing.T) {
|
||||||
|
_, err := LoadDefault(write(t, "cache:\n repo_size_limit: banana\n"))
|
||||||
|
require.Error(t, err)
|
||||||
|
assert.Contains(t, err.Error(), "banana")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
func TestLoadDefault_LoadsJobHooks(t *testing.T) {
|
func TestLoadDefault_LoadsJobHooks(t *testing.T) {
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
path := filepath.Join(dir, "config.yaml")
|
path := filepath.Join(dir, "config.yaml")
|
||||||
@@ -368,3 +416,15 @@ func TestLoadDefault_ShippedConfigsChangeNothing(t *testing.T) {
|
|||||||
}
|
}
|
||||||
assert.Empty(t, hook.AllEntries())
|
assert.Empty(t, hook.AllEntries())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestLoadDefault_ClampsFetchTimeout(t *testing.T) {
|
||||||
|
path := filepath.Join(t.TempDir(), "config.yaml")
|
||||||
|
require.NoError(t, os.WriteFile(path, []byte(`
|
||||||
|
runner:
|
||||||
|
fetch_timeout: 120s
|
||||||
|
`), 0o600))
|
||||||
|
|
||||||
|
cfg, err := LoadDefault(path)
|
||||||
|
require.NoError(t, err)
|
||||||
|
assert.Equal(t, RequestTimeout, cfg.Runner.FetchTimeout)
|
||||||
|
}
|
||||||
|
|||||||
@@ -26,7 +26,10 @@ const (
|
|||||||
kindSection
|
kindSection
|
||||||
)
|
)
|
||||||
|
|
||||||
var durationType = reflect.TypeFor[time.Duration]()
|
var (
|
||||||
|
durationType = reflect.TypeFor[time.Duration]()
|
||||||
|
sizeType = reflect.TypeFor[Size]()
|
||||||
|
)
|
||||||
|
|
||||||
// GetValue renders a flat list or mapping one entry per line, and anything nested as YAML.
|
// GetValue renders a flat list or mapping one entry per line, and anything nested as YAML.
|
||||||
func GetValue(file, path string) (string, error) {
|
func GetValue(file, path string) (string, error) {
|
||||||
@@ -537,6 +540,13 @@ func scalarNode(typ reflect.Type, value string) (*yaml.Node, error) {
|
|||||||
return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: duration.String()}, nil
|
return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: duration.String()}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if typ == sizeType {
|
||||||
|
if _, err := parseSize(value); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!str", Value: value}, nil
|
||||||
|
}
|
||||||
|
|
||||||
switch typ.Kind() {
|
switch typ.Kind() {
|
||||||
case reflect.Bool:
|
case reflect.Bool:
|
||||||
parsed, err := strconv.ParseBool(value)
|
parsed, err := strconv.ParseBool(value)
|
||||||
|
|||||||
@@ -0,0 +1,7 @@
|
|||||||
|
// Copyright 2026 The Gitea Authors. All rights reserved.
|
||||||
|
// SPDX-License-Identifier: MIT
|
||||||
|
|
||||||
|
// Package disk reports free space on the volume holding a path. Platforms without an
|
||||||
|
// implementation return an error, so callers treat the check as unavailable rather than
|
||||||
|
// as a full disk.
|
||||||
|
package disk
|
||||||
@@ -3,10 +3,11 @@
|
|||||||
|
|
||||||
//go:build !aix && !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows
|
//go:build !aix && !darwin && !dragonfly && !freebsd && !linux && !netbsd && !openbsd && !solaris && !windows
|
||||||
|
|
||||||
package run
|
package disk
|
||||||
|
|
||||||
import "fmt"
|
import "fmt"
|
||||||
|
|
||||||
func freeDiskBytes(path string) (uint64, error) {
|
// FreeBytes reports the space available to an unprivileged user on the volume holding path.
|
||||||
|
func FreeBytes(path string) (uint64, error) {
|
||||||
return 0, fmt.Errorf("free disk space checks are not supported for %s", path)
|
return 0, fmt.Errorf("free disk space checks are not supported for %s", path)
|
||||||
}
|
}
|
||||||
@@ -3,11 +3,12 @@
|
|||||||
|
|
||||||
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
|
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
|
||||||
|
|
||||||
package run
|
package disk
|
||||||
|
|
||||||
import "golang.org/x/sys/unix"
|
import "golang.org/x/sys/unix"
|
||||||
|
|
||||||
func freeDiskBytes(path string) (uint64, error) {
|
// FreeBytes reports the space available to an unprivileged user on the volume holding path.
|
||||||
|
func FreeBytes(path string) (uint64, error) {
|
||||||
var stat unix.Statfs_t
|
var stat unix.Statfs_t
|
||||||
if err := unix.Statfs(path, &stat); err != nil {
|
if err := unix.Statfs(path, &stat); err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
@@ -3,11 +3,12 @@
|
|||||||
|
|
||||||
//go:build windows
|
//go:build windows
|
||||||
|
|
||||||
package run
|
package disk
|
||||||
|
|
||||||
import "golang.org/x/sys/windows"
|
import "golang.org/x/sys/windows"
|
||||||
|
|
||||||
func freeDiskBytes(path string) (uint64, error) {
|
// FreeBytes reports the space available to an unprivileged user on the volume holding path.
|
||||||
|
func FreeBytes(path string) (uint64, error) {
|
||||||
pathPtr, err := windows.UTF16PtrFromString(path)
|
pathPtr, err := windows.UTF16PtrFromString(path)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, err
|
return 0, err
|
||||||
@@ -26,6 +26,9 @@ import (
|
|||||||
"google.golang.org/protobuf/types/known/timestamppb"
|
"google.golang.org/protobuf/types/known/timestamppb"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// errOutputsNotSent travels the same return path as transport failures but is not one.
|
||||||
|
var errOutputsNotSent = errors.New("there are still outputs that have not been sent")
|
||||||
|
|
||||||
// Size limits for the outputs reported to the server.
|
// Size limits for the outputs reported to the server.
|
||||||
const (
|
const (
|
||||||
maxOutputKeyLen = 255
|
maxOutputKeyLen = 255
|
||||||
@@ -58,6 +61,11 @@ type Reporter struct {
|
|||||||
|
|
||||||
state *runnerv1.TaskState
|
state *runnerv1.TaskState
|
||||||
stateChanged bool
|
stateChanged bool
|
||||||
|
// reportFailing keeps an outage to one log line at each end.
|
||||||
|
reportFailing map[string]bool
|
||||||
|
// serverResult is what the server decided, e.g. the zombie reaper failing
|
||||||
|
// the task. Guarded by stateMu.
|
||||||
|
serverResult runnerv1.Result
|
||||||
stateMu sync.RWMutex
|
stateMu sync.RWMutex
|
||||||
outputsMu sync.Mutex
|
outputsMu sync.Mutex
|
||||||
outputs map[string]jobOutput
|
outputs map[string]jobOutput
|
||||||
@@ -77,6 +85,8 @@ type Reporter struct {
|
|||||||
// closeTimeout bounds each RPC attempt in the final flush, on a context
|
// closeTimeout bounds each RPC attempt in the final flush, on a context
|
||||||
// detached from r.ctx so a server cancel can't abort the acknowledgement.
|
// detached from r.ctx so a server cancel can't abort the acknowledgement.
|
||||||
closeTimeout time.Duration
|
closeTimeout time.Duration
|
||||||
|
// daemonWait bounds how long Close waits for the daemon loop to acknowledge.
|
||||||
|
daemonWait time.Duration
|
||||||
|
|
||||||
// Event notification channels (non-blocking, buffered 1)
|
// Event notification channels (non-blocking, buffered 1)
|
||||||
logNotify chan struct{} // signal: new log rows arrived
|
logNotify chan struct{} // signal: new log rows arrived
|
||||||
@@ -119,10 +129,13 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C
|
|||||||
state: &runnerv1.TaskState{
|
state: &runnerv1.TaskState{
|
||||||
Id: task.Id,
|
Id: task.Id,
|
||||||
},
|
},
|
||||||
|
reportFailing: map[string]bool{},
|
||||||
daemon: make(chan struct{}),
|
daemon: make(chan struct{}),
|
||||||
heartbeatStop: make(chan struct{}),
|
heartbeatStop: make(chan struct{}),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
rv.daemonWait = 6 * rv.effectiveCloseTimeout()
|
||||||
|
|
||||||
if task.Secrets["ACTIONS_STEP_DEBUG"] == "true" {
|
if task.Secrets["ACTIONS_STEP_DEBUG"] == "true" {
|
||||||
rv.debugOutputEnabled = true
|
rv.debugOutputEnabled = true
|
||||||
}
|
}
|
||||||
@@ -293,6 +306,21 @@ func (r *Reporter) Fire(entry *log.Entry) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Only the daemon loop calls this, so reportFailing needs no lock.
|
||||||
|
func (r *Reporter) noteReport(method string, err error) {
|
||||||
|
if errors.Is(err, errOutputsNotSent) {
|
||||||
|
err = nil // the RPC itself succeeded
|
||||||
|
}
|
||||||
|
switch {
|
||||||
|
case err != nil && !r.reportFailing[method]:
|
||||||
|
r.reportFailing[method] = true
|
||||||
|
log.Warnf("%s error: %v, retrying until reconnected", method, err)
|
||||||
|
case err == nil && r.reportFailing[method]:
|
||||||
|
delete(r.reportFailing, method)
|
||||||
|
log.Infof("%s reconnected", method)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (r *Reporter) RunDaemon() {
|
func (r *Reporter) RunDaemon() {
|
||||||
go r.runDaemonLoop()
|
go r.runDaemonLoop()
|
||||||
}
|
}
|
||||||
@@ -319,6 +347,8 @@ func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r *Reporter) runDaemonLoop() {
|
func (r *Reporter) runDaemonLoop() {
|
||||||
|
defer close(r.daemon)
|
||||||
|
|
||||||
logTicker := time.NewTicker(r.logReportInterval)
|
logTicker := time.NewTicker(r.logReportInterval)
|
||||||
stateTicker := time.NewTicker(r.stateReportInterval)
|
stateTicker := time.NewTicker(r.stateReportInterval)
|
||||||
|
|
||||||
@@ -337,11 +367,11 @@ func (r *Reporter) runDaemonLoop() {
|
|||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-logTicker.C:
|
case <-logTicker.C:
|
||||||
_ = r.ReportLog(false)
|
r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
|
||||||
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
|
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
|
||||||
|
|
||||||
case <-stateTicker.C:
|
case <-stateTicker.C:
|
||||||
_ = r.ReportState(false)
|
r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
|
||||||
|
|
||||||
case <-r.logNotify:
|
case <-r.logNotify:
|
||||||
r.stateMu.RLock()
|
r.stateMu.RLock()
|
||||||
@@ -349,7 +379,7 @@ func (r *Reporter) runDaemonLoop() {
|
|||||||
r.stateMu.RUnlock()
|
r.stateMu.RUnlock()
|
||||||
|
|
||||||
if n >= r.logBatchSize {
|
if n >= r.logBatchSize {
|
||||||
_ = r.ReportLog(false)
|
r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
|
||||||
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
|
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
|
||||||
} else if !maxLatencyActive && n > 0 {
|
} else if !maxLatencyActive && n > 0 {
|
||||||
maxLatencyTimer.Reset(r.logReportMaxLatency)
|
maxLatencyTimer.Reset(r.logReportMaxLatency)
|
||||||
@@ -358,25 +388,23 @@ func (r *Reporter) runDaemonLoop() {
|
|||||||
|
|
||||||
case <-r.stateNotify:
|
case <-r.stateNotify:
|
||||||
// Step transition or job result — flush both immediately for frontend UX.
|
// Step transition or job result — flush both immediately for frontend UX.
|
||||||
_ = r.ReportLog(false)
|
r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
|
||||||
_ = r.ReportState(false)
|
r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
|
||||||
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
|
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
|
||||||
|
|
||||||
case <-maxLatencyTimer.C:
|
case <-maxLatencyTimer.C:
|
||||||
maxLatencyActive = false
|
maxLatencyActive = false
|
||||||
_ = r.ReportLog(false)
|
r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
|
||||||
|
|
||||||
case <-r.ctx.Done():
|
case <-r.ctx.Done():
|
||||||
// Stop heartbeating on cancel so Gitea sees the runner as offline
|
// Stop heartbeating on cancel so Gitea sees the runner as offline
|
||||||
// during cleanup and won't assign an overlapping task. Close() still
|
// during cleanup and won't assign an overlapping task. Close() still
|
||||||
// delivers the final flush on a detached context (flushFinal).
|
// delivers the final flush on a detached context (flushFinal).
|
||||||
close(r.daemon)
|
|
||||||
return
|
return
|
||||||
|
|
||||||
case <-r.heartbeatStop:
|
case <-r.heartbeatStop:
|
||||||
// Stop heartbeating during post-task script execution. Close() still
|
// Stop heartbeating during post-task script execution. Close() still
|
||||||
// delivers the final flush on a detached context (flushFinal).
|
// delivers the final flush on a detached context (flushFinal).
|
||||||
close(r.daemon)
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -384,7 +412,6 @@ func (r *Reporter) runDaemonLoop() {
|
|||||||
closed := r.closed
|
closed := r.closed
|
||||||
r.stateMu.RUnlock()
|
r.stateMu.RUnlock()
|
||||||
if closed {
|
if closed {
|
||||||
close(r.daemon)
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -435,28 +462,23 @@ func (r *Reporter) Close(lastWords string) error {
|
|||||||
r.stateMu.Lock()
|
r.stateMu.Lock()
|
||||||
r.closed = true
|
r.closed = true
|
||||||
if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED {
|
if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED {
|
||||||
// When r.ctx has been cancelled (server returned RESULT_CANCELLED via
|
// No result of its own, so say why it stopped.
|
||||||
// rpcCtx/ReportState, see line 590) the job is being torn down on the
|
result, words := runnerv1.Result_RESULT_FAILURE, "Early termination"
|
||||||
// cancellation path: surface that explicitly instead of attributing it
|
switch {
|
||||||
// to a generic failure.
|
case r.serverResult != runnerv1.Result_RESULT_UNSPECIFIED:
|
||||||
cancelled := errors.Is(r.ctx.Err(), context.Canceled)
|
result, words = r.serverResult, "Ended by the server"
|
||||||
if lastWords == "" {
|
case errors.Is(r.ctx.Err(), context.Canceled):
|
||||||
if cancelled {
|
result, words = runnerv1.Result_RESULT_CANCELLED, "Cancelled"
|
||||||
lastWords = "Cancelled"
|
|
||||||
} else {
|
|
||||||
lastWords = "Early termination"
|
|
||||||
}
|
}
|
||||||
|
if lastWords == "" {
|
||||||
|
lastWords = words
|
||||||
}
|
}
|
||||||
for _, v := range r.state.Steps {
|
for _, v := range r.state.Steps {
|
||||||
if v.Result == runnerv1.Result_RESULT_UNSPECIFIED {
|
if v.Result == runnerv1.Result_RESULT_UNSPECIFIED {
|
||||||
v.Result = runnerv1.Result_RESULT_CANCELLED
|
v.Result = runnerv1.Result_RESULT_CANCELLED
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if cancelled {
|
r.state.Result = result
|
||||||
r.state.Result = runnerv1.Result_RESULT_CANCELLED
|
|
||||||
} else {
|
|
||||||
r.state.Result = runnerv1.Result_RESULT_FAILURE
|
|
||||||
}
|
|
||||||
r.logRows = append(r.logRows, &runnerv1.LogRow{
|
r.logRows = append(r.logRows, &runnerv1.LogRow{
|
||||||
Time: timestamppb.Now(),
|
Time: timestamppb.Now(),
|
||||||
Content: lastWords,
|
Content: lastWords,
|
||||||
@@ -476,9 +498,8 @@ func (r *Reporter) Close(lastWords string) error {
|
|||||||
// Wait for Acknowledge
|
// Wait for Acknowledge
|
||||||
select {
|
select {
|
||||||
case <-r.daemon:
|
case <-r.daemon:
|
||||||
case <-time.After(60 * time.Second):
|
case <-time.After(r.daemonWait):
|
||||||
close(r.daemon)
|
log.Errorf("No Response from RunDaemon for %s, continue best effort", r.daemonWait)
|
||||||
log.Error("No Response from RunDaemon for 60s, continue best effort")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Gitea's UpdateLog short-circuits on len(Rows)==0 before honoring NoMore,
|
// Gitea's UpdateLog short-circuits on len(Rows)==0 before honoring NoMore,
|
||||||
@@ -569,8 +590,10 @@ func (r *Reporter) ReportLog(noMore bool) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
r.stateMu.Lock()
|
r.stateMu.Lock()
|
||||||
r.logRows = r.logRows[ack-r.logOffset:]
|
|
||||||
submitted := r.logOffset + len(rows)
|
submitted := r.logOffset + len(rows)
|
||||||
|
// A server can ack beyond what it was sent; clamp to stay within the buffer.
|
||||||
|
ack = min(ack, submitted)
|
||||||
|
r.logRows = r.logRows[ack-r.logOffset:]
|
||||||
r.logOffset = ack
|
r.logOffset = ack
|
||||||
remaining := len(r.logRows)
|
remaining := len(r.logRows)
|
||||||
r.stateMu.Unlock()
|
r.stateMu.Unlock()
|
||||||
@@ -655,11 +678,16 @@ func (r *Reporter) ReportState(reportResult bool) error {
|
|||||||
}
|
}
|
||||||
r.outputsMu.Unlock()
|
r.outputsMu.Unlock()
|
||||||
|
|
||||||
if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED {
|
// A terminal result means the server is done with this task; keep running and
|
||||||
|
// the job holds a capacity slot until runner.timeout.
|
||||||
|
if state := resp.Msg.State; state != nil && state.Result != runnerv1.Result_RESULT_UNSPECIFIED {
|
||||||
|
r.stateMu.Lock()
|
||||||
|
r.serverResult = state.Result
|
||||||
|
r.stateMu.Unlock()
|
||||||
r.cancel()
|
r.cancel()
|
||||||
}
|
}
|
||||||
if len(noSent) > 0 {
|
if len(noSent) > 0 {
|
||||||
return fmt.Errorf("there are still outputs that have not been sent: %v", noSent)
|
return fmt.Errorf("%w: %v", errOutputsNotSent, noSent)
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ import (
|
|||||||
"net/url"
|
"net/url"
|
||||||
"slices"
|
"slices"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -19,10 +20,12 @@ import (
|
|||||||
"gitea.com/gitea/runner/act/runner"
|
"gitea.com/gitea/runner/act/runner"
|
||||||
"gitea.com/gitea/runner/internal/pkg/client/mocks"
|
"gitea.com/gitea/runner/internal/pkg/client/mocks"
|
||||||
"gitea.com/gitea/runner/internal/pkg/config"
|
"gitea.com/gitea/runner/internal/pkg/config"
|
||||||
|
"gitea.com/gitea/runner/internal/pkg/metrics"
|
||||||
|
|
||||||
connect_go "connectrpc.com/connect"
|
connect_go "connectrpc.com/connect"
|
||||||
runnerv1 "gitea.dev/actionslib/runner/v1"
|
runnerv1 "gitea.dev/actionslib/runner/v1"
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
|
logrustest "github.com/sirupsen/logrus/hooks/test"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/mock"
|
"github.com/stretchr/testify/mock"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
@@ -1169,3 +1172,132 @@ func TestReporter_masksEncodedSecrets(t *testing.T) {
|
|||||||
assert.NotContains(t, row.Content, base64.StdEncoding.EncodeToString([]byte(secret)))
|
assert.NotContains(t, row.Content, base64.StdEncoding.EncodeToString([]byte(secret)))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A server that acknowledges more rows than were sent must not take the runner
|
||||||
|
// process down with it.
|
||||||
|
func TestReporter_AckIndexBeyondBuffer(t *testing.T) {
|
||||||
|
client := mocks.NewClient(t)
|
||||||
|
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
|
||||||
|
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: 1000}), nil
|
||||||
|
})
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
taskCtx, err := structpb.NewStruct(map[string]any{})
|
||||||
|
require.NoError(t, err)
|
||||||
|
cfg, _ := config.LoadDefault("")
|
||||||
|
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
|
||||||
|
reporter.ResetSteps(1)
|
||||||
|
|
||||||
|
require.NoError(t, reporter.Fire(&log.Entry{
|
||||||
|
Message: "hello",
|
||||||
|
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||||
|
Level: log.InfoLevel,
|
||||||
|
}))
|
||||||
|
|
||||||
|
require.NoError(t, reporter.ReportLog(false))
|
||||||
|
}
|
||||||
|
|
||||||
|
// Close has to survive giving up on a daemon still parked in an RPC.
|
||||||
|
func TestReporter_CloseWithStuckDaemon(t *testing.T) {
|
||||||
|
release := make(chan struct{})
|
||||||
|
entered := make(chan struct{})
|
||||||
|
var once sync.Once
|
||||||
|
|
||||||
|
client := mocks.NewClient(t)
|
||||||
|
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
|
||||||
|
once.Do(func() {
|
||||||
|
close(entered)
|
||||||
|
<-release
|
||||||
|
})
|
||||||
|
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
|
||||||
|
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
|
||||||
|
}), nil
|
||||||
|
})
|
||||||
|
client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
|
||||||
|
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil
|
||||||
|
}).Maybe()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
taskCtx, err := structpb.NewStruct(map[string]any{})
|
||||||
|
require.NoError(t, err)
|
||||||
|
cfg, _ := config.LoadDefault("")
|
||||||
|
cfg.Runner.LogReportInterval = 10 * time.Millisecond
|
||||||
|
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
|
||||||
|
reporter.daemonWait = time.Millisecond
|
||||||
|
reporter.RunDaemon()
|
||||||
|
reporter.ResetSteps(1)
|
||||||
|
|
||||||
|
require.NoError(t, reporter.Fire(&log.Entry{
|
||||||
|
Message: "hello",
|
||||||
|
Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true},
|
||||||
|
Level: log.InfoLevel,
|
||||||
|
}))
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-entered:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("daemon never reached UpdateLog")
|
||||||
|
}
|
||||||
|
go func() {
|
||||||
|
time.Sleep(10 * reporter.daemonWait)
|
||||||
|
close(release)
|
||||||
|
}()
|
||||||
|
|
||||||
|
require.NoError(t, reporter.Close(""))
|
||||||
|
}
|
||||||
|
|
||||||
|
// The zombie reaper marks a task failed, and cancelling the context is how the
|
||||||
|
// runner stops, so the two must not be conflated into "cancelled".
|
||||||
|
func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) {
|
||||||
|
var lastState *runnerv1.TaskState
|
||||||
|
client := mocks.NewClient(t)
|
||||||
|
client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) {
|
||||||
|
lastState = req.Msg.State
|
||||||
|
return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{
|
||||||
|
State: &runnerv1.TaskState{Result: runnerv1.Result_RESULT_FAILURE},
|
||||||
|
}), nil
|
||||||
|
})
|
||||||
|
client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) {
|
||||||
|
return connect_go.NewResponse(&runnerv1.UpdateLogResponse{
|
||||||
|
AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)),
|
||||||
|
}), nil
|
||||||
|
}).Maybe()
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
defer cancel()
|
||||||
|
taskCtx, err := structpb.NewStruct(map[string]any{})
|
||||||
|
require.NoError(t, err)
|
||||||
|
cfg, _ := config.LoadDefault("")
|
||||||
|
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
|
||||||
|
close(reporter.daemon) // no daemon loop to acknowledge
|
||||||
|
|
||||||
|
require.NoError(t, reporter.ReportState(false))
|
||||||
|
require.ErrorIs(t, ctx.Err(), context.Canceled)
|
||||||
|
require.NoError(t, reporter.Close(""))
|
||||||
|
|
||||||
|
require.NotNil(t, lastState)
|
||||||
|
assert.Equal(t, runnerv1.Result_RESULT_FAILURE, lastState.Result)
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestReporter_NoteReport(t *testing.T) {
|
||||||
|
hook := logrustest.NewGlobal()
|
||||||
|
defer hook.Reset()
|
||||||
|
|
||||||
|
reporter := &Reporter{reportFailing: map[string]bool{}}
|
||||||
|
|
||||||
|
// A pending output is not a transport failure and must not read as one.
|
||||||
|
reporter.noteReport(metrics.LabelMethodUpdateTask, fmt.Errorf("wrapped: %w", errOutputsNotSent))
|
||||||
|
assert.Empty(t, hook.AllEntries())
|
||||||
|
|
||||||
|
// An outage logs once at each end, not every report interval for hours.
|
||||||
|
reporter.noteReport(metrics.LabelMethodUpdateTask, errors.New("connection refused"))
|
||||||
|
reporter.noteReport(metrics.LabelMethodUpdateTask, errors.New("connection refused"))
|
||||||
|
require.Len(t, hook.AllEntries(), 1)
|
||||||
|
assert.Contains(t, hook.LastEntry().Message, "connection refused")
|
||||||
|
|
||||||
|
reporter.noteReport(metrics.LabelMethodUpdateTask, nil)
|
||||||
|
require.Len(t, hook.AllEntries(), 2)
|
||||||
|
assert.Contains(t, hook.LastEntry().Message, "reconnected")
|
||||||
|
}
|
||||||
|
|||||||
@@ -6,13 +6,9 @@
|
|||||||
# Cloudflare R2 bucket, using curl's built-in AWS SigV4 signer (R2 is
|
# Cloudflare R2 bucket, using curl's built-in AWS SigV4 signer (R2 is
|
||||||
# S3-API compatible).
|
# S3-API compatible).
|
||||||
#
|
#
|
||||||
# This is the R2 half of the release process's parallel S3+R2 upload
|
# It is invoked once per release artifact via a goreleaser
|
||||||
# period: goreleaser's `blobs:` pipe still uploads every release
|
# `publishers:` entry, and is the only object storage upload of the
|
||||||
# artifact to AWS S3, and this script is invoked once per artifact
|
# release process.
|
||||||
# (via a goreleaser `publishers:` entry) to mirror the same artifact
|
|
||||||
# into R2. Once the migration away from S3 is complete, the `blobs:`
|
|
||||||
# block and the AWS_* secrets can be dropped without touching this
|
|
||||||
# script.
|
|
||||||
#
|
#
|
||||||
# Usage:
|
# Usage:
|
||||||
# upload-r2.sh <local-file> <remote-key>
|
# upload-r2.sh <local-file> <remote-key>
|
||||||
@@ -24,7 +20,7 @@
|
|||||||
# preflight step in CI: goreleaser custom publishers run as the very
|
# preflight step in CI: goreleaser custom publishers run as the very
|
||||||
# last step of the publish pipeline, so without a preflight check a
|
# last step of the publish pipeline, so without a preflight check a
|
||||||
# missing R2_* secret would only be discovered after the Gitea release
|
# missing R2_* secret would only be discovered after the Gitea release
|
||||||
# has already been created and every artifact already uploaded to S3.
|
# has already been created.
|
||||||
#
|
#
|
||||||
# Required environment variables:
|
# Required environment variables:
|
||||||
# R2_ENDPOINT Base URL of the R2 endpoint, e.g.
|
# R2_ENDPOINT Base URL of the R2 endpoint, e.g.
|
||||||
|
|||||||
Reference in New Issue
Block a user