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