Monorepo for Tangled
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189package spindle
import ( "context" "encoding/json" "errors" "fmt" "io" "net/http" "os" "time"
"tangled.org/core/eventstream" "tangled.org/core/log" "tangled.org/core/spindle/models"
"github.com/go-chi/chi/v5" "github.com/gorilla/websocket" "github.com/hpcloud/tail")
var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024,}
func (s *Spindle) Events(w http.ResponseWriter, r *http.Request) { l := log.SubLogger(s.l, "eventstream") l.Debug("received new connection")
err := eventstream.Stream(w, r, eventstream.StreamConfig{ Backend: s.db, Notifier: s.n, Logger: l, }) if err != nil && !errors.Is(err, eventstream.ErrDrainCap) { l.Error("event stream ended with error", "err", err) }}
func (s *Spindle) Logs(w http.ResponseWriter, r *http.Request) { wid, err := getWorkflowID(r) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return }
l := s.l.With("handler", "Logs") l = s.l.With("wid", wid)
conn, err := upgrader.Upgrade(w, r, nil) if err != nil { l.Error("websocket upgrade failed", "err", err) http.Error(w, "failed to upgrade", http.StatusInternalServerError) return } defer func() { _ = conn.WriteControl( websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "log stream complete"), time.Now().Add(time.Second), ) conn.Close() }() l.Debug("upgraded http to wss")
ctx, cancel := context.WithCancel(r.Context()) defer cancel()
go func() { for { if _, _, err := conn.NextReader(); err != nil { l.Debug("client disconnected", "err", err) cancel() return } } }()
if err := s.streamLogsFromDisk(ctx, conn, wid); err != nil { l.Info("log stream ended", "err", err) }
l.Info("logs connection closed")}
func (s *Spindle) streamLogsFromDisk(ctx context.Context, conn *websocket.Conn, wid models.WorkflowId) error { status, err := s.db.GetStatus(wid) if err != nil { return err } isFinished := models.StatusKind(status.Status).IsFinish()
filePath := models.LogFilePath(s.cfg.Server.LogDir, wid)
if status.Status == models.StatusKindFailed.String() && status.Error != nil { if _, err := os.Stat(filePath); os.IsNotExist(err) { msgs := []models.LogLine{ { Kind: models.LogKindControl, Content: "", StepId: 0, StepKind: models.StepKindUser, }, { Kind: models.LogKindData, Content: *status.Error, }, }
for _, msg := range msgs { b, err := json.Marshal(msg) if err != nil { return err }
if err := conn.WriteMessage(websocket.TextMessage, b); err != nil { return fmt.Errorf("failed to write to websocket: %w", err) } }
return nil } }
config := tail.Config{ Follow: !isFinished, ReOpen: !isFinished, MustExist: false, Location: &tail.SeekInfo{ Offset: 0, Whence: io.SeekStart, }, // Logger: tail.DiscardingLogger, }
t, err := tail.TailFile(filePath, config) if err != nil { return fmt.Errorf("failed to tail log file: %w", err) } defer t.Stop()
for { select { case <-ctx.Done(): return ctx.Err() case line := <-t.Lines: if line == nil && isFinished { return fmt.Errorf("tail completed") }
if line == nil { return fmt.Errorf("tail channel closed unexpectedly") }
if line.Err != nil { return fmt.Errorf("error tailing log file: %w", line.Err) }
if err := conn.WriteMessage(websocket.TextMessage, []byte(line.Text)); err != nil { return fmt.Errorf("failed to write to websocket: %w", err) } case <-time.After(30 * time.Second): // send a keep-alive if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil { return fmt.Errorf("failed to write control: %w", err) } } }}
func getWorkflowID(r *http.Request) (models.WorkflowId, error) { knot := chi.URLParam(r, "knot") rkey := chi.URLParam(r, "rkey") name := chi.URLParam(r, "name")
if knot == "" || rkey == "" || name == "" { return models.WorkflowId{}, fmt.Errorf("missing required parameters") }
return models.WorkflowId{ PipelineId: models.PipelineId{ Knot: knot, Rkey: rkey, }, Name: name, }, nil}