+199

spindle/xrpc/ci_pipeline_subscribe_logs.go

Reviewed

··· 1 1 + package xrpc 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "net/http" 7 7 + "sync" 8 8 + "time" 9 9 + 10 10 + "github.com/bluesky-social/indigo/atproto/atclient" 11 11 + "github.com/bluesky-social/indigo/atproto/syntax" 12 12 + "github.com/gorilla/websocket" 13 13 + "tangled.org/core/api/tangled" 14 14 + ) 15 15 + 16 16 + func (x *Xrpc) HandleCiPipelineSubscribeLogs(w http.ResponseWriter, r *http.Request) { 17 17 + var ( 18 18 + pipelineQuery = r.URL.Query().Get("pipeline") 19 19 + workflows = r.URL.Query()["workflows"] 20 20 + ) 21 21 + 22 22 + pipeline, err := syntax.ParseTID(pipelineQuery) 23 23 + if err != nil { 24 24 + writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("pipeline parameter invalid: %s", pipelineQuery)}) 25 25 + } 26 26 + 27 27 + x.handleSubscribeLogs(w, r, pipeline, workflows) 28 28 + } 29 29 + 30 30 + var wsUpgrader = websocket.Upgrader{ 31 31 + ReadBufferSize: 10_000, 32 32 + WriteBufferSize: 10_000, 33 33 + } 34 34 + 35 35 + func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipeline syntax.TID, workflows []string) { 36 36 + l := x.Logger.With("pipeline", pipeline, "workflows", workflows) 37 37 + 38 38 + ctx, cancel := context.WithCancel(r.Context()) 39 39 + defer cancel() 40 40 + 41 41 + conn, err := wsUpgrader.Upgrade(w, r, w.Header()) 42 42 + if err != nil { 43 43 + // TODO: writeJson(w, ) 44 44 + return 45 45 + } 46 46 + defer conn.Close() 47 47 + 48 48 + lastWriteLk := sync.Mutex{} 49 49 + lastWrite := time.Now() 50 50 + 51 51 + // Start a goroutine to ping the client every 30 seconds to check if it's 52 52 + // still alive. If the client doesn't respond to a ping within 5 seconds, 53 53 + // we'll close the connection and teardown the consumer. 54 54 + go func() { 55 55 + ticker := time.NewTicker(30 * time.Second) 56 56 + defer ticker.Stop() 57 57 + 58 58 + for { 59 59 + select { 60 60 + case <-ticker.C: 61 61 + lastWriteLk.Lock() 62 62 + lw := lastWrite 63 63 + lastWriteLk.Unlock() 64 64 + 65 65 + if time.Since(lw) < 30*time.Second { 66 66 + continue 67 67 + } 68 68 + 69 69 + if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil { 70 70 + l.Warn("failed to ping client", "err", err) 71 71 + cancel() 72 72 + return 73 73 + } 74 74 + case <-ctx.Done(): 75 75 + return 76 76 + } 77 77 + } 78 78 + }() 79 79 + 80 80 + conn.SetPingHandler(func(message string) error { 81 81 + err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60)) 82 82 + if err == websocket.ErrCloseSent { 83 83 + return nil 84 84 + } 85 85 + return err 86 86 + }) 87 87 + 88 88 + // Start a goroutine to read messages from the client and discard them. 89 89 + go func() { 90 90 + for { 91 91 + _, _, err := conn.ReadMessage() 92 92 + if err != nil { 93 93 + l.Warn("failed to read message from client", "err", err) 94 94 + cancel() 95 95 + return 96 96 + } 97 97 + } 98 98 + }() 99 99 + 100 100 + evts := make(chan tangled.CiPipelineSubscribeLogs_Event, 16) 101 101 + go mockEvents(ctx, evts) 102 102 + 103 103 + for { 104 104 + select { 105 105 + case <-ctx.Done(): 106 106 + return 107 107 + case evt, ok := <-evts: 108 108 + if !ok { 109 109 + l.Error("event stream closed unexpectedly") 110 110 + return 111 111 + } 112 112 + 113 113 + wc, err := conn.NextWriter(websocket.BinaryMessage) 114 114 + if err != nil { 115 115 + l.Error("failed to get next writer", "err", err) 116 116 + return // err 117 117 + } 118 118 + 119 119 + err = evt.Serialize(wc) 120 120 + if err != nil { 121 121 + return // err 122 122 + } 123 123 + 124 124 + if err := wc.Close(); err != nil { 125 125 + l.Warn("failed to flush-close our event write", "err", err) 126 126 + return 127 127 + } 128 128 + 129 129 + lastWriteLk.Lock() 130 130 + lastWrite = time.Now() 131 131 + lastWriteLk.Unlock() 132 132 + } 133 133 + } 134 134 + } 135 135 + 136 136 + func strptr(s string) *string { return &s } 137 137 + 138 138 + // mockEvents streams a fixed, fabricated log sequence onto out, ignoring the 139 139 + // real pipeline/workflow inputs. Used to exercise the appview log viewer until 140 140 + // the real event source is wired up. 141 141 + func mockEvents(ctx context.Context, out chan<- tangled.CiPipelineSubscribeLogs_Event) { 142 142 + const wf = "build" 143 143 + now := func() string { return time.Now().UTC().Format(time.RFC3339) } 144 144 + 145 145 + send := func(ev tangled.CiPipelineSubscribeLogs_Event) bool { 146 146 + select { 147 147 + case out <- ev: 148 148 + return true 149 149 + case <-ctx.Done(): 150 150 + return false 151 151 + } 152 152 + } 153 153 + dataEv := func(step int64, content string) tangled.CiPipelineSubscribeLogs_Event { 154 154 + return tangled.CiPipelineSubscribeLogs_Event{Data: &tangled.CiPipelineSubscribeLogs_Data{ 155 155 + Step: step, Content: content, Stream: "stdout", Time: now(), Workflow: wf, 156 156 + }} 157 157 + } 158 158 + endEv := func(step int64) tangled.CiPipelineSubscribeLogs_Event { 159 159 + return tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{ 160 160 + Status: strptr("end"), Step: step, Time: now(), Workflow: wf, 161 161 + }} 162 162 + } 163 163 + 164 164 + // system step: clone (collapsed in the viewer) 165 165 + if !send(tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{ 166 166 + Status: strptr("start"), Kind: strptr("system"), Step: 1, 167 167 + Content: "clone repository", Command: strptr("git clone https://example.com/repo"), Time: now(), Workflow: wf, 168 168 + }}) { 169 169 + return 170 170 + } 171 171 + for _, line := range []string{"Cloning into 'repo'...\n", "remote: Enumerating objects\n", "\x1b[32mClone complete\x1b[0m\n"} { 172 172 + if !send(dataEv(1, line)) { 173 173 + return 174 174 + } 175 175 + } 176 176 + if !send(endEv(1)) { 177 177 + return 178 178 + } 179 179 + 180 180 + // user step: test — stream a counter line every second until the client leaves 181 181 + if !send(tangled.CiPipelineSubscribeLogs_Event{Control: &tangled.CiPipelineSubscribeLogs_Control{ 182 182 + Status: strptr("start"), Kind: strptr("user"), Step: 2, 183 183 + Content: "run tests", Command: strptr("go test ./..."), Time: now(), Workflow: wf, 184 184 + }}) { 185 185 + return 186 186 + } 187 187 + ticker := time.NewTicker(time.Second) 188 188 + defer ticker.Stop() 189 189 + for i := 1; ; i++ { 190 190 + select { 191 191 + case <-ctx.Done(): 192 192 + return 193 193 + case <-ticker.C: 194 194 + if !send(dataEv(2, fmt.Sprintf("test log line %d\n", i))) { 195 195 + return 196 196 + } 197 197 + } 198 198 + } 199 199 + }

+10

spindle/xrpc/xrpc.go

Reviewed

··· 48 48 49 49 // service query endpoints (no auth required) 50 50 r.Get("/"+tangled.OwnerNSID, x.Owner) 51 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 64 + 65 65 + func writeJson(w http.ResponseWriter, status int, response any) error { 66 66 + w.Header().Set("Content-Type", "application/json") 67 67 + w.WriteHeader(status) 68 68 + if err := json.NewEncoder(w).Encode(response); err != nil { 69 69 + return err 70 70 + } 71 71 + return nil 72 72 + }