mirror of
https://gitea.com/gitea/act_runner.git
synced 2026-08-19 19:46:28 +00:00
feat: add size-based cache eviction (#1170)
The cache server retired entries 30 days after creation regardless of use, so a job that ran often enough to keep its cache warm still lost it on a fixed schedule. Nothing bounded the disk either. Retention now counts from last access alone, and a repository over its limit sheds least recently accessed entries until it fits, enforced on commit as well as on the periodic sweep. ```yaml cache: retention: 168h # remove entries not accessed for seven days repo_size_limit: 10GB # cap each repository size_limit: 0 # cap the whole cache, off by default sweep_interval: 1h # minimum time between sweeps ``` Sizes accept `10GB`, `512mb`, `1TiB` or a plain byte count, binary either way. Leave a key out for its default; `0` turns a limit off, and `0s` does the same for `retention`. Whatever these allow, the cache also sheds entries to keep free space above `health_check.min_free_disk_space_mb` when health checks are enabled, so it cannot grow past the point where the runner stops accepting work. Supporting fixes: serving an entry stamps its access time, so a find cannot hand a job a download URL for an entry the next eviction is about to remove; an entry larger than the limit is dropped on its own account rather than emptying its repository to make room; and a blob that cannot be unlinked keeps its row, so the next sweep retries instead of orphaning bytes no limit can account for. Closes https://gitea.com/gitea/runner/issues/1168 --------- Co-authored-by: silverwind <me@silverwind.io> Reviewed-on: https://gitea.com/gitea/runner/pulls/1170 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
This commit is contained in:
@@ -273,6 +273,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.
|
||||
|
||||
**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**
|
||||
|
||||
`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:
|
||||
|
||||
+271
-93
@@ -5,6 +5,7 @@
|
||||
package artifactcache
|
||||
|
||||
import (
|
||||
"cmp"
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/rand"
|
||||
@@ -20,6 +21,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -27,6 +29,7 @@ import (
|
||||
"time"
|
||||
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
"gitea.com/gitea/runner/internal/pkg/disk"
|
||||
|
||||
"github.com/julienschmidt/httprouter"
|
||||
"github.com/sirupsen/logrus"
|
||||
@@ -103,19 +106,38 @@ type Handler struct {
|
||||
|
||||
credMu sync.RWMutex
|
||||
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.
|
||||
//
|
||||
// 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.
|
||||
func StartHandler(dir, outboundIP string, port uint16, internalSecret string, logger logrus.FieldLogger) (*Handler, error) {
|
||||
func StartHandler(opts Options) (*Handler, error) {
|
||||
dir, logger := opts.Dir, opts.Logger
|
||||
h := &Handler{
|
||||
creds: make(map[string]*credEntry),
|
||||
internalSecret: internalSecret,
|
||||
internalSecret: opts.InternalSecret,
|
||||
policy: opts.Policy.withDefaults(),
|
||||
freeDisk: disk.FreeBytes,
|
||||
}
|
||||
|
||||
if logger == nil {
|
||||
@@ -145,8 +167,8 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo
|
||||
}
|
||||
h.storage = storage
|
||||
|
||||
if outboundIP != "" {
|
||||
h.outboundIP = outboundIP
|
||||
if opts.OutboundIP != "" {
|
||||
h.outboundIP = opts.OutboundIP
|
||||
} else if ip := common.GetOutboundIP(); ip == nil {
|
||||
return nil, errors.New("unable to determine outbound IP address")
|
||||
} else {
|
||||
@@ -185,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
|
||||
// routable from inside the container network. Authentication is enforced
|
||||
// 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 {
|
||||
return nil, err
|
||||
}
|
||||
@@ -390,6 +412,9 @@ func (h *Handler) lookupCache(db *bolthold.Store, repo string, keys []string, ve
|
||||
_ = db.Delete(cache.ID, cache)
|
||||
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
|
||||
}
|
||||
|
||||
@@ -527,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.
|
||||
cache.Size = written
|
||||
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()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
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
|
||||
@@ -821,12 +855,43 @@ func (h *Handler) touchCache(id uint64, requireIncomplete bool) error {
|
||||
}
|
||||
|
||||
const (
|
||||
keepUsed = 30 * 24 * time.Hour
|
||||
keepUnused = 7 * 24 * time.Hour
|
||||
keepTemp = 5 * time.Minute
|
||||
keepOld = 5 * time.Minute
|
||||
miB = 1024 * 1024
|
||||
|
||||
defaultSweepInterval = time.Hour
|
||||
|
||||
// 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() {
|
||||
if h.gcing.Load() {
|
||||
return
|
||||
@@ -836,7 +901,7 @@ func (h *Handler) gcCache() {
|
||||
}
|
||||
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())
|
||||
return
|
||||
}
|
||||
@@ -849,95 +914,208 @@ func (h *Handler) gcCache() {
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
// Remove the caches which are not completed for a while, they are most likely to be broken.
|
||||
var caches []*Cache
|
||||
if err := db.Find(&caches, bolthold.
|
||||
Where("UsedAt").Lt(time.Now().Add(-keepTemp).Unix()).
|
||||
And("Complete").Eq(false),
|
||||
); err != nil {
|
||||
h.logger.Warnf("find caches: %v", err)
|
||||
} else {
|
||||
for _, cache := range caches {
|
||||
h.storage.Remove(cache.ID)
|
||||
if err := db.Delete(cache.ID, cache); err != nil {
|
||||
h.logger.Warnf("delete cache: %v", err)
|
||||
continue
|
||||
}
|
||||
h.logger.Infof("deleted cache: %+v", cache)
|
||||
}
|
||||
h.evictIncomplete(db)
|
||||
h.evictExpired(db)
|
||||
h.evictSuperseded(db)
|
||||
h.evictOversized(db)
|
||||
h.evictForFreeSpace(db)
|
||||
}
|
||||
|
||||
// evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never
|
||||
// accounted for.
|
||||
func (h *Handler) evictForFreeSpace(db *bolthold.Store) {
|
||||
free, err := h.freeDisk(h.dir)
|
||||
if err != nil {
|
||||
h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full
|
||||
return
|
||||
}
|
||||
if free >= uint64(h.policy.MinFreeDisk) {
|
||||
return
|
||||
}
|
||||
|
||||
// Remove the old caches which have not been used recently.
|
||||
caches = caches[:0]
|
||||
if err := db.Find(&caches, bolthold.
|
||||
Where("UsedAt").Lt(time.Now().Add(-keepUnused).Unix()),
|
||||
); err != nil {
|
||||
h.logger.Warnf("find caches: %v", err)
|
||||
} else {
|
||||
for _, cache := range caches {
|
||||
h.storage.Remove(cache.ID)
|
||||
if err := db.Delete(cache.ID, cache); err != nil {
|
||||
h.logger.Warnf("delete cache: %v", err)
|
||||
continue
|
||||
}
|
||||
h.logger.Infof("deleted cache: %+v", cache)
|
||||
}
|
||||
caches := h.completedByUse(db)
|
||||
total, shortfall := totalSize(caches), h.policy.MinFreeDisk-int64(free)
|
||||
if total <= shortfall {
|
||||
// Say so, or shedding everything and still being short reads as the backstop working.
|
||||
h.logger.Warnf("cache volume is %d MiB short of the free space floor with only %d MiB of cache on it; something else is filling it", shortfall/miB, total/miB)
|
||||
}
|
||||
h.evictTo(db, caches, total-shortfall, "the cache volume")
|
||||
}
|
||||
|
||||
// Remove the old caches which are too old.
|
||||
caches = caches[:0]
|
||||
if err := db.Find(&caches, bolthold.
|
||||
Where("CreatedAt").Lt(time.Now().Add(-keepUsed).Unix()),
|
||||
); err != nil {
|
||||
h.logger.Warnf("find caches: %v", err)
|
||||
} else {
|
||||
for _, cache := range caches {
|
||||
h.storage.Remove(cache.ID)
|
||||
if err := db.Delete(cache.ID, cache); err != nil {
|
||||
h.logger.Warnf("delete cache: %v", err)
|
||||
continue
|
||||
}
|
||||
h.logger.Infof("deleted cache: %+v", cache)
|
||||
}
|
||||
// evictIncomplete removes uploads that stopped part way, which are most likely broken.
|
||||
func (h *Handler) evictIncomplete(db *bolthold.Store) {
|
||||
h.sweep(db, bolthold.
|
||||
Where("UsedAt").Lt(time.Now().Add(-uploadStallTimeout).Unix()).
|
||||
And("Complete").Eq(false).
|
||||
Index("UsedAt"))
|
||||
}
|
||||
|
||||
func (h *Handler) evictExpired(db *bolthold.Store) {
|
||||
if h.policy.Retention <= 0 {
|
||||
return
|
||||
}
|
||||
// Never below inUseGrace, or a short retention would outrun a signed URL already issued.
|
||||
window := max(h.policy.Retention, inUseGrace)
|
||||
h.sweep(db, bolthold.Where("UsedAt").Lt(time.Now().Add(-window).Unix()).Index("UsedAt"))
|
||||
}
|
||||
|
||||
// Remove the old caches with the same key and version within the same
|
||||
// repository, keep the latest one. Aggregation must include Repo so two
|
||||
// repos that happen to share a (key, version) do not evict each other —
|
||||
// otherwise per-repo scoping holds for reads but one repo can age
|
||||
// another out after keepOld.
|
||||
// Also keep the olds which have been used recently for a while in case of the cache is still in use.
|
||||
if results, err := db.FindAggregate(
|
||||
&Cache{},
|
||||
bolthold.Where("Complete").Eq(true),
|
||||
"Repo", "Key", "Version",
|
||||
); err != nil {
|
||||
// evictSuperseded removes entries a newer one with the same key and version replaced. The
|
||||
// aggregation includes Repo so two repos sharing a (key, version) do not evict each other.
|
||||
func (h *Handler) evictSuperseded(db *bolthold.Store) {
|
||||
results, err := db.FindAggregate(&Cache{}, bolthold.Where("Complete").Eq(true).Index("Complete"), "Repo", "Key", "Version")
|
||||
if err != nil {
|
||||
h.logger.Warnf("find aggregate caches: %v", err)
|
||||
} else {
|
||||
for _, result := range results {
|
||||
if result.Count() <= 1 {
|
||||
return
|
||||
}
|
||||
var caches []*Cache
|
||||
for _, result := range results {
|
||||
if result.Count() <= 1 {
|
||||
continue
|
||||
}
|
||||
result.Sort("CreatedAt")
|
||||
caches = caches[:0]
|
||||
result.Reduction(&caches)
|
||||
for _, cache := range caches[:len(caches)-1] {
|
||||
if inUse(cache) {
|
||||
continue
|
||||
}
|
||||
result.Sort("CreatedAt")
|
||||
caches = caches[:0]
|
||||
result.Reduction(&caches)
|
||||
for _, cache := range caches[:len(caches)-1] {
|
||||
if time.Since(time.Unix(cache.UsedAt, 0)) < keepOld {
|
||||
// Keep it since it has been used recently, even if it's old.
|
||||
// Or it could break downloading in process.
|
||||
continue
|
||||
}
|
||||
h.storage.Remove(cache.ID)
|
||||
if err := db.Delete(cache.ID, cache); err != nil {
|
||||
h.logger.Warnf("delete cache: %v", err)
|
||||
continue
|
||||
}
|
||||
h.logger.Infof("deleted cache: %+v", cache)
|
||||
}
|
||||
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 {
|
||||
h.logger.Warnf("delete cache: %v", err)
|
||||
return false
|
||||
}
|
||||
h.logger.Infof("deleted cache: %+v", cache)
|
||||
return true
|
||||
}
|
||||
|
||||
func (h *Handler) responseJSON(w http.ResponseWriter, r *http.Request, code int, v ...any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
var data []byte
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
@@ -41,6 +42,9 @@ func (b *bearerTransport) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
|
||||
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;
|
||||
// tests use it to reach the get handler directly without going through a
|
||||
// find/cache-hit round trip.
|
||||
@@ -50,7 +54,7 @@ func signArtifactURL(h *Handler, id int64) string {
|
||||
|
||||
func TestHandler(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
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]))
|
||||
}
|
||||
|
||||
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
|
||||
{
|
||||
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) {
|
||||
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)
|
||||
|
||||
defer func() {
|
||||
@@ -752,8 +756,8 @@ func TestHandler_gcCache(t *testing.T) {
|
||||
Key: "test_key_2",
|
||||
Version: "test_version",
|
||||
Complete: false,
|
||||
UsedAt: now.Add(-(keepTemp + time.Second)).Unix(),
|
||||
CreatedAt: now.Add(-(keepTemp + time.Hour)).Unix(),
|
||||
UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(),
|
||||
CreatedAt: now.Add(-(inUseGrace + time.Hour)).Unix(),
|
||||
},
|
||||
Kept: false,
|
||||
},
|
||||
@@ -763,21 +767,21 @@ func TestHandler_gcCache(t *testing.T) {
|
||||
Key: "test_key_3",
|
||||
Version: "test_version",
|
||||
Complete: true,
|
||||
UsedAt: now.Add(-(keepUnused + time.Second)).Unix(),
|
||||
CreatedAt: now.Add(-(keepUnused + time.Hour)).Unix(),
|
||||
UsedAt: now.Add(-(testRetention + time.Second)).Unix(),
|
||||
CreatedAt: now.Add(-(testRetention + time.Hour)).Unix(),
|
||||
},
|
||||
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{
|
||||
Key: "test_key_3",
|
||||
Version: "test_version",
|
||||
Complete: true,
|
||||
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.
|
||||
@@ -785,7 +789,7 @@ func TestHandler_gcCache(t *testing.T) {
|
||||
Key: "test_key_1",
|
||||
Version: "test_version",
|
||||
Complete: true,
|
||||
UsedAt: now.Add(-(keepOld - time.Minute)).Unix(),
|
||||
UsedAt: now.Add(-(inUseGrace - time.Minute)).Unix(),
|
||||
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
||||
},
|
||||
Kept: true,
|
||||
@@ -796,7 +800,7 @@ func TestHandler_gcCache(t *testing.T) {
|
||||
Key: "test_key_1",
|
||||
Version: "test_version",
|
||||
Complete: true,
|
||||
UsedAt: now.Add(-(keepOld + time.Second)).Unix(),
|
||||
UsedAt: now.Add(-(inUseGrace + time.Second)).Unix(),
|
||||
CreatedAt: now.Add(-(time.Hour + time.Second)).Unix(),
|
||||
},
|
||||
Kept: false,
|
||||
@@ -829,11 +833,265 @@ func TestHandler_gcCache(t *testing.T) {
|
||||
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:
|
||||
// unauthenticated access to management endpoints is now refused with 401.
|
||||
func TestHandler_RejectsMissingBearer(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -866,7 +1124,7 @@ func TestHandler_RejectsMissingBearer(t *testing.T) {
|
||||
// accepted after RegisterJob; stale/forged tokens cannot be replayed.
|
||||
func TestHandler_RejectsUnknownBearer(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
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.
|
||||
func TestHandler_UnregisterRevokes(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -917,7 +1175,7 @@ func TestHandler_UnregisterRevokes(t *testing.T) {
|
||||
// invisible to queries scoped to repoB.
|
||||
func TestHandler_CrossRepoIsolation(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
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.
|
||||
func TestHandler_ArtifactSignature(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
handler.RegisterJob(testToken, JobCredential{Repo: testRepo})
|
||||
@@ -1016,7 +1274,7 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
||||
|
||||
t.Run("signature from a different server", func(t *testing.T) {
|
||||
dir2 := filepath.Join(t.TempDir(), "artifactcache2")
|
||||
other, err := StartHandler(dir2, "", 0, "", nil)
|
||||
other, err := StartHandler(Options{Dir: dir2})
|
||||
require.NoError(t, err)
|
||||
defer other.Close()
|
||||
otherURL := signArtifactURL(other, 1)
|
||||
@@ -1038,13 +1296,13 @@ func TestHandler_ArtifactSignature(t *testing.T) {
|
||||
func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
|
||||
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)
|
||||
exp := time.Now().Add(artifactURLTTL).Unix()
|
||||
sig := first.computeSignature("", 42, exp)
|
||||
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)
|
||||
defer second.Close()
|
||||
|
||||
@@ -1056,7 +1314,7 @@ func TestHandler_SecretPersistsAcrossRestarts(t *testing.T) {
|
||||
// the auth refactor.
|
||||
func TestHandler_ArtifactSignatureDownload(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
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.
|
||||
func TestHandler_RegisterJob_RefCounted(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -1125,10 +1383,10 @@ func TestHandler_RegisterJob_RefCounted(t *testing.T) {
|
||||
|
||||
// TestHandler_GC_PerRepoDedup ensures duplicate-pruning does not evict
|
||||
// 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) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
handler.RegisterJob("tok-a", JobCredential{Repo: "owner/repoA"})
|
||||
@@ -1142,7 +1400,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) {
|
||||
db, err := handler.openDB()
|
||||
require.NoError(t, err)
|
||||
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}
|
||||
b := &Cache{Repo: "owner/repoB", Key: key, Version: version, Complete: true, CreatedAt: now, UsedAt: now, Size: 1}
|
||||
require.NoError(t, insertCache(db, a))
|
||||
@@ -1179,7 +1437,7 @@ func TestHandler_GC_PerRepoDedup(t *testing.T) {
|
||||
// register/revoke when the feature is off.
|
||||
func TestHandler_InternalAPI_Disabled(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
handler, err := StartHandler(dir, "", 0, "", nil)
|
||||
handler, err := StartHandler(Options{Dir: dir})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -1197,7 +1455,7 @@ func TestHandler_InternalAPI_Disabled(t *testing.T) {
|
||||
func TestHandler_InternalAPI_AuthAndUsage(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "artifactcache")
|
||||
const secret = "internal-secret"
|
||||
handler, err := StartHandler(dir, "", 0, secret, nil)
|
||||
handler, err := StartHandler(Options{Dir: dir, InternalSecret: secret})
|
||||
require.NoError(t, err)
|
||||
defer handler.Close()
|
||||
|
||||
|
||||
@@ -77,6 +77,7 @@ func (h *Handler) v2CreateCacheEntry(w http.ResponseWriter, r *http.Request, _ h
|
||||
h.twirpError(w, r, twirpInternal, err)
|
||||
return
|
||||
} 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)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -10,7 +10,6 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -67,16 +66,6 @@ func getURL(t *testing.T, url string) []byte {
|
||||
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
|
||||
// with the upload URL it used.
|
||||
func saveV2(t *testing.T, handler *Handler, key, version string, content []byte) (finalized map[string]any, uploadURL string) {
|
||||
@@ -98,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
|
||||
// or to replace a finalized entry.
|
||||
func TestCacheServiceV2RoundTrip(t *testing.T) {
|
||||
handler := startTestHandler(t)
|
||||
handler := newTestHandler(t, Policy{})
|
||||
content := []byte("the cached archive")
|
||||
|
||||
unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath)
|
||||
@@ -127,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
|
||||
// blocks that arrive out of order must still be assembled the way the client asked.
|
||||
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"})
|
||||
uploadURL, _ := created["signed_upload_url"].(string)
|
||||
@@ -164,7 +153,7 @@ func TestCacheServiceV2BlockUpload(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestCacheServiceV2Lookups(t *testing.T) {
|
||||
handler := startTestHandler(t)
|
||||
handler := newTestHandler(t, Policy{})
|
||||
saved, _ := saveV2(t, handler, "deps-abc", "v1", []byte("x"))
|
||||
require.Equal(t, true, saved["ok"])
|
||||
|
||||
|
||||
@@ -24,7 +24,7 @@ func TestFrontResultsService(t *testing.T) {
|
||||
}))
|
||||
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)
|
||||
defer handler.Close()
|
||||
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)
|
||||
}
|
||||
|
||||
func (s *Storage) Remove(id uint64) {
|
||||
_ = os.Remove(s.filename(id))
|
||||
_ = os.RemoveAll(s.tempDir(id))
|
||||
// Remove deletes an entry's blob and any staged parts. It reports failure so the caller can
|
||||
// keep the entry and retry, rather than dropping the only reference to bytes on disk.
|
||||
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 {
|
||||
|
||||
@@ -144,7 +144,7 @@ func TestCacheServiceV2EndToEnd(t *testing.T) {
|
||||
restore := patchedAction(t, "actions/cache", actionsCacheRef, "dist/restore/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)
|
||||
t.Cleanup(func() { _ = handler.Close() })
|
||||
const token, repo = "e2e-runtime-token", "testuser/testrepo"
|
||||
@@ -291,7 +291,7 @@ func TestUploadArtifactThroughTheResultsService(t *testing.T) {
|
||||
}))
|
||||
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)
|
||||
t.Cleanup(func() { _ = handler.Close() })
|
||||
// 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")
|
||||
|
||||
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)
|
||||
t.Cleanup(func() { _ = handler.Close() })
|
||||
const token = "setup-runtime-token"
|
||||
|
||||
@@ -14,6 +14,7 @@ require (
|
||||
github.com/distribution/reference v0.6.0
|
||||
github.com/docker/cli v29.7.1+incompatible
|
||||
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-git/v5 v5.19.2
|
||||
github.com/gobwas/glob v0.2.3
|
||||
@@ -61,7 +62,6 @@ require (
|
||||
github.com/cyphar/filepath-securejoin v0.6.1 // indirect
|
||||
github.com/davecgh/go-spew v1.1.1 // 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/fatih/color v1.19.0 // indirect
|
||||
github.com/felixge/httpsnoop v1.0.4 // indirect
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"os/signal"
|
||||
|
||||
"gitea.com/gitea/runner/act/artifactcache"
|
||||
"gitea.com/gitea/runner/internal/app/run"
|
||||
"gitea.com/gitea/runner/internal/pkg/config"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
@@ -52,13 +53,14 @@ func runCacheServer(configFile *string, cacheArgs *cacheServerArgs) func(cmd *co
|
||||
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")
|
||||
}
|
||||
cacheHandler, err := artifactcache.StartHandler(
|
||||
dir,
|
||||
host,
|
||||
port,
|
||||
secret,
|
||||
log.StandardLogger().WithField("module", "cache_request"),
|
||||
)
|
||||
cacheHandler, err := artifactcache.StartHandler(artifactcache.Options{
|
||||
Dir: dir,
|
||||
OutboundIP: host,
|
||||
Port: port,
|
||||
InternalSecret: secret,
|
||||
Policy: run.CachePolicy(cfg),
|
||||
Logger: log.StandardLogger().WithField("module", "cache_request"),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@ import (
|
||||
"gitea.com/gitea/runner/act/common"
|
||||
"gitea.com/gitea/runner/act/runner"
|
||||
"gitea.com/gitea/runner/internal/app/run"
|
||||
"gitea.com/gitea/runner/internal/pkg/config"
|
||||
|
||||
"gitea.dev/actionslib/pkg/model"
|
||||
"github.com/joho/godotenv"
|
||||
@@ -374,7 +375,10 @@ func runExec(ctx context.Context, execArgs *executeArgs) func(cmd *cobra.Command
|
||||
}
|
||||
|
||||
// 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 {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -27,6 +27,7 @@ import (
|
||||
"gitea.com/gitea/runner/act/runner"
|
||||
"gitea.com/gitea/runner/internal/pkg/client"
|
||||
"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/metrics"
|
||||
"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
|
||||
if cfg.Cache.Enabled == nil || *cfg.Cache.Enabled {
|
||||
if cfg.Cache.ExternalServer != "" {
|
||||
warnIgnoredCachePolicy(cfg)
|
||||
// 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, "/") + "/"
|
||||
} else {
|
||||
warnIgnoredCacheSecret(cfg)
|
||||
handler, err := artifactcache.StartHandler(
|
||||
cfg.Cache.Dir,
|
||||
cfg.Cache.Host,
|
||||
cfg.Cache.Port,
|
||||
"",
|
||||
log.StandardLogger().WithField("module", "cache_request"),
|
||||
)
|
||||
handler, err := artifactcache.StartHandler(artifactcache.Options{
|
||||
Dir: cfg.Cache.Dir,
|
||||
OutboundIP: cfg.Cache.Host,
|
||||
Port: cfg.Cache.Port,
|
||||
Policy: CachePolicy(cfg),
|
||||
Logger: log.StandardLogger().WithField("module", "cache_request"),
|
||||
})
|
||||
if err != nil {
|
||||
log.Errorf("cannot init cache server, it will be disabled: %v", err)
|
||||
// go on
|
||||
@@ -693,7 +695,7 @@ func checkFreeDisk(cfg *config.Config) (bool, string) {
|
||||
root = filepath.FromSlash("/" + strings.TrimLeft(cfg.Container.WorkdirParent, "/"))
|
||||
}
|
||||
root = nearestExistingPath(root)
|
||||
available, err := freeDiskBytes(root)
|
||||
available, err := disk.FreeBytes(root)
|
||||
if err != nil {
|
||||
return false, fmt.Sprintf("cannot determine free disk space for %s: %v", root, err)
|
||||
}
|
||||
@@ -726,6 +728,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.
|
||||
func warnIgnoredCacheSecret(cfg *config.Config) {
|
||||
if cfg.Cache.ExternalServer != "" {
|
||||
|
||||
@@ -25,7 +25,7 @@ func emptyCfg() *config.Config { return &config.Config{} }
|
||||
|
||||
func TestRunner_registerCacheForTask(t *testing.T) {
|
||||
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)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -62,7 +62,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) {
|
||||
|
||||
t.Run("empty token", func(t *testing.T) {
|
||||
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)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -77,7 +77,7 @@ func TestRunner_registerCacheForTask_NoOps(t *testing.T) {
|
||||
// /find, no auth on the signed archiveLocation download.
|
||||
func TestRunner_CacheFullFlow_MatchesToolkit(t *testing.T) {
|
||||
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)
|
||||
defer handler.Close()
|
||||
|
||||
@@ -162,7 +162,7 @@ func decodeJSON(resp *http.Response, v any) error {
|
||||
func TestRunner_ExternalCacheServer_RegisterRevoke(t *testing.T) {
|
||||
dir := filepath.Join(t.TempDir(), "remote-cache")
|
||||
const secret = "shared-secret-for-tests"
|
||||
remote, err := artifactcache.StartHandler(dir, "127.0.0.2", 0, secret, nil) // advertised, never dialled
|
||||
remote, err := artifactcache.StartHandler(artifactcache.Options{Dir: dir, OutboundIP: "127.0.0.2", InternalSecret: secret}) // advertised, never dialled
|
||||
require.NoError(t, err)
|
||||
defer remote.Close()
|
||||
external := strings.Replace(remote.ExternalURL(), "127.0.0.2", "127.0.0.1", 1)
|
||||
|
||||
@@ -159,6 +159,20 @@ cache:
|
||||
# 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.
|
||||
#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:
|
||||
# Specifies the network to which the container will connect.
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/docker/go-units"
|
||||
"github.com/joho/godotenv"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"go.yaml.in/yaml/v4"
|
||||
@@ -81,6 +82,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.
|
||||
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.
|
||||
|
||||
// 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.
|
||||
@@ -142,7 +184,7 @@ type Config struct {
|
||||
// LoadDefault returns the default configuration.
|
||||
// If file is not empty, it will be used to load the configuration.
|
||||
func LoadDefault(file string) (*Config, error) {
|
||||
cfg := &Config{}
|
||||
cfg := &Config{Cache: DefaultCache()}
|
||||
definedRunnerKeys := map[string]bool{}
|
||||
if file != "" {
|
||||
content, err := os.ReadFile(file)
|
||||
|
||||
@@ -168,6 +168,37 @@ runner:
|
||||
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) {
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "config.yaml")
|
||||
|
||||
@@ -26,7 +26,10 @@ const (
|
||||
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.
|
||||
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
|
||||
}
|
||||
|
||||
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() {
|
||||
case reflect.Bool:
|
||||
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
|
||||
|
||||
package run
|
||||
package disk
|
||||
|
||||
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)
|
||||
}
|
||||
@@ -3,11 +3,12 @@
|
||||
|
||||
//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
|
||||
|
||||
package run
|
||||
package disk
|
||||
|
||||
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
|
||||
if err := unix.Statfs(path, &stat); err != nil {
|
||||
return 0, err
|
||||
@@ -3,11 +3,12 @@
|
||||
|
||||
//go:build windows
|
||||
|
||||
package run
|
||||
package disk
|
||||
|
||||
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)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
Reference in New Issue
Block a user