1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "fmt"
8 "strings"
9 "time"
10 "unicode/utf8"
11
12 "connectrpc.com/connect"
13
14 ermite "bnema.dev/ermite-api/gen/ermite"
15
16 "bnema.dev/ermite-runner/internal/execstream/execpb"
17)
18
19// Features of the server that the host uses (GetVersionResponse.features).
20const (
21 FeatureAnnotations = "ci.annotations"
22 FeatureSummaries = "ci.summaries"
23)
24
25// Limits of Annotation and UploadSummaryRequest (runner.proto). The executor
26// keeps within them; the guest is untrusted code, so the host enforces them
27// again, by dropping or cutting what is over rather than failing the job: the
28// server rejects a whole call that breaks one.
29const (
30 maxAnnotationsPerLevel = 10
31 maxAnnotationMessage = 4096
32 maxAnnotationTitle = 255
33 maxAnnotationFile = 1024
34 maxSummaryBytes = 1 << 20
35)
36
37// ServerFeatures asks the server for its features. It returns nil when it
38// cannot be reached: the optional results are then not sent.
39type ServerFeatures func(ctx context.Context) ([]string, error)
40
41// jobFeatures is what the server accepts for this job.
42type jobFeatures struct{ annotations, summaries bool }
43
44func featuresOf(list []string) jobFeatures {
45 var f jobFeatures
46 for _, x := range list {
47 switch x {
48 case FeatureAnnotations:
49 f.annotations = true
50 case FeatureSummaries:
51 f.summaries = true
52 }
53 }
54 return f
55}
56
57// loadFeatures reads the features of the server for this job (GetVersion).
58func (j *jobRun) loadFeatures(ctx context.Context, rt *Runner) jobFeatures {
59 if rt.Features == nil {
60 return jobFeatures{}
61 }
62 cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
63 defer cancel()
64 list, err := rt.Features(cctx)
65 if err != nil {
66 j.log.Warn().Err(err).Msg("GetVersion failed: annotations and summaries are not sent for this job")
67 return jobFeatures{}
68 }
69 return featuresOf(list)
70}
71
72func levelOf(l execpb.AnnotationLevel) ermite.AnnotationLevel {
73 switch l {
74 case execpb.AnnotationLevel_ANNOTATION_LEVEL_NOTICE:
75 return ermite.AnnotationLevel_ANNOTATION_LEVEL_NOTICE
76 case execpb.AnnotationLevel_ANNOTATION_LEVEL_WARNING:
77 return ermite.AnnotationLevel_ANNOTATION_LEVEL_WARNING
78 case execpb.AnnotationLevel_ANNOTATION_LEVEL_FAILURE:
79 return ermite.AnnotationLevel_ANNOTATION_LEVEL_FAILURE
80 }
81 return ermite.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED
82}
83
84// cutRunes bounds s to n characters (code points), after making it valid UTF-8.
85// A string that is cut is copied: a substring would keep the whole allocation
86// of what the guest sent (up to a frame of 2 MiB) alive for as long as the
87// short string is, and a guest can open 1,000 steps.
88func cutRunes(s string, n int) string {
89 s = strings.ToValidUTF8(s, "\ufffd")
90 if utf8.RuneCountInString(s) <= n {
91 return s
92 }
93 i := 0
94 for k := range s {
95 if i == n {
96 return strings.Clone(s[:k])
97 }
98 i++
99 }
100 return s
101}
102
103// annotationsFrom turns the annotations of a step into the ones of the
104// contract: the ones that break a rule of the contract are dropped (no level,
105// no message, a message over 4096 characters, more than ten of a level), the
106// numbers are made consistent (negative values and an end before the start
107// become "not given").
108func annotationsFrom(step int32, in []*execpb.Annotation) []*ermite.Annotation {
109 var out []*ermite.Annotation
110 count := map[ermite.AnnotationLevel]int{}
111 for _, a := range in {
112 lvl := levelOf(a.GetLevel())
113 msg := strings.ToValidUTF8(a.GetMessage(), "\ufffd")
114 if lvl == ermite.AnnotationLevel_ANNOTATION_LEVEL_UNSPECIFIED || msg == "" || utf8.RuneCountInString(msg) > maxAnnotationMessage {
115 continue
116 }
117 if count[lvl] >= maxAnnotationsPerLevel {
118 continue
119 }
120 count[lvl]++
121 n := func(v int32) int32 { return max(v, 0) }
122 o := &ermite.Annotation{
123 Level: lvl, Title: cutRunes(a.GetTitle(), maxAnnotationTitle), Message: msg, StepNumber: step,
124 StartLine: n(a.GetStartLine()), EndLine: n(a.GetEndLine()), StartColumn: n(a.GetStartColumn()), EndColumn: n(a.GetEndColumn()),
125 }
126 if f := strings.ToValidUTF8(a.GetFile(), "\ufffd"); utf8.RuneCountInString(f) <= maxAnnotationFile {
127 o.File = f
128 }
129 if o.EndLine != 0 && o.EndLine < o.StartLine {
130 o.EndLine = 0
131 }
132 out = append(out, o)
133 }
134 return out
135}
136
137// stepAnnotations records the annotations of a step; they are sent with its
138// completion (UpdateJob replaces the annotations of the step by this list).
139func (j *jobRun) stepAnnotations(a *execpb.StepAnnotations) {
140 j.mu.Lock()
141 defer j.mu.Unlock()
142 if j.violation != "" || !j.features.annotations {
143 return
144 }
145 if why := j.resultFrameProblemLocked("annotations", a.Step, j.annSeen); why != "" {
146 j.mu.Unlock()
147 j.violate("%s", why)
148 j.mu.Lock()
149 return
150 }
151 if j.annLists == nil {
152 j.annLists = map[int32][]*ermite.Annotation{}
153 }
154 j.annSeen[a.Step] = true
155 list := annotationsFrom(a.Step, a.Annotations)
156 // The list stays in memory until the call that carries it is answered,
157 // first here while its step runs, then in the outbox: all of it counts in
158 // the one budget of the outbox, from now (ReserveAnnotations) to the answer
159 // (sendStep). When there is no room (the server has been failing for long)
160 // the list is dropped, and the log says so.
161 n := annotationBytes(list)
162 if !j.outbox.ReserveAnnotations(n) {
163 j.log.Warn().Int32("step", a.Step).Int("bytes", n).Msg("too much is waiting to be sent to the server: the annotations of this step are dropped")
164 // Only a count is kept (j.mu is held): the final report tells the user.
165 j.annDropped += len(list)
166 return
167 }
168 j.annLists[a.Step] = list
169}
170
171// annotationsBytes is the size of the lists of a call.
172func annotationsBytes(m map[int32][]*ermite.Annotation) int {
173 n := 0
174 for _, l := range m {
175 n += annotationBytes(l)
176 }
177 return n
178}
179
180func annotationBytes(l []*ermite.Annotation) int {
181 n := 0
182 for _, a := range l {
183 n += len(a.Title) + len(a.Message) + len(a.File) + 64
184 }
185 return n
186}
187
188// resultFrameProblemLocked checks a frame that carries a result of a step
189// (annotations, summary). The executor sends each at most once per step, while
190// the step runs, just before it completes (the runner reads GITHUB_STEP_SUMMARY
191// once, after the step, and keeps the issues of a step in one list): anything
192// else is the guest misbehaving, and would let it queue unbounded data.
193func (j *jobRun) resultFrameProblemLocked(what string, step int32, seen map[int32]bool) string {
194 u, ok := j.steps[step]
195 switch {
196 case !ok:
197 return fmt.Sprintf("%s for step %d, which has not started", what, step)
198 case u.State == ermite.StepState_STEP_STATE_COMPLETED:
199 return fmt.Sprintf("%s for step %d, which is completed", what, step)
200 case seen[step]:
201 return fmt.Sprintf("%s for step %d, which already sent them", what, step)
202 }
203 return ""
204}
205
206// takeAnnotations returns the list recorded for step n, once.
207func (j *jobRun) takeAnnotations(n int32) map[int32][]*ermite.Annotation {
208 l, ok := j.annLists[n]
209 if !ok {
210 return nil
211 }
212 delete(j.annLists, n)
213 return map[int32][]*ermite.Annotation{n: l}
214}
215
216// stepSummary queues UploadSummary for a step that wrote GITHUB_STEP_SUMMARY.
217// The bytes count in the bounded memory of the outbox, like log lines: the call
218// blocks while too much is queued.
219func (j *jobRun) stepSummary(s *execpb.StepSummary) {
220 j.mu.Lock()
221 if j.violation != "" || !j.features.summaries {
222 j.mu.Unlock()
223 return
224 }
225 if why := j.resultFrameProblemLocked("a summary", s.Step, j.sumSeen); why != "" {
226 j.mu.Unlock()
227 j.violate("%s", why)
228 return
229 }
230 j.sumSeen[s.Step] = true
231 j.mu.Unlock()
232 md := strings.ToValidUTF8(s.Markdown, "\ufffd")
233 if len(md) > maxSummaryBytes {
234 j.log.Warn().Int32("step", s.Step).Int("bytes", len(md)).Msg("a step summary over 1 MiB is dropped")
235 return
236 }
237 j.outbox.Summary(s.Step, md)
238}
239
240// uploadSummaryClient is what the outbox needs for summaries.
241var _ interface {
242 UploadSummary(context.Context, *connect.Request[ermite.UploadSummaryRequest]) (*connect.Response[ermite.UploadSummaryResponse], error)
243} = jobClient(nil)
244
245// allPendingAnnotations are the annotations for CompleteJob: the lists no
246// UpdateJob has carried, and those of steps that never completed (the
247// executor was stopped), in step order.
248func (j *jobRun) allPendingAnnotations() []*ermite.Annotation {
249 m := map[int32][]*ermite.Annotation{}
250 for _, a := range j.outbox.PendingAnnotations() {
251 m[a.StepNumber] = append(m[a.StepNumber], a)
252 }
253 j.mu.Lock()
254 for n, l := range j.annLists {
255 m[n] = l
256 }
257 j.mu.Unlock()
258 return flattenAnnotations(m)
259}
260
261// droppedAnnotationsLine is the warning the user gets when annotations were
262// dropped because the backlog of reports to the server was full. The text is
263// ermite's own (the official runner has no such backlog); it carries no
264// annotation content, only the count, which the host coalesces over the whole
265// job. Empty when nothing was dropped.
266func droppedAnnotationsLine(n int) string {
267 switch {
268 case n <= 0:
269 return ""
270 case n == 1:
271 return "##[warning]1 annotation was not reported because the reporting backlog was full."
272 }
273 return fmt.Sprintf("##[warning]%d annotations were not reported because the reporting backlog was full.", n)
274}
275
276// reportDroppedAnnotations appends the one warning line to the job log, after
277// the outbox is closed (the queue is full, or gone: it is not used) and before
278// CompleteJob, with a slice of the report budget of its own so that a server
279// that keeps failing cannot take the time CompleteJob needs.
280func (j *jobRun) reportDroppedAnnotations(ctx context.Context, budget time.Duration) {
281 j.mu.Lock()
282 line := droppedAnnotationsLine(j.annDropped)
283 j.mu.Unlock()
284 if line == "" {
285 return
286 }
287 actx, cancel := context.WithTimeout(ctx, budget)
288 defer cancel()
289 j.appendLogLine(actx, line)
290}