1// SPDX-License-Identifier: MIT
2
3// Package cachesvc is the GitHub Actions cache service (API v1) of the runner
4// (plan CI-3, D8). Actions that use @actions/cache (setup-go, setup-node,
5// actions/cache…) talk to it through ACTIONS_CACHE_URL: the toolkit takes the v1
6// protocol for any server URL that is not github.com, *.ghe.com or *.localhost
7// (toolkit/packages/cache/src/internal/config.ts:1-20).
8//
9// One Service per runner host. Entries live under <state>/cache/ (files plus a
10// JSON index, atomic finalisation, LRU eviction by total size) and are scoped by
11// (repository, ref, key, version). A job reaches the service with a bearer token
12// (ACTIONS_RUNTIME_TOKEN): a JWT signed with HS256 under a key that exists only
13// in the memory of the runner, over its job id, repository, write ref, read
14// refs, expiry and scopes. The same token serves the results service of the
15// job (package resultssvc), which the channel of a sandbox carries beside the
16// cache (WithResults).
17//
18// The service never listens on a network. Each sandbox has a channel of its own
19// (Serve): a Unix socket, bind-mounted into a container or reached through
20// Firecracker's vsock proxy for a VM, on which the relay in the guest delivers
21// the connections of the actions, which reach it on the guest's loopback
22// (DefaultPort). The host opens no port and its firewall has no rule for it.
23package cachesvc
24
25import (
26 "errors"
27 "fmt"
28 "net"
29 "net/http"
30 "strconv"
31 "strings"
32 "sync"
33 "time"
34
35 "github.com/bnema/zerowrap"
36
37 "bnema.dev/ermite-runner/internal/actionscache"
38)
39
40// Defaults.
41const (
42 // DefaultMax is the size cap, GitHub's limit per repository.
43 DefaultMax = 10 << 30
44 // DefaultPort is the TCP port of the cache on the loopback of the guest
45 // (ACTIONS_CACHE_URL=http://127.0.0.1:9089/), and the vsock port of the
46 // cache streams of a VM. It is not a port of the host.
47 DefaultPort = 9089
48 // EnvMax is ERMITE_RUNNER_CACHE_MAX, the size cap (1G, 500M…).
49 EnvMax = "ERMITE_RUNNER_CACHE_MAX"
50 // DefaultReservationTTL is how long an upload may stay uncommitted.
51 DefaultReservationTTL = time.Hour
52 flushEvery = 30 * time.Second
53 // DefaultProgressTimeout is how long an upload or a download may make no
54 // progress before its connection is cut.
55 DefaultProgressTimeout = 60 * time.Second
56 // DefaultConnsPerListener and DefaultConns bound the open connections of one
57 // sandbox's listener and of the whole service. upload-artifact sends 5
58 // blocks at once by default, besides downloads and the cache.
59 DefaultConnsPerListener = 32
60 DefaultConns = 128
61 // DefaultRegistryPerChannel bounds the registry requests of one channel in
62 // progress at once, so that one sandbox's pulls leave room for its cache and
63 // artifact traffic. dockerd pulls 3 layers at once by default.
64 DefaultRegistryPerChannel = 8
65 // RegistryDownloadsPerChannel bounds the cold Docker Hub downloads one
66 // channel may have started at once (dockerd's default
67 // max-concurrent-downloads), so that one sandbox cannot take every slot of
68 // the registry proxy.
69 RegistryDownloadsPerChannel = 3
70 // RegistryPath is the prefix of the Docker registry API on the channel.
71 RegistryPath = "/v2/"
72)
73
74// Config configures a Service.
75type Config struct {
76 // Dir is the cache directory (<state>/cache).
77 Dir string
78 // Max is the size cap in bytes; 0 means DefaultMax.
79 Max int64
80 // ReservationTTL is the life of an uncommitted reservation; 0 means
81 // DefaultReservationTTL.
82 ReservationTTL time.Duration
83 // ReservationMaxLife is the absolute life of a reservation from its creation
84 // (the token's expiry still ends it sooner); 0 means DefaultReservationTTL.
85 ReservationMaxLife time.Duration
86 // Now is the clock; nil means time.Now.
87 Now func() time.Time
88 Log zerowrap.Logger
89 // Archives, when not nil, is the host's cache of action archives, served on
90 // the channels (see ArchiveURL).
91 Archives *actionscache.Cache
92 // Registry is an anonymous pull-only registry on the private channel.
93 Registry http.Handler
94 // The limits below take the Default… values (or those of the store) when
95 // zero; tests set them.
96 ProgressTimeout time.Duration
97 ConnsPerListener, Conns int
98 RegistryPerChannel int
99 MinCharge int64
100 MaxPending, MaxPendingPerJob int
101 // hookChunk, when set, is called by an upload once its range is admitted and
102 // before its body is read (tests: a barrier with a chunk in flight).
103 hookChunk func(id int64)
104}
105
106// Service is the cache service.
107type Service struct {
108 store *store
109 sign *signer
110 now func() time.Time
111 log zerowrap.Logger
112
113 mu sync.Mutex
114 channels map[string]*listener
115 stop chan struct{}
116 wg sync.WaitGroup
117 closed bool
118
119 archives *actionscache.Cache
120 registry http.Handler
121 // registryPerChannel caps the registry requests in progress per handler.
122 registryPerChannel int
123 hookChunk func(id int64)
124 progress time.Duration
125 perListener int
126 connsGlobal chan struct{}
127}
128
129type listener struct {
130 srv *http.Server
131 done chan struct{}
132 bw *bandwidth // the channel's limiter; nil when unlimited (read by tests)
133}
134
135// New opens the store and starts the housekeeping goroutine; Close stops it.
136func New(cfg Config) (*Service, error) {
137 if cfg.Max <= 0 {
138 cfg.Max = DefaultMax
139 }
140 if cfg.ReservationTTL <= 0 {
141 cfg.ReservationTTL = DefaultReservationTTL
142 }
143 if cfg.Now == nil {
144 cfg.Now = time.Now
145 }
146 sg, err := newSigner()
147 if err != nil {
148 return nil, err
149 }
150 st, err := openStore(storeConfig{dir: cfg.Dir, max: cfg.Max, ttl: cfg.ReservationTTL, now: cfg.Now,
151 minCharge: cfg.MinCharge, maxPending: cfg.MaxPending, maxPerJob: cfg.MaxPendingPerJob, maxLife: cfg.ReservationMaxLife, log: cfg.Log})
152 if err != nil {
153 return nil, err
154 }
155 if cfg.ProgressTimeout <= 0 {
156 cfg.ProgressTimeout = DefaultProgressTimeout
157 }
158 if cfg.ConnsPerListener <= 0 {
159 cfg.ConnsPerListener = DefaultConnsPerListener
160 }
161 if cfg.Conns <= 0 {
162 cfg.Conns = DefaultConns
163 }
164 if cfg.RegistryPerChannel <= 0 {
165 cfg.RegistryPerChannel = DefaultRegistryPerChannel
166 }
167 s := &Service{store: st, sign: sg, now: cfg.Now, log: cfg.Log, channels: map[string]*listener{}, stop: make(chan struct{}),
168 archives: cfg.Archives, registry: cfg.Registry, hookChunk: cfg.hookChunk, progress: cfg.ProgressTimeout, perListener: cfg.ConnsPerListener, connsGlobal: make(chan struct{}, cfg.Conns), registryPerChannel: cfg.RegistryPerChannel}
169 s.wg.Add(1)
170 go s.housekeeping()
171 return s, nil
172}
173
174// housekeeping writes the usage times of the index and drops stale reservations.
175func (s *Service) housekeeping() {
176 defer s.wg.Done()
177 t := time.NewTicker(flushEvery)
178 defer t.Stop()
179 for {
180 select {
181 case <-s.stop:
182 return
183 case <-t.C:
184 s.store.expire()
185 if err := s.store.flush(); err != nil {
186 s.log.Warn().Err(err).Msg("cache index")
187 }
188 }
189 }
190}
191
192// RegistryEnabled reports whether the channel serves an anonymous registry.
193func (s *Service) RegistryEnabled() bool { return s.registry != nil }
194
195// Token returns the ACTIONS_RUNTIME_TOKEN of a job. It is valid for ttl, which
196// should be the job timeout plus a margin. sc is the job's cache scope (the
197// zero Scope for a job without a cache: the cache then refuses the token);
198// scopes are added to its scp claim (resultssvc.Scope for a job with the
199// results service).
200func (s *Service) Token(jobID string, sc Scope, ttl time.Duration, scopes ...string) string {
201 now := s.now()
202 exp := now.Add(ttl).Unix()
203 return s.sign.token(Claims{JobID: jobID, Scope: sc, Expiry: exp, ExpiresAt: exp, IssuedAt: now.Unix(),
204 Scopes: strings.Join(append([]string{genericReadScope}, scopes...), " ")})
205}
206
207// RuntimeToken checks a runtime token for the results service (resultssvc.Auth):
208// made by this runner and not expired, whatever its cache scope. It returns
209// the job, the scp claim and the expiry.
210func (s *Service) RuntimeToken(tok string) (jobID, scopes string, expiry time.Time, err error) {
211 c, err := s.sign.verifyAny(tok, s.now())
212 if err != nil {
213 return "", "", time.Time{}, err
214 }
215 return c.JobID, c.Scopes, time.Unix(c.Expiry, 0), nil
216}
217
218// BlobMAC signs the blob URLs of the results service (resultssvc.Auth), under
219// a key domain of their own.
220func (s *Service) BlobMAC(parts ...[]byte) []byte {
221 return s.sign.mac("blob", parts...)
222}
223
224// start serves HTTP on ln, subject to the connection caps, until the listener
225// is closed. job, when not empty, restricts the tokens accepted to that job's.
226func (s *Service) start(name string, ln net.Listener, job string, bw *bandwidth, results http.Handler) *listener {
227 if bw != nil {
228 ln = throttleListener{ln, bw}
229 }
230 ln = newLimitListener(ln, s.perListener, s.connsGlobal)
231 l := &listener{
232 srv: &http.Server{
233 Handler: s.channelHandler(job, results),
234 ReadHeaderTimeout: 30 * time.Second,
235 // Bodies are chunks of up to tens of MiB and archives of GiB: only
236 // idle connections time out.
237 IdleTimeout: 2 * time.Minute,
238 },
239 done: make(chan struct{}),
240 bw: bw,
241 }
242 go func() {
243 defer close(l.done)
244 if err := l.srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
245 s.log.Error().Err(err).Str("listener", name).Msg("cache listener")
246 }
247 }()
248 return l
249}
250
251// Serve serves the cache on ln, the host end of the channel of one sandbox
252// (id): a listener that is not on the network, such as the Unix socket on which
253// the relay of that sandbox's guest delivers its connections. Each sandbox has
254// its own listener and so its own handler, and nothing but that sandbox's guest
255// can connect to it. job is the id of the job the sandbox runs: only tokens made
256// for it are accepted on this channel, besides the checks of every token. The
257// connection caps apply as on any listener, and so does WithBandwidth when given:
258// one limiter per call, shared by all the connections of the channel, so that
259// the sandbox's cache traffic as a whole stays under its rate. Serve takes
260// ownership of ln: Stop (or CloseAll) closes it, and so does a failed Serve.
261func (s *Service) Serve(id string, ln net.Listener, job string, opts ...ServeOption) error {
262 var sc serveConfig
263 for _, o := range opts {
264 o(&sc)
265 }
266 if job == "" {
267 _ = ln.Close()
268 return errors.New("cache: a channel needs the job it serves")
269 }
270 s.mu.Lock()
271 defer s.mu.Unlock()
272 if s.closed {
273 _ = ln.Close()
274 return errors.New("cache: service closed")
275 }
276 if _, ok := s.channels[id]; ok {
277 _ = ln.Close()
278 return fmt.Errorf("cache: already serving %s", id)
279 }
280 s.channels[id] = s.start(id, ln, job, newBandwidth(sc.bytesPerSec), sc.results)
281 return nil
282}
283
284// Stop ends the service of the channel of sandbox id, closing its listener and
285// every connection. Unknown ids are ignored.
286func (s *Service) Stop(id string) {
287 s.mu.Lock()
288 l := s.channels[id]
289 delete(s.channels, id)
290 s.mu.Unlock()
291 if l != nil {
292 _ = l.srv.Close()
293 <-l.done
294 }
295}
296
297// CloseAll stops every channel and the housekeeping, and writes the index.
298func (s *Service) CloseAll() error {
299 s.mu.Lock()
300 var ids []string
301 for id := range s.channels {
302 ids = append(ids, id)
303 }
304 already := s.closed
305 s.closed = true
306 s.mu.Unlock()
307 for _, id := range ids {
308 s.Stop(id)
309 }
310 if already {
311 return nil
312 }
313 close(s.stop)
314 s.wg.Wait()
315 return s.store.flush()
316}
317
318// ParseSize reads a size: a number of bytes or a number with a K, M, G or T
319// suffix (powers of 1024), optionally followed by iB or B, such as 10G, 500M
320// or 2GiB. Surrounding spaces are ignored; empty means def. Zero in any unit
321// (0, 0G, 0GiB) is 0. The error names no variable: the caller prefixes it.
322func ParseSize(v string, def int64) (int64, error) {
323 v = strings.TrimSpace(v)
324 if v == "" {
325 return def, nil
326 }
327 s := []byte(v)
328 for len(s) > 0 && strings.IndexByte("bBiI", s[len(s)-1]) >= 0 {
329 s = s[:len(s)-1]
330 }
331 mult := int64(1)
332 if n := len(s); n > 0 {
333 switch s[n-1] {
334 case 'k', 'K':
335 mult, s = 1<<10, s[:n-1]
336 case 'm', 'M':
337 mult, s = 1<<20, s[:n-1]
338 case 'g', 'G':
339 mult, s = 1<<30, s[:n-1]
340 case 't', 'T':
341 mult, s = 1<<40, s[:n-1]
342 }
343 }
344 n, err := strconv.ParseInt(string(s), 10, 64)
345 if err != nil || n < 0 || n > (1<<62)/mult {
346 return 0, fmt.Errorf("%q is not a size (examples: 10G, 500M)", v)
347 }
348 return n * mult, nil
349}