fix(lmeval): Send Job logs to Pod StdOut (#358)

* Send job logs to pod logs

* Preserve pipewriter with newline introduction

* Fix error return of the pod log writer

* Update driver.go

---------

Co-authored-by: Rui Vieira <[email protected]>

controllers/lmes/driver/driver.go

@@ -131,7 +131,8 @@ func (d *driverImpl) Run() error {
 		}
 	}
 	toConsole(filepath.Join(d.Option.OutputPath, "stdout.log"))
-	toConsole(filepath.Join(d.Option.OutputPath, "stderr.log"))
+	// TODO: Remove so that log is not printed again at job's completion
+	// toConsole(filepath.Join(d.Option.OutputPath, "stderr.log"))
 
 	d.updateCompleteStatus(execErr)
 
@@ -283,6 +284,29 @@ func (dc *driverComm) notifyShutdownWait() {
 	dc.connection <- 1
 }
 
+type PodLogAndPipeWriter struct {
+	pw *io.PipeWriter
+}
+
+func (f PodLogAndPipeWriter) Write(p []byte) (n int, err error) {
+	// format lm-eval output to stdout and the pipewriter
+	line := string(p)
+	res, _ := fmt.Print(line)
+	if _, err := f.pw.Write(p); err != nil {
+		return 0, err
+	}
+
+	// carriage returns do not correctly display, so replace with newlines
+	// carriage returns also break the pipewriter, so make sure we get newlines as well.
+	if strings.ContainsAny(line, "\r") && !strings.ContainsAny(line, "\n") {
+		fmt.Print("\n")
+		if _, err := f.pw.Write([]byte("\n")); err != nil {
+			return 0, err
+		}
+	}
+	return res, nil
+}
+
 func (d *driverImpl) exec() error {
 	// create Unitxt task recipes
 	if err := d.createTaskRecipes(); err != nil {
@@ -317,7 +341,7 @@ func (d *driverImpl) exec() error {
 	// have a pipe to check the output and report progress
 	// lm-eval's outputs are in the stderr
 	pr, pw := io.Pipe()
-	mwriter := io.MultiWriter(stderr, pw)
+	mwriter := io.MultiWriter(stderr, PodLogAndPipeWriter{pw})
 	scanner := bufio.NewScanner(pr)
 
 	executor := exec.Command(d.Option.Args[0], args...)