+20

appview/pages/pages.go

··· 974 974 return p.executeRepo("repo/pipelines/pipelines", w, params) 975 975 } 976 976 977 977 + type LogBlockParams struct { 978 978 + Id int 979 979 + Name string 980 980 + Command string 981 981 + Collapsed bool 982 982 + } 983 983 + 984 984 + func (p *Pages) LogBlock(w io.Writer, params LogBlockParams) error { 985 985 + return p.executePlain("repo/pipelines/fragments/logBlock", w, params) 986 986 + } 987 987 + 988 988 + type LogLineParams struct { 989 989 + Id int 990 990 + Content string 991 991 + } 992 992 + 993 993 + func (p *Pages) LogLine(w io.Writer, params LogLineParams) error { 994 994 + return p.executePlain("repo/pipelines/fragments/logLine", w, params) 995 995 + } 996 996 + 977 997 type WorkflowParams struct { 978 998 LoggedInUser *oauth.User 979 999 RepoInfo repoinfo.RepoInfo

+16

appview/pages/templates/repo/pipelines/fragments/logBlock.html

··· 1 1 + {{ define "repo/pipelines/fragments/logBlock" }} 2 2 + <div id="lines" hx-swap-oob="beforeend"> 3 3 + <details id="step-{{ .Id }}" {{if not .Collapsed}}open{{end}} class="group bg-gray-100 px-2 dark:bg-gray-900"> 4 4 + <summary class="sticky top-0 py-1 list-none cursor-pointer py-2 bg-gray-100 dark:bg-gray-900 hover:text-gray-500 hover:dark:text-gray-400"> 5 5 + <div class="group-open:hidden flex items-center gap-1"> 6 6 + {{ i "chevron-right" "w-4 h-4" }} {{ .Name }} 7 7 + </div> 8 8 + <div class="hidden group-open:flex items-center gap-1"> 9 9 + {{ i "chevron-down" "w-4 h-4" }} {{ .Name }} 10 10 + </div> 11 11 + </summary> 12 12 + <div class="text-blue-600 dark:text-blue-300 font-mono">{{ .Command }}</div> 13 13 + <div id="step-body-{{ .Id }}" class="font-mono"></div> 14 14 + </details> 15 15 + </div> 16 16 + {{ end }}

+6

appview/pages/templates/repo/pipelines/fragments/logLine.html

··· 1 1 + {{ define "repo/pipelines/fragments/logLine" }} 2 2 + <div id="step-body-{{ .Id }}" hx-swap-oob="beforeend"> 3 3 + <p>{{ .Content }}</p> 4 4 + </div> 5 5 + {{ end }} 6 6 +

+5 -13

appview/pages/templates/repo/pipelines/pipelines.html

··· 8 8 9 9 {{ define "repoContent" }} 10 10 <div class="flex justify-between items-center gap-4"> 11 11 - <div class="flex gap-4"> 12 12 - </div> 13 13 - 14 14 - </div> 15 15 - <div class="error" id="issues"></div> 16 16 - {{ end }} 17 17 - 18 18 - {{ define "repoAfter" }} 19 19 - <section 20 20 - class="w-full flex flex-col gap-2 mt-2" 21 21 - > 11 11 + <div class="w-full flex flex-col gap-2"> 22 12 {{ range .Pipelines }} 23 13 {{ block "pipeline" (list $ .) }} {{ end }} 24 14 {{ else }} ··· 26 16 No pipelines run for this repository. 27 17 </p> 28 18 {{ end }} 29 29 - </section> 19 19 + </div> 20 20 + </div> 30 21 {{ end }} 22 22 + 31 23 32 24 {{ define "pipeline" }} 33 25 {{ $root := index . 0 }} 34 26 {{ $p := index . 1 }} 35 35 - <div class="py-4 px-6 bg-white dark:bg-gray-800 dark:text-white"> 27 27 + <div class="py-2 bg-white dark:bg-gray-800 dark:text-white"> 36 28 {{ block "pipelineHeader" $ }} {{ end }} 37 29 </div> 38 30 {{ end }}

+3 -5

appview/pages/templates/repo/pipelines/workflow.html

··· 24 24 {{ $active := .Workflow }} 25 25 {{ with .Pipeline }} 26 26 {{ $id := .Id }} 27 27 - <div class="grid grid-cols-1 rounded border border-gray-200 dark:border-gray-700 divide-y divide-gray-200 dark:divide-gray-700"> 27 27 + <div class="sticky top-2 grid grid-cols-1 rounded border border-gray-200 dark:border-gray-700 divide-y divide-gray-200 dark:divide-gray-700"> 28 28 {{ range $name, $all := .Statuses }} 29 29 <a href="/{{ $.RepoInfo.FullName }}/pipelines/{{ $id }}/workflow/{{ $name }}" class="no-underline hover:no-underline hover:bg-gray-100/25 hover:dark:bg-gray-700/25"> 30 30 <div ··· 58 58 59 59 {{ define "logs" }} 60 60 <div id="log-stream" 61 61 - class="p-2 bg-gray-100 dark:bg-gray-900 font-mono text-sm min-h-96 max-h-screen overflow-auto flex flex-col-reverse [overflow-anchor:auto_!important]" 61 61 + class="text-sm" 62 62 hx-ext="ws" 63 63 ws-connect="/{{ $.RepoInfo.FullName }}/pipelines/{{ .Pipeline.Id }}/workflow/{{ .Workflow }}/logs"> 64 64 - <div id="lines"> 65 65 - <!-- Each log line should be rendered with class="item" like below --> 66 66 - <!-- <div class="item">[INFO] Log line here</div> --> 64 64 + <div id="lines" class="flex flex-col gap-2"> 67 65 </div> 68 66 </div> 69 67 {{ end }}

+72 -58

appview/pipelines/pipelines.go

··· 1 1 package pipelines 2 2 3 3 import ( 4 4 + "bytes" 4 5 "context" 5 6 "encoding/json" 6 6 - "fmt" 7 7 - "html" 8 7 "log/slog" 9 8 "net/http" 10 9 "strings" ··· 170 169 171 170 ctx, cancel := context.WithCancel(r.Context()) 172 171 defer cancel() 173 173 - go func() { 174 174 - for { 175 175 - if _, _, err := clientConn.NextReader(); err != nil { 176 176 - l.Error("failed to read", "err", err) 177 177 - cancel() 178 178 - return 179 179 - } 180 180 - } 181 181 - }() 182 172 183 173 user := p.oauth.GetUser(r) 184 174 f, err := p.repoResolver.Resolve(r) ··· 238 228 defer spindleConn.Close() 239 229 240 230 // create a channel for incoming messages 241 241 - msgChan := make(chan []byte, 10) 242 242 - errChan := make(chan error, 1) 243 243 - 231 231 + evChan := make(chan logEvent, 100) 244 232 // start a goroutine to read from spindle 245 245 - go func() { 246 246 - defer close(msgChan) 247 247 - defer close(errChan) 248 248 - 249 249 - for { 250 250 - _, msg, err := spindleConn.ReadMessage() 251 251 - if err != nil { 252 252 - if websocket.IsCloseError(err, 253 253 - websocket.CloseNormalClosure, 254 254 - websocket.CloseGoingAway, 255 255 - websocket.CloseAbnormalClosure) { 256 256 - errChan <- nil // signal graceful end 257 257 - } else { 258 258 - errChan <- err 259 259 - } 260 260 - return 261 261 - } 262 262 - msgChan <- msg 263 263 - } 264 264 - }() 233 233 + go readLogs(spindleConn, evChan) 265 234 266 235 stepIdx := 0 236 236 + var fragment bytes.Buffer 267 237 for { 268 238 select { 269 239 case <-ctx.Done(): 270 240 l.Info("client disconnected") 271 241 return 272 272 - case err := <-errChan: 273 273 - if err != nil { 274 274 - l.Error("error reading from spindle", "err", err) 242 242 + 243 243 + case ev, ok := <-evChan: 244 244 + if !ok { 245 245 + continue 275 246 } 276 247 277 277 - if err == nil { 278 278 - l.Info("log tail complete") 248 248 + if ev.err != nil && ev.isCloseError() { 249 249 + l.Debug("graceful shutdown, tail complete", "err", err) 250 250 + return 251 251 + } 252 252 + if ev.err != nil { 253 253 + l.Error("error reading from spindle", "err", err) 254 254 + return 279 255 } 280 256 281 281 - return 282 282 - case msg := <-msgChan: 283 257 var logLine spindlemodel.LogLine 284 284 - if err = json.Unmarshal(msg, &logLine); err != nil { 258 258 + if err = json.Unmarshal(ev.msg, &logLine); err != nil { 285 259 l.Error("failed to parse logline", "err", err) 286 260 continue 287 261 } 288 262 289 289 - var fragment []byte 263 263 + fragment.Reset() 264 264 + 290 265 switch logLine.Kind { 291 266 case spindlemodel.LogKindControl: 292 267 // control messages create a new step block 293 268 stepIdx++ 294 294 - fragment = fmt.Appendf(nil, ` 295 295 - <div id="lines" hx-swap-oob="beforeend"> 296 296 - <details id="step-%d" open> 297 297 - <summary>%s</summary> 298 298 - <div id="step-body-%d"></div> 299 299 - </details> 300 300 - </div> 301 301 - `, stepIdx, logLine.Content, stepIdx) 269 269 + collapsed := false 270 270 + if logLine.StepKind == spindlemodel.StepKindSystem { 271 271 + collapsed = true 272 272 + } 273 273 + err = p.pages.LogBlock(&fragment, pages.LogBlockParams{ 274 274 + Id: stepIdx, 275 275 + Name: logLine.Content, 276 276 + Command: logLine.StepCommand, 277 277 + Collapsed: collapsed, 278 278 + }) 302 279 case spindlemodel.LogKindData: 303 280 // data messages simply insert new log lines into current step 304 304 - escaped := html.EscapeString(logLine.Content) 305 305 - fragment = fmt.Appendf(nil, ` 306 306 - <div id="step-body-%d" hx-swap-oob="beforeend"> 307 307 - <p>%s</p> 308 308 - </div> 309 309 - `, stepIdx, escaped) 281 281 + err = p.pages.LogLine(&fragment, pages.LogLineParams{ 282 282 + Id: stepIdx, 283 283 + Content: logLine.Content, 284 284 + }) 285 285 + } 286 286 + if err != nil { 287 287 + l.Error("failed to render log line", "err", err) 288 288 + return 310 289 } 311 290 312 312 - if err = clientConn.WriteMessage(websocket.TextMessage, fragment); err != nil { 291 291 + if err = clientConn.WriteMessage(websocket.TextMessage, fragment.Bytes()); err != nil { 313 292 l.Error("error writing to client", "err", err) 314 293 return 315 294 } 295 295 + 316 296 case <-time.After(30 * time.Second): 317 297 l.Debug("sent keepalive") 318 298 if err = clientConn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil { 319 299 l.Error("failed to write control", "err", err) 300 300 + return 320 301 } 321 302 } 322 303 } 323 304 } 305 305 + 306 306 + // either a message or an error 307 307 + type logEvent struct { 308 308 + msg []byte 309 309 + err error 310 310 + } 311 311 + 312 312 + func (ev *logEvent) isCloseError() bool { 313 313 + return websocket.IsCloseError( 314 314 + ev.err, 315 315 + websocket.CloseNormalClosure, 316 316 + websocket.CloseGoingAway, 317 317 + websocket.CloseAbnormalClosure, 318 318 + ) 319 319 + } 320 320 + 321 321 + // read logs from spindle and pass through to chan 322 322 + func readLogs(conn *websocket.Conn, ch chan logEvent) { 323 323 + defer close(ch) 324 324 + 325 325 + for { 326 326 + if conn == nil { 327 327 + return 328 328 + } 329 329 + 330 330 + _, msg, err := conn.ReadMessage() 331 331 + if err != nil { 332 332 + ch <- logEvent{err: err} 333 333 + return 334 334 + } 335 335 + ch <- logEvent{msg: msg} 336 336 + } 337 337 + }

+2 -2

spindle/engine/engine.go

··· 315 315 } 316 316 defer wfLogger.Close() 317 317 318 318 - ctl := wfLogger.ControlWriter() 319 319 - ctl.Write([]byte(step.Command)) 318 318 + ctl := wfLogger.ControlWriter(stepIdx, step) 319 319 + ctl.Write([]byte(step.Name)) 320 320 321 321 logs, err := e.docker.ContainerLogs(ctx, containerID, container.LogsOptions{ 322 322 Follow: true,

+27 -12

spindle/engine/logger.go

··· 41 41 42 42 func (l *WorkflowLogger) DataWriter(stream string) io.Writer { 43 43 // TODO: emit stream 44 44 - return &jsonWriter{logger: l, kind: models.LogKindData} 44 44 + return &dataWriter{ 45 45 + logger: l, 46 46 + stream: stream, 47 47 + } 45 48 } 46 49 47 47 - func (l *WorkflowLogger) ControlWriter() io.Writer { 48 48 - return &jsonWriter{logger: l, kind: models.LogKindControl} 50 50 + func (l *WorkflowLogger) ControlWriter(idx int, step models.Step) io.Writer { 51 51 + return &controlWriter{ 52 52 + logger: l, 53 53 + idx: idx, 54 54 + step: step, 55 55 + } 49 56 } 50 57 51 51 - type jsonWriter struct { 58 58 + type dataWriter struct { 52 59 logger *WorkflowLogger 53 53 - kind models.LogKind 60 60 + stream string 54 61 } 55 62 56 56 - func (w *jsonWriter) Write(p []byte) (int, error) { 63 63 + func (w *dataWriter) Write(p []byte) (int, error) { 57 64 line := strings.TrimRight(string(p), "\r\n") 65 65 + entry := models.NewDataLogLine(line, w.stream) 66 66 + if err := w.logger.encoder.Encode(entry); err != nil { 67 67 + return 0, err 68 68 + } 69 69 + return len(p), nil 70 70 + } 58 71 59 59 - entry := models.LogLine{ 60 60 - Kind: w.kind, 61 61 - Content: line, 62 62 - } 72 72 + type controlWriter struct { 73 73 + logger *WorkflowLogger 74 74 + idx int 75 75 + step models.Step 76 76 + } 63 77 78 78 + func (w *controlWriter) Write(_ []byte) (int, error) { 79 79 + entry := models.NewControlLogLine(w.idx, w.step) 64 80 if err := w.logger.encoder.Encode(entry); err != nil { 65 81 return 0, err 66 82 } 67 67 - 68 68 - return len(p), nil 83 83 + return len(w.step.Name), nil 69 84 }

+21

spindle/models/models.go

··· 88 88 Stream string `json:"stream,omitempty"` 89 89 90 90 // fields if kind is "control" 91 91 + StepId int `json:"step_id,omitempty"` 92 92 + StepKind StepKind `json:"step_kind,omitempty"` 93 93 + StepCommand string `json:"step_command,omitempty"` 94 94 + } 95 95 + 96 96 + func NewDataLogLine(content, stream string) LogLine { 97 97 + return LogLine{ 98 98 + Kind: LogKindData, 99 99 + Content: content, 100 100 + Stream: stream, 101 101 + } 102 102 + } 103 103 + 104 104 + func NewControlLogLine(idx int, step Step) LogLine { 105 105 + return LogLine{ 106 106 + Kind: LogKindControl, 107 107 + Content: step.Name, 108 108 + StepId: idx, 109 109 + StepKind: step.Kind, 110 110 + StepCommand: step.Command, 111 111 + } 91 112 }

+15 -1

spindle/models/pipeline.go

··· 15 15 Command string 16 16 Name string 17 17 Environment map[string]string 18 18 + Kind StepKind 18 19 } 20 20 + 21 21 + type StepKind int 22 22 + 23 23 + const ( 24 24 + // steps injected by the CI runner 25 25 + StepKindSystem StepKind = iota 26 26 + // steps defined by the user in the original pipeline 27 27 + StepKindUser 28 28 + ) 19 29 20 30 type Workflow struct { 21 31 Steps []Step ··· 46 56 sstep.Environment = stepEnvToMap(tstep.Environment) 47 57 sstep.Command = tstep.Command 48 58 sstep.Name = tstep.Name 59 59 + sstep.Kind = StepKindUser 49 60 swf.Steps = append(swf.Steps, sstep) 50 61 } 51 62 swf.Name = twf.Name ··· 59 70 setup.addStep(nixConfStep()) 60 71 setup.addStep(cloneStep(*twf, *pl.TriggerMetadata.Repo, cfg.Server.Dev)) 61 72 setup.addStep(checkoutStep(*twf, *pl.TriggerMetadata)) 62 62 - setup.addStep(dependencyStep(*twf)) 73 73 + // this step could be empty 74 74 + if s := dependencyStep(*twf); s != nil { 75 75 + setup.addStep(*s) 76 76 + } 63 77 64 78 // append setup steps in order to the start of workflow steps 65 79 swf.Steps = append(*setup, swf.Steps...)

+3 -3

spindle/models/setup_steps.go

··· 83 83 // For dependencies using a custom registry (i.e. not nixpkgs), it collects 84 84 // all packages and adds a single 'nix profile install' step to the 85 85 // beginning of the workflow's step list. 86 86 - func dependencyStep(twf tangled.Pipeline_Workflow) Step { 86 86 + func dependencyStep(twf tangled.Pipeline_Workflow) *Step { 87 87 var customPackages []string 88 88 89 89 for _, d := range twf.Dependencies { ··· 111 111 "NIX_SHOW_DOWNLOAD_PROGRESS": "0", 112 112 }, 113 113 } 114 114 - return installStep 114 114 + return &installStep 115 115 } 116 116 - return Step{} 116 116 + return nil 117 117 }