remote.go
1package runner
2
3import (
4 "archive/tar"
5 "compress/gzip"
6 "context"
7 "errors"
8 "fmt"
9 "io"
10 "net/http"
11 "os"
12 "path/filepath"
13 "regexp"
14 "strings"
15 "sync"
16 "time"
17
18 "git.kilimanjaro.io/ci/internal/checkout"
19 "git.kilimanjaro.io/ci/internal/engine"
20 "git.kilimanjaro.io/ci/internal/lex"
21)
22
23// Remote locks: a local run (`ci run` on the workstation) takes a
24// concurrency group from the runner, as a lease on the same file lock the
25// runner's segments use, so a laptop deploy and a CI deploy never overlap.
26// The client renews the lease while it holds the group; a lease not renewed
27// within its TTL is released.
28
29// LeaseTTL is how long a lease lives without renewal.
30const LeaseTTL = 2 * time.Minute
31
32type lease struct {
33 holder string
34 release func()
35 expires time.Time
36}
37
38type leases struct {
39 mu sync.Mutex
40 m map[string]*lease
41}
42
43func (r *Runner) leaseTable() *leases {
44 r.mu.Lock()
45 defer r.mu.Unlock()
46 if r.leases == nil {
47 r.leases = &leases{m: map[string]*lease{}}
48 }
49 return r.leases
50}
51
52var groupName = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`)
53
54// Lock takes or renews group for holder; it returns who holds it if not.
55func (r *Runner) Lock(ctx context.Context, group, holder string) (bool, string, error) {
56 if !groupName.MatchString(group) || holder == "" {
57 return false, "", errors.New("invalid group or holder")
58 }
59 t := r.leaseTable()
60 t.mu.Lock()
61 defer t.mu.Unlock()
62 if l, ok := t.m[group]; ok {
63 if l.holder == holder {
64 l.expires = time.Now().Add(LeaseTTL)
65 return true, holder, nil
66 }
67 if time.Now().Before(l.expires) {
68 return false, l.holder, nil
69 }
70 l.release()
71 delete(t.m, group)
72 }
73 release, by, err := engine.FileLocker{Dir: r.Cfg.LocksDir}.TryLock(ctx, group, holder)
74 if err != nil || release == nil {
75 return false, by, err
76 }
77 t.m[group] = &lease{holder: holder, release: release, expires: time.Now().Add(LeaseTTL)}
78 return true, holder, nil
79}
80
81// Unlock releases holder's lease on group.
82func (r *Runner) Unlock(group, holder string) {
83 t := r.leaseTable()
84 t.mu.Lock()
85 defer t.mu.Unlock()
86 if l, ok := t.m[group]; ok && l.holder == holder {
87 l.release()
88 delete(t.m, group)
89 }
90}
91
92func (r *Runner) serveLock(w http.ResponseWriter, req *http.Request) {
93 holder := req.URL.Query().Get("holder")
94 if req.Method == http.MethodDelete {
95 r.Unlock(req.PathValue("group"), holder)
96 writeJSON(w, 200, map[string]any{"released": true})
97 return
98 }
99 ok, by, err := r.Lock(req.Context(), req.PathValue("group"), holder)
100 if err != nil {
101 fail(w, 400, err)
102 return
103 }
104 code := 200
105 if !ok {
106 code = 409
107 }
108 writeJSON(w, code, map[string]any{"locked": ok, "holder": by, "ttl": LeaseTTL.Seconds()})
109}
110
111// Imported local runs: `ci run --report` uploads a snapshot of a local run
112// (state, events, results, artifacts; no checkouts) so it shows in the
113// appview's history, and its manual actions (rollback) can run here.
114
115var runID = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{5,80}$`)
116
117// importable names what a snapshot may contain.
118func importable(name string) bool {
119 switch name {
120 case "state.json", "events.ndjson", "manifest.yml":
121 return true
122 }
123 return strings.HasPrefix(name, "state/") || strings.HasPrefix(name, "results/") || strings.HasPrefix(name, "artifacts/")
124}
125
126// Import stores a reported local run (a tar.gz of its run dir) as run id.
127func (r *Runner) Import(id string, body io.Reader) error {
128 if !runID.MatchString(id) {
129 return fmt.Errorf("invalid run id %q", id)
130 }
131 if x, err := r.GetRun(id); err == nil && x.Pipeline != "" {
132 return fmt.Errorf("run %s is not an imported run", id)
133 }
134 dir := filepath.Join(r.Cfg.RunsDir, id)
135 tmp := dir + ".import"
136 _ = os.RemoveAll(tmp)
137 if err := os.MkdirAll(tmp, 0o755); err != nil {
138 return err
139 }
140 defer func() { _ = os.RemoveAll(tmp) }()
141 zr, err := gzip.NewReader(io.LimitReader(body, 256<<20))
142 if err != nil {
143 return err
144 }
145 tr := tar.NewReader(zr)
146 for {
147 h, err := tr.Next()
148 if err == io.EOF {
149 break
150 }
151 if err != nil {
152 return err
153 }
154 name := filepath.Clean(h.Name)
155 if h.Typeflag != tar.TypeReg || !importable(name) || strings.Contains(name, "..") {
156 continue // checkouts, links and anything else stay behind
157 }
158 p := filepath.Join(tmp, name)
159 if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
160 return err
161 }
162 f, err := os.Create(p)
163 if err != nil {
164 return err
165 }
166 _, err = io.Copy(f, io.LimitReader(tr, 64<<20))
167 f.Close()
168 if err != nil {
169 return err
170 }
171 }
172 st, err := engine.Load(engine.Dir(tmp))
173 if err != nil {
174 return fmt.Errorf("snapshot: %w", err)
175 }
176 if st.Meta.Run != id {
177 return fmt.Errorf("snapshot is of run %s, not %s", st.Meta.Run, id)
178 }
179 if st.Meta.Host != "local" {
180 return errors.New("only local runs can be imported")
181 }
182 _ = os.RemoveAll(dir)
183 if err := os.Rename(tmp, dir); err != nil {
184 return err
185 }
186 if err := os.MkdirAll(tmp, 0o755); err != nil { // for the deferred RemoveAll
187 return err
188 }
189 if _, err := r.GetRun(id); err == nil {
190 return r.updateRun(id, `status = ?, dir = ?`, st.Status, dir)
191 }
192 return r.insertRun(Run{ID: id, Workflow: st.Meta.Workflow, Repo: st.Meta.Repo, Dir: dir,
193 Status: st.Status, Created: st.Created, Updated: now()})
194}
195
196func (r *Runner) serveImport(w http.ResponseWriter, req *http.Request) {
197 if err := r.Import(req.PathValue("id"), req.Body); err != nil {
198 fail(w, 400, err)
199 return
200 }
201 writeJSON(w, 200, map[string]string{"run": req.PathValue("id")})
202}
203
204// relayout gives a manual run of an imported local run checkouts of its
205// own: the local paths mean nothing here, so each repo is checked out again
206// at the commit the local run recorded. A dirty working copy can't be
207// reproduced and is refused.
208func (r *Runner) relayout(ctx context.Context, meta *engine.Meta, m []lex.Repo, dir engine.Dir) error {
209 if meta.Dirty {
210 return errors.New("the local run used a working copy with changes; run its actions locally (ci action --run)")
211 }
212 if meta.Sha == "" {
213 return errors.New("the local run recorded no commit")
214 }
215 revs := map[string]string{}
216 for _, c := range meta.Checkouts {
217 if c.Rev == "linked" || c.Rev == "" {
218 if c.Repo != meta.Repo {
219 return fmt.Errorf("repo %s was a linked local directory (--repo); run its actions locally", c.Repo)
220 }
221 continue
222 }
223 revs[c.Repo] = c.Rev
224 }
225 mirror, err := r.src.Mirror(ctx, meta.Repo)
226 if err != nil {
227 return err
228 }
229 origin, _ := r.src.URL(meta.Repo)
230 extras := make([]lex.Repo, len(m))
231 for i, e := range m {
232 e.Rev = revs[e.Repo]
233 extras[i] = e
234 }
235 checked, err := checkout.Layout(ctx, r.src, dir.Path(), checkout.Main{Repo: meta.Repo, From: mirror, Sha: meta.Sha, Origin: origin}, extras, nil)
236 if err != nil {
237 return err
238 }
239 meta.Root = filepath.Join(dir.Path(), meta.Repo)
240 meta.Checkouts = nil
241 for _, c := range checked {
242 meta.Checkouts = append(meta.Checkouts, engine.Checkout{Repo: c.Repo, Path: c.Path, Rev: c.Rev, Env: c.Env})
243 }
244 meta.Host = "runner"
245 return nil
246}
