feat: adds task prefix to run target command logs

Author: nxtcoder17

parser/parse-task.go

@@ -123,12 +123,13 @@ func ParseTask(ctx context.Context, prf *types.ParsedRunfile, task types.Task) (
 	}
 
 	return &types.ParsedTask{
+		Name:        task.Name,
 		Shell:       task.Shell,
 		WorkingDir:  *task.Dir,
 		Interactive: task.Interactive,
 		Env:         taskEnv,
 		Commands:    commands,
 		Watch:       watch,
-		Parallel: task.Parallel,
+		Parallel:    task.Parallel,
 	}, nil
 }

parser/parser-runfile.go

@@ -18,7 +18,10 @@ func parseRunfile(runfile *types.Runfile) (*types.ParsedRunfile, error) {
 		Tasks: make(map[string]types.Task),
 	}
 
-	prf.Tasks = runfile.Tasks
+	for k, task := range runfile.Tasks {
+		task.Name = k
+		prf.Tasks[k] = task
+	}
 
 	m, err := parseIncludes(runfile.Includes)
 	if err != nil {
@@ -27,6 +30,7 @@ func parseRunfile(runfile *types.Runfile) (*types.ParsedRunfile, error) {
 
 	for k, iprf := range m {
 		for taskName, task := range iprf.Tasks {
+			task.Name = k
 			task.Metadata.RunfilePath = &iprf.Metadata.RunfilePath
 			prf.Tasks[fmt.Sprintf("%s:%s", k, taskName)] = task
 		}

runner/run-task.go

@@ -4,6 +4,7 @@ import (
 	// "bytes"
 	"context"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -53,7 +54,17 @@ func isTTY() bool {
 	return ((stdout.Mode() & os.ModeCharDevice) == os.ModeCharDevice) && ((stderr.Mode() & os.ModeCharDevice) == os.ModeCharDevice)
 }
 
-func createCommands(ctx Context, prf *types.ParsedRunfile, pt *types.ParsedTask, args runTaskArgs) ([]executor.CommandGroup, error) {
+type CreateCommandGroupArgs struct {
+	Runfile *types.ParsedRunfile
+	Task    *types.ParsedTask
+
+	Prefix string
+
+	Stdout io.Writer
+	Stderr io.Writer
+}
+
+func createCommandGroups(ctx Context, prf *types.ParsedRunfile, pt *types.ParsedTask, stdout *LogWriter, stderr *LogWriter) ([]executor.CommandGroup, error) {
 	var cmds []executor.CommandGroup
 
 	for _, cmd := range pt.Commands {
@@ -70,7 +81,7 @@ func createCommands(ctx Context, prf *types.ParsedRunfile, pt *types.ParsedTask,
 					return nil, errors.WithErr(err).KV("env-vars", prf.Env)
 				}
 
-				rtCommands, err := createCommands(ctx, prf, rtp, args)
+				rtCommands, err := createCommandGroups(ctx, prf, rtp, stdout, stderr)
 				if err != nil {
 					return nil, errors.WithErr(err).KV("env-vars", prf.Env)
 				}
@@ -101,8 +112,8 @@ func createCommands(ctx Context, prf *types.ParsedRunfile, pt *types.ParsedTask,
 						Cmd:         *cmd.Command,
 						WorkingDir:  pt.WorkingDir,
 						interactive: pt.Interactive,
-						Stdout:      os.Stdout,
-						Stderr:      os.Stderr,
+						Stdout:      stdout.WithPrefix(pt.Name),
+						Stderr:      stdout.WithPrefix(pt.Name),
 					})
 				})
 
@@ -296,7 +307,9 @@ func runTask(ctx Context, prf *types.ParsedRunfile, args runTaskArgs) error {
 		return errors.WithErr(err)
 	}
 
-	execCommands, err := createCommands(ctx, prf, pt, args)
+	logStdout := &LogWriter{w: os.Stdout}
+
+	execCommands, err := createCommandGroups(ctx, prf, pt, logStdout, logStdout)
 	if err != nil {
 		return err
 	}

runner/run-task_test.go

@@ -24,7 +24,7 @@ func TestCreateCommands(t *testing.T) {
 	}
 
 	for _, tt := range tests {
-		cg, err := createCommands(Context{Context: context.TODO(), Logger: slog.Default()}, &tt.prf, &tt.pt, tt.rta)
+		cg, err := createCommandGroups(Context{Context: context.TODO(), Logger: slog.Default()}, &tt.prf, &tt.pt, tt.rta)
 		if err != nil {
 			t.Error(err)
 		}

runner/writer.go

@@ -0,0 +1,78 @@
+package runner
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+)
+
+type LineWriter struct {
+	prefix string
+	w      io.Writer
+}
+
+// Write implements io.Writer.
+func (lw *LineWriter) Write(p []byte) (n int, err error) {
+	return lw.w.Write(p)
+}
+
+type LogWriter struct {
+	w  io.Writer
+	mu sync.Mutex
+	wg sync.WaitGroup
+}
+
+// Write implements io.Writer.
+func (s *LogWriter) Write(p []byte) (n int, err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.w.Write(p)
+}
+
+var _ io.Writer = (*LogWriter)(nil)
+
+func (s *LogWriter) WithPrefix(prefix string) io.Writer {
+	pr, pw := io.Pipe()
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+		copyStream(prefix, s.w, pr)
+	}()
+	return &LineWriter{prefix: prefix, w: pw}
+}
+
+func (s *LogWriter) Wait() {
+	s.wg.Wait()
+}
+
+func copyBytes(prefix string, src []byte) []byte {
+	dest := make([]byte, 0, len(src))
+	for i := range src {
+		if i == 0 || src[i-1] == '\n' {
+			dest = append(dest, []byte(fmt.Sprintf("[%s] ", prefix))...)
+		}
+		dest = append(dest, src[i])
+	}
+	return dest
+}
+
+func copyStream(prefix string, dest io.Writer, src io.Reader) {
+	r := bufio.NewReader(src)
+	for {
+		b, err := r.ReadBytes('\n')
+		if err != nil {
+			// logger.Info("stdout", "msg", string(msg[:n]), "err", err)
+			fmt.Println("ERR: ", err)
+			if errors.Is(err, io.EOF) {
+				dest.Write([]byte(fmt.Sprintf("[%s] ", prefix)))
+				dest.Write(b)
+				return
+			}
+		}
+
+		dest.Write([]byte(fmt.Sprintf("[%s] ", prefix)))
+		dest.Write(b)
+	}
+}

types/parsed-types.go

@@ -10,6 +10,9 @@ type ParsedRunfile struct {
 }
 
 type ParsedTask struct {
+	// Name should be resolved from key itself
+	Name string `json:"-"`
+
 	Shell       []string          `json:"shell"`
 	WorkingDir  string            `json:"workingDir"`
 	Watch       *TaskWatch        `json:"watch,omitempty"`