1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "encoding/base64"
8 "errors"
9 "fmt"
10 "io"
11 "net/http"
12 "time"
13
14 ermite "bnema.dev/ermite-api/gen/ermite"
15
16 "bnema.dev/ermite-runner/internal/actionscache"
17 "bnema.dev/ermite-runner/internal/execstream/execpb"
18)
19
20const (
21 // archiveFetchTimeout and maxArchiveBytes are the executor's limits for an
22 // archive (actions_download.go).
23 archiveFetchTimeout = 20 * time.Minute
24 maxArchiveBytes = 1 << 30
25)
26
27// cacheArchives points the actions of a resolution at the host's archive cache.
28// The host downloads each archive once per (forge, repository, resolved sha) and
29// the sandbox reads it from the cache channel instead of the forge, so that a
30// busy runner does not trip the forge's rate limit (codeload 429).
31//
32// It only ever improves a resolution: an action that cannot be cached (no cache,
33// no cache scope for this job, a ref that did not resolve to a sha, a download that
34// failed, a full disk) keeps its original URL and token, and the sandbox
35// downloads it as before, with its retries and its messages. The sandbox prints
36// "Download action repository" in either case, so a hit reads like a miss in the
37// job's log; the hit or miss is logged here.
38func (j *jobRun) cacheArchives(ctx context.Context, orig []*ermite.ResolvedAction, conv []*execpb.ResolvedAction) {
39 rt := j.r
40 if rt == nil || rt.Archives == nil || rt.Cache == nil || len(orig) != len(conv) {
41 return
42 }
43 if _, ok, err := cacheScope(j.job, j.contexts); err != nil || !ok {
44 return // action archives are served only to jobs with a cache scope
45 }
46 ttl := time.Duration(j.timeoutMinutes())*time.Minute + cacheTokenMargin
47 for i, a := range orig {
48 c := conv[i]
49 if c.Error != "" || a.Error != "" || a.TarballUrl == "" {
50 continue
51 }
52 origin := originOf(a.TarballUrl)
53 key, ok := actionscache.Key(origin, a.NameWithOwner, a.ResolvedSha)
54 if !ok {
55 continue
56 }
57 // fetched: this call ran the download itself. An invalid key, a wait
58 // that was cancelled or a joined download that failed are no misses.
59 var fetched bool
60 var fetchTime time.Duration
61 var bytes int64
62 hit, verify, err := rt.Archives.Ensure(ctx, key, func(ctx context.Context, w io.Writer) error {
63 fetched = true
64 start := time.Now()
65 cw := &archiveCountWriter{Writer: w}
66 err := j.fetchArchive(ctx, a, origin, cw)
67 fetchTime += time.Since(start)
68 bytes += cw.n
69 return err
70 })
71 if fetched {
72 j.mu.Lock()
73 j.timeline.addMs("archive_miss_ms", fetchTime)
74 j.timeline.add("archive_misses", 1)
75 j.timeline.add("archive_miss_bytes", bytes)
76 j.mu.Unlock()
77 }
78 log := j.log.With().Str("action", a.NameWithOwner).Str("sha", a.ResolvedSha).Logger()
79 if err != nil {
80 log.Warn().Err(err).Msg("action archive not cached; the sandbox downloads it")
81 continue
82 }
83 u := rt.Cache.ArchiveURL(j.job.JobId, key, ttl)
84 if u == "" {
85 continue
86 }
87 event := log.Info().Bool("hit", hit).Int64("verify_ms", verify.Milliseconds())
88 if fetched {
89 event = event.Int64("fetch_ms", fetchTime.Milliseconds()).Int64("bytes", bytes)
90 }
91 event.Msg("action archive cache")
92 c.TarballUrl, c.Token, c.AuthOrigin = u, "", ""
93 }
94}
95
96// fetchArchive downloads the archive of a from the forge, with the token only
97// when the server gave one for an origin that is ermite (like resolveResponse).
98func (j *jobRun) fetchArchive(ctx context.Context, a *ermite.ResolvedAction, origin string, w io.Writer) error {
99 token := ""
100 if a.Source == ermite.ActionSource_ACTION_SOURCE_ERMITE && a.Token != "" && (origin == j.publicOrigin || origin == j.serverOrigin) {
101 token = a.Token
102 }
103 ctx, cancel := context.WithTimeout(ctx, archiveFetchTimeout)
104 defer cancel()
105 req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.TarballUrl, nil)
106 if err != nil {
107 return err
108 }
109 if token != "" {
110 req.Header.Set("Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte("x-access-token:"+token)))
111 }
112 req.Header.Set("User-Agent", "ermite-runner")
113 resp, err := actionscache.NewClient(j.r.ArchiveHTTP).Do(req)
114 if err != nil {
115 return err
116 }
117 defer func() { _ = resp.Body.Close() }()
118 if resp.StatusCode < 200 || resp.StatusCode > 299 {
119 return fmt.Errorf("status %d", resp.StatusCode)
120 }
121 n, err := io.Copy(w, io.LimitReader(resp.Body, maxArchiveBytes+1))
122 if err == nil && n > maxArchiveBytes {
123 err = errors.New("the archive is larger than 1 GiB")
124 }
125 return err
126}
127
128// archiveCountWriter counts bytes accepted by the cache.
129type archiveCountWriter struct {
130 io.Writer
131 n int64
132}
133
134func (w *archiveCountWriter) Write(p []byte) (int, error) {
135 n, err := w.Writer.Write(p)
136 w.n += int64(n)
137 return n, err
138}