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