Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136package logview
import ( "bufio" "context" "errors" "fmt" "io" "os" "strings"
"github.com/hpcloud/tail" "tangled.org/core/spindle/artifactstore" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models")
// streams a workflow's log lines: tails the log file while running, reads// the uploaded artifact once finished, same for every role. stop ends a// live follow early, the channel closes when the source drains or ctx endsfunc Follow(ctx context.Context, d *db.DB, reader artifactstore.Reader, logDir string, wid models.WorkflowId, finished bool) (<-chan *tail.Line, func(), error) { lines, stop, _, err := FollowDrain(ctx, d, reader, logDir, wid, finished) return lines, stop, err}
// drain reads to EOF first and the last lines still come outfunc FollowDrain(ctx context.Context, d *db.DB, reader artifactstore.Reader, logDir string, wid models.WorkflowId, finished bool) (<-chan *tail.Line, func(), func(), error) { logPath := models.LogFilePath(logDir, wid) var lookup func() (*db.FinishedLog, error) if finished && reader != nil && d != nil { lookup = func() (*db.FinishedLog, error) { return d.GetFinishedLog(wid) } } rc, err := archiveOrLocal(ctx, lookup, reader, logPath, finished) if err != nil { return nil, nil, nil, err } if rc != nil { ch, cancel := streamArchive(ctx, rc) return ch, cancel, cancel, nil }
t, err := tail.TailFile(logPath, tail.Config{ Follow: !finished, ReOpen: !finished, MustExist: false, Location: &tail.SeekInfo{Offset: 0, Whence: io.SeekStart}, }) if err != nil { return nil, nil, nil, err } stop := func() { // nobody reads t.Lines anymore, drain it or the tail blocks on stop go func() { for range t.Lines { } }() _ = t.Stop() } drain := func() { _ = t.StopAtEOF() } return t.Lines, stop, drain, nil}
// a missing local file means the archive row exists, since the file is only removed after the row is savedfunc archiveOrLocal(ctx context.Context, lookup func() (*db.FinishedLog, error), reader artifactstore.Reader, logPath string, finished bool) (io.ReadCloser, error) { if lookup == nil { // stat it first, tail waits forever on a missing file if finished { if _, err := os.Stat(logPath); err != nil { return nil, fmt.Errorf("finished workflow log unavailable: %w", err) } } return nil, nil } rc, artifactErr := openArchive(ctx, lookup, reader) if artifactErr == nil { return rc, nil } if _, err := os.Stat(logPath); err == nil { return nil, nil } else if !errors.Is(err, os.ErrNotExist) { return nil, fmt.Errorf("stat local workflow log: %w", err) } rc, retryErr := openArchive(ctx, lookup, reader) if retryErr != nil { return nil, fmt.Errorf("finished workflow log unavailable: %w", retryErr) } return rc, nil}
func openArchive(ctx context.Context, lookup func() (*db.FinishedLog, error), reader artifactstore.Reader) (io.ReadCloser, error) { fl, err := lookup() if err != nil { return nil, fmt.Errorf("find archived workflow log: %w", err) } if fl.Ref == "" { return nil, errors.New("archived workflow log has an empty ref") } rc, err := reader.Open(ctx, fl.Ref) if err != nil { return nil, fmt.Errorf("open archived workflow log: %w", err) } return rc, nil}
func streamArchive(ctx context.Context, rc io.ReadCloser) (<-chan *tail.Line, func()) { ch := make(chan *tail.Line, 64) followCtx, cancel := context.WithCancel(ctx) go func() { defer close(ch) defer rc.Close() send := func(line *tail.Line) bool { select { case <-followCtx.Done(): return false case ch <- line: return true } } // bufio.Scanner gives up past its token cap, and log lines can be longer reader := bufio.NewReader(rc) for { text, err := reader.ReadString('\n') if text != "" && !send(&tail.Line{Text: strings.TrimSuffix(strings.TrimSuffix(text, "\n"), "\r")}) { return } if err != nil { if err != io.EOF { send(&tail.Line{Err: err}) } return } } }() return ch, cancel}