Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321package xrpc
import ( "context" "encoding/json" "fmt" "io" "net/http" "sync" "time"
"github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/gorilla/websocket" "github.com/hpcloud/tail" "tangled.org/core/api/tangled" "tangled.org/core/spindle/models")
func (x *Xrpc) HandleCiSubscribePipelineLogs(w http.ResponseWriter, r *http.Request) { var ( pipelineQuery = r.URL.Query().Get("pipeline") workflows = r.URL.Query()["workflows"] )
pipeline, err := syntax.ParseTID(pipelineQuery) if err != nil { writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: fmt.Sprintf("pipeline parameter invalid: %s", pipelineQuery)}) return }
x.handleSubscribeLogs(w, r, pipeline, workflows)}
var wsUpgrader = websocket.Upgrader{ ReadBufferSize: 10_000, WriteBufferSize: 10_000,}
func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipeline syntax.TID, workflows []string) { l := x.Logger.With("pipeline", pipeline, "workflows", workflows)
// 1. query the event from database to get the knot var eventJson string err := x.Db.QueryRow( `select event from events where nsid = ? and rkey = ?`, tangled.PipelineNSID, pipeline.String(), ).Scan(&eventJson) if err != nil { l.Error("failed to find pipeline event", "err", err) writeJson(w, http.StatusNotFound, atclient.ErrorBody{Name: "NotFound", Message: fmt.Sprintf("pipeline not found: %s", pipeline.String())}) return }
var tpl tangled.Pipeline if err := json.Unmarshal([]byte(eventJson), &tpl); err != nil { l.Error("failed to unmarshal pipeline event", "err", err) writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "failed to parse pipeline event"}) return }
if tpl.TriggerMetadata == nil || tpl.TriggerMetadata.Repo == nil { l.Error("pipeline event trigger metadata is incomplete") writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "pipeline event trigger metadata is incomplete"}) return } knot := tpl.TriggerMetadata.Repo.Knot
// 2. if workflows is empty, default to all workflows defined in the pipeline if len(workflows) == 0 { for _, wf := range tpl.Workflows { if wf != nil && wf.Name != "" { workflows = append(workflows, wf.Name) } } }
if len(workflows) == 0 { writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "no workflows specified or found"}) return }
// 3. upgrade to websocket ctx, cancel := context.WithCancel(r.Context()) defer cancel()
conn, err := wsUpgrader.Upgrade(w, r, w.Header()) if err != nil { l.Error("websocket upgrade failed", "err", err) return } defer conn.Close()
lastWriteLk := sync.Mutex{} lastWrite := time.Now()
// Ping loop go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop()
for { select { case <-ticker.C: lastWriteLk.Lock() lw := lastWrite lastWriteLk.Unlock()
if time.Since(lw) < 30*time.Second { continue }
if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil { l.Warn("failed to ping client", "err", err) cancel() return } case <-ctx.Done(): return } } }()
conn.SetPingHandler(func(message string) error { err := conn.WriteControl(websocket.PongMessage, []byte(message), time.Now().Add(time.Second*60)) if err == websocket.ErrCloseSent { return nil } return err })
// Read discard loop go func() { for { _, _, err := conn.ReadMessage() if err != nil { l.Warn("failed to read message from client", "err", err) cancel() return } } }()
eventsChan := make(chan tangled.CiSubscribePipelineLogs_Event, 128) wg := sync.WaitGroup{}
// 4. start a tail reader goroutine for each workflow for _, wf := range workflows { wg.Add(1) go func(wfName string) { defer wg.Done()
wid := models.WorkflowId{ PipelineId: models.PipelineId{ Knot: knot, Rkey: pipeline.String(), }, Name: wfName, }
// check if finished, but poll database to know when it finishes var isFinished bool status, err := x.Db.GetStatus(wid) if err == nil { isFinished = models.StatusKind(status.Status).IsFinish() }
filePath := models.LogFilePath(x.Config.Server.LogDir, wid)
tailConfig := tail.Config{ Follow: !isFinished, ReOpen: !isFinished, MustExist: false, Location: &tail.SeekInfo{ Offset: 0, Whence: io.SeekStart, }, }
t, err := tail.TailFile(filePath, tailConfig) if err != nil { l.Error("failed to tail log file", "workflow", wfName, "err", err) return } defer t.Stop()
// if we are following, poll status in database to stop tailing when finished if !isFinished { go func() { ticker := time.NewTicker(2 * time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: status, err := x.Db.GetStatus(wid) if err == nil && models.StatusKind(status.Status).IsFinish() { t.Stop() return } } } }() }
for { select { case <-ctx.Done(): return case line, ok := <-t.Lines: if !ok || line == nil { return }
if line.Err != nil { l.Warn("error tailing log file", "workflow", wfName, "err", line.Err) return }
var logLine models.LogLine if err := json.Unmarshal([]byte(line.Text), &logLine); err != nil { // if it's not JSON, treat it as a raw data line logLine = models.NewDataLogLine(0, line.Text, "stdout") }
var ev tangled.CiSubscribePipelineLogs_Event timeStr := logLine.Time.Format(time.RFC3339) if logLine.Time.IsZero() { timeStr = time.Now().Format(time.RFC3339) }
if logLine.Kind == models.LogKindControl { stepKindStr := "user" if logLine.StepKind == models.StepKindSystem { stepKindStr = "system" } ev = tangled.CiSubscribePipelineLogs_Event{Control: &tangled.CiSubscribePipelineLogs_Control{ Time: timeStr, Workflow: wfName, Step: int64(logLine.StepId), Content: logLine.Content, Command: strptrOrNil(logLine.StepCommand), Status: strptrOrNil(string(logLine.StepStatus)), Kind: strptrOrNil(stepKindStr), }} } else { streamType := logLine.Stream if streamType != "stdout" && streamType != "stderr" { streamType = "stdout" } ev = tangled.CiSubscribePipelineLogs_Event{Data: &tangled.CiSubscribePipelineLogs_Data{ Time: timeStr, Workflow: wfName, Step: int64(logLine.StepId), Content: logLine.Content + "\n", // Append newline back since logger trims it Stream: streamType, }} }
select { case eventsChan <- ev: case <-ctx.Done(): return } } } }(wf) }
// Closer goroutine for eventsChan go func() { wg.Wait() close(eventsChan) }()
// Main writer loop for { select { case <-ctx.Done(): return case evt, ok := <-eventsChan: if !ok { return }
wc, err := conn.NextWriter(websocket.BinaryMessage) if err != nil { l.Error("failed to get next writer", "err", err) return }
err = evt.Serialize(wc) if err != nil { l.Error("failed to serialize event", "err", err) wc.Close() return }
if err := wc.Close(); err != nil { l.Warn("failed to flush-close event write", "err", err) return }
lastWriteLk.Lock() lastWrite = time.Now() lastWriteLk.Unlock() } }}
func strptr(s string) *string { return &s }
func strptrOrNil(s string) *string { if s == "" { return nil } return &s}