Address comments.

Author: shunping

sdks/go/container/tools/buffered_logging.go

@@ -16,6 +16,7 @@
 package tools
 
 import (
+	"bytes"
 	"context"
 	"log"
 	"os"
@@ -60,31 +61,39 @@ func (b *BufferedLogger) Write(p []byte) (int, error) {
 	if b.logger == nil {
 		return os.Stderr.Write(p)
 	}
-	n, err := b.builder.Write(p)
+
 	if b.logs == nil {
 		b.logs = make([]string, 0, initialLogSize)
 	}
-	s := b.builder.String()
+
 	start := 0
 	for {
-		nl := strings.IndexByte(s[start:], '\n')
+		// Look for the next newline in the incoming byte slice directly
+		nl := bytes.IndexByte(p[start:], '\n')
 		if nl == -1 {
 			break
 		}
-		line := s[start : start+nl]
-		b.logs = append(b.logs, strings.TrimSuffix(line, "\r"))
+
+		// Write the segment up to the newline into the builder
+		b.builder.Write(p[start : start+nl])
+
+		// The builder now contains any previous partial line + the current complete segment
+		b.logs = append(b.logs, strings.TrimSuffix(b.builder.String(), "\r"))
+		b.builder.Reset()
+
 		start += nl + 1
 	}
-	if start > 0 {
-		b.builder.Reset()
-		if start < len(s) {
-			b.builder.WriteString(s[start:])
-		}
+
+	// Buffer any remaining bytes that didn't end in a newline
+	if start < len(p) {
+		b.builder.Write(p[start:])
 	}
+
 	if b.now().Sub(b.lastFlush) > b.flushInterval {
 		b.FlushAtDebug(b.periodicFlushContext)
 	}
-	return n, err
+
+	return len(p), nil
 }
 
 // Flush flushes the contents of the buffer to the logging service.

sdks/python/container/piputil.go

@@ -42,14 +42,14 @@ const pipLogFlushInterval time.Duration = 15 * time.Second
 const unrecoverableURL string = "https://beam.apache.org/documentation/sdks/python-unrecoverable-errors/index.html#pip-dependency-resolution-failures"
 
 // executeWithLogger runs the program with os.Stdin, piping stdout and stderr to bufLogger,
-// and flushes bufLogger based on the execution result.
+// and flushes the logger at ERROR severity on failure or DEBUG severity on success.
 func executeWithLogger(ctx context.Context, bufLogger *tools.BufferedLogger, prog string, args ...string) error {
 	err := execx.ExecuteEnvWithIO(nil, os.Stdin, bufLogger, bufLogger, prog, args...)
 	return bufLogger.Flush(ctx, err)
 }
 
 // executeWithOutput runs the program with os.Stdin, capturing stdout in a byte buffer
-// while piping stderr to bufLogger, and flushes bufLogger based on the execution result.
+// while piping stderr to bufLogger, and flushes the logger at ERROR severity on failure or DEBUG severity on success.
 func executeWithOutput(ctx context.Context, bufLogger *tools.BufferedLogger, prog string, args ...string) ([]byte, error) {
 	var stdout bytes.Buffer
 	err := execx.ExecuteEnvWithIO(nil, os.Stdin, &stdout, bufLogger, prog, args...)