spindle/xrpc: implement ci.pipeline.subscribeLogs · matthiasplappert.tngl.sh/tangled-core@67f4c80

+1

flake.nix

··· 374 374 pkgs.iproute2 375 375 ]; 376 376 shellHook = '' 377 + export CC=${pkgs.stdenv.cc}/bin/cc 377 378 mkdir -p appview/pages/static 378 379 # temporary self-heal for workspaces that copied static assets as read-only 379 380 [ -d appview/pages/static/icons ] && [ ! -w appview/pages/static/icons ] && chmod -R u+rwX appview/pages/static

+3 -1

lexutil/client.go

··· 44 44 func (c *Client) LexDo(ctx context.Context, method string, inputEncoding string, endpoint string, params map[string]any, bodyData any, out any) error { 45 45 switch method { 46 46 case Subscription: 47 - if process, ok := out.(processFn); ok { 47 + if process, ok := out.(func(context.Context, *cbg.CborReader) error); ok { 48 + return c.LexSubscribe(ctx, endpoint, params, process) 49 + } else if process, ok := out.(processFn); ok { 48 50 return c.LexSubscribe(ctx, endpoint, params, process) 49 51 } else if redialer, ok := out.(Redialer); ok { 50 52 return c.LexSubscribeWithRedialer(ctx, endpoint, params, redialer)

-30

spindle/stream.go

··· 2 2 3 3 import ( 4 4 "context" 5 - "encoding/json" 6 5 "errors" 7 6 "fmt" 8 7 "io" 9 8 "net/http" 10 - "os" 11 9 "time" 12 10 13 11 "tangled.org/core/eventstream" ··· 93 91 94 92 filePath := models.LogFilePath(s.cfg.Server.LogDir, wid) 95 93 96 - if status.Status == models.StatusKindFailed.String() && status.Error != nil { 97 - if _, err := os.Stat(filePath); os.IsNotExist(err) { 98 - msgs := []models.LogLine{ 99 - { 100 - Kind: models.LogKindControl, 101 - Content: "", 102 - StepId: 0, 103 - StepKind: models.StepKindUser, 104 - }, 105 - { 106 - Kind: models.LogKindData, 107 - Content: *status.Error, 108 - }, 109 - } 110 94 111 - for _, msg := range msgs { 112 - b, err := json.Marshal(msg) 113 - if err != nil { 114 - return err 115 - } 116 - 117 - if err := conn.WriteMessage(websocket.TextMessage, b); err != nil { 118 - return fmt.Errorf("failed to write to websocket: %w", err) 119 - } 120 - } 121 - 122 - return nil 123 - } 124 - } 125 95 126 96 config := tail.Config{ 127 97 Follow: !isFinished,

+322

spindle/xrpc/ci_pipeline_subscribe_logs.go

··· 1 + package xrpc 2 + 3 + import ( 4 + "context" 5 + "encoding/json" 6 + "fmt" 7 + "io" 8 + "net/http" 9 + "sync" 10 + "time" 11 + 12 + "github.com/bluesky-social/indigo/atproto/atclient" 13 + "github.com/bluesky-social/indigo/atproto/syntax" 14 + "github.com/gorilla/websocket" 15 + "github.com/hpcloud/tail" 16 + "tangled.org/core/api/tangled" 17 + "tangled.org/core/spindle/models" 18 + ) 19 + 20 + func (x *Xrpc) HandleCiPipelineSubscribeLogs(w http.ResponseWriter, r *http.Request) { 21 + var ( 22 + pipelineQuery = r.URL.Query().Get("pipeline") 23 + workflows = r.URL.Query()["workflows"] 24 + ) 25 + 26 + pipeline, err := syntax.ParseTID(pipelineQuery) 27 + if err != nil { 28 + writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("pipeline parameter invalid: %s", pipelineQuery)}) 29 + return 30 + } 31 + 32 + x.handleSubscribeLogs(w, r, pipeline, workflows) 33 + } 34 + 35 + var wsUpgrader = websocket.Upgrader{ 36 + ReadBufferSize: 10_000, 37 + WriteBufferSize: 10_000, 38 + } 39 + 40 + func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipeline syntax.TID, workflows []string) { 41 + l := x.Logger.With("pipeline", pipeline, "workflows", workflows) 42 + 43 + // 1. query the event from database to get the knot 44 + var eventJson string 45 + err := x.Db.QueryRow( 46 + `select event from events where nsid = ? and rkey = ?`, 47 + tangled.PipelineNSID, 48 + pipeline.String(), 49 + ).Scan(&eventJson) 50 + if err != nil { 51 + l.Error("failed to find pipeline event", "err", err) 52 + writeJson(w, http.StatusNotFound, atclient.ErrorBody{Name: "NotFound", Message: fmt.Sprintf("pipeline not found: %s", pipeline.String())}) 53 + return 54 + } 55 + 56 + var tpl tangled.Pipeline 57 + if err := json.Unmarshal([]byte(eventJson), &tpl); err != nil { 58 + l.Error("failed to unmarshal pipeline event", "err", err) 59 + writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "failed to parse pipeline event"}) 60 + return 61 + } 62 + 63 + if tpl.TriggerMetadata == nil || tpl.TriggerMetadata.Repo == nil { 64 + l.Error("pipeline event trigger metadata is incomplete") 65 + writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "pipeline event trigger metadata is incomplete"}) 66 + return 67 + } 68 + knot := tpl.TriggerMetadata.Repo.Knot 69 + 70 + // 2. if workflows is empty, default to all workflows defined in the pipeline 71 + if len(workflows) == 0 { 72 + for _, wf := range tpl.Workflows { 73 + if wf != nil && wf.Name != "" { 74 + workflows = append(workflows, wf.Name) 75 + } 76 + } 77 + } 78 + 79 + if len(workflows) == 0 { 80 + writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "no workflows specified or found"}) 81 + return 82 + } 83 + 84 + // 3. upgrade to websocket 85 + ctx, cancel := context.WithCancel(r.Context()) 86 + defer cancel() 87 + 88 + conn, err := wsUpgrader.Upgrade(w, r, w.Header()) 89 + if err != nil { 90 + l.Error("websocket upgrade failed", "err", err) 91 + return 92 + } 93 + defer conn.Close() 94 + 95 + lastWriteLk := sync.Mutex{} 96 + lastWrite := time.Now() 97 + 98 + // Ping loop 99 + go func() { 100 + ticker := time.NewTicker(30 * time.Second) 101 + defer ticker.Stop() 102 + 103 + for { 104 + select { 105 + case <-ticker.C: 106 + lastWriteLk.Lock() 107 + lw := lastWrite 108 + lastWriteLk.Unlock() 109 + 110 + if time.Since(lw) < 30*time.Second { 111 + continue 112 + } 113 + 114 + if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil { 115 + l.Warn("failed to ping client", "err", err) 116 + cancel() 117 + return 118 + } 119 + case <-ctx.Done(): 120 + return 121 + } 122 + } 123 + }() 124 + 125 + conn.SetPingHandler(func(message string) error { 126 + err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60)) 127 + if err == websocket.ErrCloseSent { 128 + return nil 129 + } 130 + return err 131 + }) 132 + 133 + // Read discard loop 134 + go func() { 135 + for { 136 + _, _, err := conn.ReadMessage() 137 + if err != nil { 138 + l.Warn("failed to read message from client", "err", err) 139 + cancel() 140 + return 141 + } 142 + } 143 + }() 144 + 145 + eventsChan := make(chan tangled.CiPipelineSubscribeLogs_Event, 128) 146 + wg := sync.WaitGroup{} 147 + 148 + // 4. start a tail reader goroutine for each workflow 149 + for _, wf := range workflows { 150 + wg.Add(1) 151 + go func(wfName string) { 152 + defer wg.Done() 153 + 154 + wid := models.WorkflowId{ 155 + PipelineId: models.PipelineId{ 156 + Knot: knot, 157 + Rkey: pipeline.String(), 158 + }, 159 + Name: wfName, 160 + } 161 + 162 + // check if finished, but poll database to know when it finishes 163 + var isFinished bool 164 + status, err := x.Db.GetStatus(wid) 165 + if err == nil { 166 + isFinished = models.StatusKind(status.Status).IsFinish() 167 + } 168 + 169 + filePath := models.LogFilePath(x.Config.Server.LogDir, wid) 170 + 171 + 172 + 173 + tailConfig := tail.Config{ 174 + Follow: !isFinished, 175 + ReOpen: !isFinished, 176 + MustExist: false, 177 + Location: &tail.SeekInfo{ 178 + Offset: 0, 179 + Whence: io.SeekStart, 180 + }, 181 + } 182 + 183 + t, err := tail.TailFile(filePath, tailConfig) 184 + if err != nil { 185 + l.Error("failed to tail log file", "workflow", wfName, "err", err) 186 + return 187 + } 188 + defer t.Stop() 189 + 190 + // if we are following, poll status in database to stop tailing when finished 191 + if !isFinished { 192 + go func() { 193 + ticker := time.NewTicker(2 * time.Second) 194 + defer ticker.Stop() 195 + for { 196 + select { 197 + case <-ctx.Done(): 198 + return 199 + case <-ticker.C: 200 + status, err := x.Db.GetStatus(wid) 201 + if err == nil && models.StatusKind(status.Status).IsFinish() { 202 + t.Stop() 203 + return 204 + } 205 + } 206 + } 207 + }() 208 + } 209 + 210 + for { 211 + select { 212 + case <-ctx.Done(): 213 + return 214 + case line, ok := <-t.Lines: 215 + if !ok || line == nil { 216 + return 217 + } 218 + 219 + if line.Err != nil { 220 + l.Warn("error tailing log file", "workflow", wfName, "err", line.Err) 221 + return 222 + } 223 + 224 + var logLine models.LogLine 225 + if err := json.Unmarshal([]byte(line.Text), &logLine); err != nil { 226 + // if it's not JSON, treat it as a raw data line 227 + logLine = models.NewDataLogLine(0, line.Text, "stdout") 228 + } 229 + 230 + var ev tangled.CiPipelineSubscribeLogs_Event 231 + timeStr := logLine.Time.Format(time.RFC3339) 232 + if logLine.Time.IsZero() { 233 + timeStr = time.Now().Format(time.RFC3339) 234 + } 235 + 236 + if logLine.Kind == models.LogKindControl { 237 + stepKindStr := "user" 238 + if logLine.StepKind == models.StepKindSystem { 239 + stepKindStr = "system" 240 + } 241 + ev = tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{ 242 + Time: timeStr, 243 + Workflow: wfName, 244 + Step: int64(logLine.StepId), 245 + Content: logLine.Content, 246 + Command: strptrOrNil(logLine.StepCommand), 247 + Status: strptrOrNil(string(logLine.StepStatus)), 248 + Kind: strptrOrNil(stepKindStr), 249 + }} 250 + } else { 251 + streamType := logLine.Stream 252 + if streamType != "stdout" && streamType != "stderr" { 253 + streamType = "stdout" 254 + } 255 + ev = tangled.CiPipelineSubscribeLogs_Event{Data: &tangled.CiPipelineSubscribeLogs_Data{ 256 + Time: timeStr, 257 + Workflow: wfName, 258 + Step: int64(logLine.StepId), 259 + Content: logLine.Content + "\n", // Append newline back since logger trims it 260 + Stream: streamType, 261 + }} 262 + } 263 + 264 + select { 265 + case eventsChan <- ev: 266 + case <-ctx.Done(): 267 + return 268 + } 269 + } 270 + } 271 + }(wf) 272 + } 273 + 274 + // Closer goroutine for eventsChan 275 + go func() { 276 + wg.Wait() 277 + close(eventsChan) 278 + }() 279 + 280 + // Main writer loop 281 + for { 282 + select { 283 + case <-ctx.Done(): 284 + return 285 + case evt, ok := <-eventsChan: 286 + if !ok { 287 + return 288 + } 289 + 290 + wc, err := conn.NextWriter(websocket.BinaryMessage) 291 + if err != nil { 292 + l.Error("failed to get next writer", "err", err) 293 + return 294 + } 295 + 296 + err = evt.Serialize(wc) 297 + if err != nil { 298 + l.Error("failed to serialize event", "err", err) 299 + wc.Close() 300 + return 301 + } 302 + 303 + if err := wc.Close(); err != nil { 304 + l.Warn("failed to flush-close event write", "err", err) 305 + return 306 + } 307 + 308 + lastWriteLk.Lock() 309 + lastWrite = time.Now() 310 + lastWriteLk.Unlock() 311 + } 312 + } 313 + } 314 + 315 + func strptr(s string) *string { return &s } 316 + 317 + func strptrOrNil(s string) *string { 318 + if s == "" { 319 + return nil 320 + } 321 + return &s 322 + }

+10

spindle/xrpc/xrpc.go

··· 48 48 49 49 // service query endpoints (no auth required) 50 50 r.Get("/"+tangled.OwnerNSID, x.Owner) 51 + r.Get("/"+tangled.CiPipelineSubscribeLogsNSID, x.HandleCiPipelineSubscribeLogs) 51 52 52 53 return r 53 54 } ··· 60 61 w.WriteHeader(status) 61 62 json.NewEncoder(w).Encode(e) 62 63 } 64 + 65 + func writeJson(w http.ResponseWriter, status int, response any) error { 66 + w.Header().Set("Content-Type", "application/json") 67 + w.WriteHeader(status) 68 + if err := json.NewEncoder(w).Encode(response); err != nil { 69 + return err 70 + } 71 + return nil 72 + }

Configure Feed

Configure Feed