Monorepo for Tangled
appview/pipelines: separate utilities out of web handler
Signed-off-by: oppiliappan <me@oppi.li>
authored by
oppiliappan
and committed by
Tangled
(May 26, 2026, 9:08 AM +0300)
12ebd851
0cb049f1
+50
-45
+44
appview/pipelines/logs.go
+44
appview/pipelines/logs.go
···
1
+
package pipelines
2
+
3
+
import (
4
+
"strings"
5
+
6
+
"github.com/gorilla/websocket"
7
+
)
8
+
9
+
type LogEvent struct {
10
+
Msg []byte
11
+
Err error
12
+
}
13
+
14
+
func (ev *LogEvent) IsCloseError() bool {
15
+
return websocket.IsCloseError(
16
+
ev.Err,
17
+
websocket.CloseNormalClosure,
18
+
websocket.CloseGoingAway,
19
+
websocket.CloseAbnormalClosure,
20
+
)
21
+
}
22
+
23
+
func ReadLogs(conn *websocket.Conn, ch chan LogEvent) {
24
+
defer close(ch)
25
+
for {
26
+
if conn == nil {
27
+
return
28
+
}
29
+
_, msg, err := conn.ReadMessage()
30
+
if err != nil {
31
+
ch <- LogEvent{Err: err}
32
+
return
33
+
}
34
+
ch <- LogEvent{Msg: msg}
35
+
}
36
+
}
37
+
38
+
func SpindleURL(dev bool, spindle, knot, rkey, workflow string) string {
39
+
scheme := "wss"
40
+
if dev {
41
+
scheme = "ws"
42
+
}
43
+
return scheme + "://" + strings.Join([]string{spindle, "logs", knot, rkey, workflow}, "/")
44
+
}
+6
-45
appview/pipelines/pipelines.go
+6
-45
appview/pipelines/pipelines.go
···
7
7
"fmt"
8
8
"log/slog"
9
9
"net/http"
10
-
"strings"
11
10
"time"
12
11
13
12
"tangled.org/core/api/tangled"
···
223
222
return
224
223
}
225
224
226
-
scheme := "wss"
227
-
if p.config.Core.Dev {
228
-
scheme = "ws"
229
-
}
230
-
231
-
url := scheme + "://" + strings.Join([]string{spindle, "logs", knot, rkey, workflow}, "/")
225
+
url := SpindleURL(p.config.Core.Dev, spindle, knot, rkey, workflow)
232
226
l = l.With("url", url)
233
227
234
228
clientConn, err := upgrader.Upgrade(w, r, nil)
···
258
252
defer spindleConn.Close()
259
253
260
254
// create a channel for incoming messages
261
-
evChan := make(chan logEvent, 100)
255
+
evChan := make(chan LogEvent, 100)
262
256
// start a goroutine to read from spindle
263
-
go readLogs(spindleConn, evChan)
257
+
go ReadLogs(spindleConn, evChan)
264
258
265
259
stepStartTimes := make(map[int]time.Time)
266
260
var fragment bytes.Buffer
···
275
269
continue
276
270
}
277
271
278
-
if ev.err != nil && ev.isCloseError() {
272
+
if ev.Err != nil && ev.IsCloseError() {
279
273
l.Debug("graceful shutdown, tail complete", "err", err)
280
274
return
281
275
}
282
-
if ev.err != nil {
276
+
if ev.Err != nil {
283
277
l.Error("error reading from spindle", "err", err)
284
278
return
285
279
}
286
280
287
281
var logLine spindlemodel.LogLine
288
-
if err = json.Unmarshal(ev.msg, &logLine); err != nil {
282
+
if err = json.Unmarshal(ev.Msg, &logLine); err != nil {
289
283
l.Error("failed to parse logline", "err", err)
290
284
continue
291
285
}
···
447
441
}
448
442
l.Debug("canceled pipeline", "uri", pipeline.AtUri())
449
443
}
450
-
451
-
// either a message or an error
452
-
type logEvent struct {
453
-
msg []byte
454
-
err error
455
-
}
456
-
457
-
func (ev *logEvent) isCloseError() bool {
458
-
return websocket.IsCloseError(
459
-
ev.err,
460
-
websocket.CloseNormalClosure,
461
-
websocket.CloseGoingAway,
462
-
websocket.CloseAbnormalClosure,
463
-
)
464
-
}
465
-
466
-
// read logs from spindle and pass through to chan
467
-
func readLogs(conn *websocket.Conn, ch chan logEvent) {
468
-
defer close(ch)
469
-
470
-
for {
471
-
if conn == nil {
472
-
return
473
-
}
474
-
475
-
_, msg, err := conn.ReadMessage()
476
-
if err != nil {
477
-
ch <- logEvent{err: err}
478
-
return
479
-
}
480
-
ch <- logEvent{msg: msg}
481
-
}
482
-
}