Skip to content

Buffer BufferedLogger by newline to avoid log splitting #39288

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shunping wants to merge 3 commits into
base: master
Choose a base branch
from
+118 −17
Open

Buffer BufferedLogger by newline to avoid log splitting #39288

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
936baa2
Buffer BufferedLogger by newline to avoid log splitting
shunping Jul 10, 2026
d00dde9
Call BuferredLogger.Printf so the deps will not be separate lines.
shunping Jul 10, 2026
bbc6f24
Address comments
shunping Jul 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 25 additions & 2 deletions sdks/go/container/tools/buffered_logging.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,8 +63,23 @@ func (b *BufferedLogger) Write(p []byte) (int, error) {
if b.logs == nil {
b.logs = make([]string, 0, initialLogSize)
}
b.logs = append(b.logs, b.builder.String())
b.builder.Reset()
s := b.builder.String()
start := 0
for {
nl := strings.IndexByte(s[start:], '\n')
if nl == -1 {
break
}
line := s[start : start+nl]
b.logs = append(b.logs, strings.TrimSuffix(line, "\r"))
start += nl + 1
}
if start > 0 {
b.builder.Reset()
if start < len(s) {
b.builder.WriteString(s[start:])
}
}
Comment thread
shunping marked this conversation as resolved.
if b.now().Sub(b.lastFlush) > b.flushInterval {
b.FlushAtDebug(b.periodicFlushContext)
}
Expand All @@ -77,6 +92,10 @@ func (b *BufferedLogger) FlushAtError(ctx context.Context) {
if b.logger == nil {
return
}
if b.builder.Len() > 0 {
b.logs = append(b.logs, strings.TrimSuffix(b.builder.String(), "\r"))
b.builder.Reset()
}
for _, message := range b.logs {
b.logger.Errorf(ctx, "%s", message)
}
Expand All @@ -90,6 +109,10 @@ func (b *BufferedLogger) FlushAtDebug(ctx context.Context) {
if b.logger == nil {
return
}
if b.builder.Len() > 0 {
b.logs = append(b.logs, strings.TrimSuffix(b.builder.String(), "\r"))
b.builder.Reset()
}
for _, message := range b.logs {
b.logger.Printf(ctx, "%s", message)
}
Expand Down
100 changes: 88 additions & 12 deletions sdks/go/container/tools/buffered_logging_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,14 @@ import (
fnpb "github.com/apache/beam/sdks/v2/go/pkg/beam/model/fnexecution_v1"
)

func getAllLogEntries(catcher *logCatcher) []*fnpb.LogEntry {
var entries []*fnpb.LogEntry
for _, list := range catcher.msgs {
entries = append(entries, list.GetLogEntries()...)
}
return entries
}

func TestBufferedLogger(t *testing.T) {
ctx := context.Background()

Expand All @@ -31,7 +39,7 @@ func TestBufferedLogger(t *testing.T) {
l := &Logger{client: catcher}
bl := NewBufferedLogger(l)

message := []byte("test message")
message := []byte("test message\n")
n, err := bl.Write(message)
if err != nil {
t.Errorf("got error %v", err)
Expand Down Expand Up @@ -77,7 +85,8 @@ func TestBufferedLogger(t *testing.T) {
l := &Logger{client: catcher}
bl := NewBufferedLogger(l)

messages := []string{"foo", "bar", "baz"}
messages := []string{"foo\n", "bar\n", "baz\n"}
expected := []string{"foo", "bar", "baz"}

for _, message := range messages {
messBytes := []byte(message)
Expand All @@ -93,10 +102,14 @@ func TestBufferedLogger(t *testing.T) {

bl.FlushAtDebug(ctx)

received := catcher.msgs[0].GetLogEntries()
received := getAllLogEntries(catcher)

if got, want := len(received), len(expected); got != want {
t.Fatalf("expected %d log entries received, got %d", want, got)
}

for i, message := range received {
if got, want := message.Message, messages[i]; got != want {
if got, want := message.Message, expected[i]; got != want {
t.Errorf("got message %q, want %q", got, want)
}

Expand Down Expand Up @@ -139,7 +152,8 @@ func TestBufferedLogger(t *testing.T) {
l := &Logger{client: catcher}
bl := NewBufferedLogger(l)

messages := []string{"foo", "bar", "baz"}
messages := []string{"foo\n", "bar\n", "baz\n"}
expected := []string{"foo", "bar", "baz"}

for _, message := range messages {
messBytes := []byte(message)
Expand All @@ -155,10 +169,14 @@ func TestBufferedLogger(t *testing.T) {

bl.FlushAtError(ctx)

received := catcher.msgs[0].GetLogEntries()
received := getAllLogEntries(catcher)

if got, want := len(received), len(expected); got != want {
t.Fatalf("expected %d log entries received, got %d", want, got)
}

for i, message := range received {
if got, want := message.Message, messages[i]; got != want {
if got, want := message.Message, expected[i]; got != want {
t.Errorf("got message %q, want %q", got, want)
}

Expand Down Expand Up @@ -195,7 +213,8 @@ func TestBufferedLogger(t *testing.T) {
startTime := time.Now()
bl.now = func() time.Time { return startTime }

messages := []string{"foo", "bar"}
messages := []string{"foo\n", "bar\n"}
expected := []string{"foo", "bar"}

for i, message := range messages {
if i > 1 {
Expand All @@ -212,7 +231,8 @@ func TestBufferedLogger(t *testing.T) {
}
}

lastMessage := "baz"
lastMessage := "baz\n"
expected = append(expected, "baz")
bl.now = func() time.Time { return startTime.Add(6 * time.Second) }
messBytes := []byte(lastMessage)
n, err := bl.Write(messBytes)
Expand All @@ -225,11 +245,14 @@ func TestBufferedLogger(t *testing.T) {
}

// Type should have auto-flushed at debug after the third message
received := catcher.msgs[0].GetLogEntries()
messages = append(messages, lastMessage)
received := getAllLogEntries(catcher)

if got, want := len(received), len(expected); got != want {
t.Fatalf("expected %d log entries received, got %d", want, got)
}

for i, message := range received {
if got, want := message.Message, messages[i]; got != want {
if got, want := message.Message, expected[i]; got != want {
t.Errorf("got message %q, want %q", got, want)
}

Expand All @@ -238,4 +261,57 @@ func TestBufferedLogger(t *testing.T) {
}
}
})

t.Run("partial write splitting", func(t *testing.T) {
catcher := &logCatcher{}
l := &Logger{client: catcher}
bl := NewBufferedLogger(l)

// Write a partial line
n, err := bl.Write([]byte("hello "))
if err != nil {
t.Errorf("got error %v", err)
}
if n != 6 {
t.Errorf("got %d, want 6", n)
}
if len(bl.logs) != 0 {
t.Errorf("expected no logs buffered yet, got %d", len(bl.logs))
}

// Write remainder and a second line
n, err = bl.Write([]byte("world\nline2\npartial"))
if err != nil {
t.Errorf("got error %v", err)
}
if n != 19 {
t.Errorf("got %d, want 19", n)
}

if got, want := len(bl.logs), 2; got != want {
t.Errorf("expected 2 logs buffered, got %d", got)
}
if got, want := bl.logs[0], "hello world"; got != want {
t.Errorf("got %q, want %q", got, want)
}
if got, want := bl.logs[1], "line2"; got != want {
t.Errorf("got %q, want %q", got, want)
}

// Flush should flush the final partial message
bl.FlushAtDebug(ctx)
received := getAllLogEntries(catcher)
if got, want := len(received), 3; got != want {
t.Fatalf("expected 3 log entries received, got %d", got)
}
if got, want := received[0].Message, "hello world"; got != want {
t.Errorf("got message %q, want %q", got, want)
}
if got, want := received[1].Message, "line2"; got != want {
t.Errorf("got message %q, want %q", got, want)
}
if got, want := received[2].Message, "partial"; got != want {
t.Errorf("got message %q, want %q", got, want)
}
})
}
8 changes: 5 additions & 3 deletions sdks/python/container/boot.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package main

import (
"bytes"
"context"
"encoding/json"
"errors"
Expand Down Expand Up @@ -580,10 +581,13 @@ func logRuntimeDependencies(ctx context.Context, bufLogger *tools.BufferedLogger
}
bufLogger.Printf(ctx, "Dependencies in %s:", phase)

@tvalentyn tvalentyn Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we please append Dependencies in to the log itself?

Maybe we should also prepend it with "Runtime dependencies" ?

Context: this way it would be easier to search in the cloud console "dependencies" and seeing the line that filters the dependenices + content.

We should do the same for submission deps as well.

Fine to do in a separate PR if you feel this is out of scope or file an issue if you'd rather not do this rn.

args = []string{"-m", "pip", "freeze", "--all"}
if err := execx.ExecuteEnvWithIO(nil, os.Stdin, bufLogger, bufLogger, pythonVersion, args...); err != nil {

var stdout bytes.Buffer
if err := execx.ExecuteEnvWithIO(nil, os.Stdin, &stdout, bufLogger, pythonVersion, args...); err != nil {

@tvalentyn tvalentyn Jul 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, @shunping ! This PR allows for additional flexibility to retain logs and output them in a better form.

My main worry is with the complexity a developer should reason about to get things right:

  • remembering the behavior or buffered logger that splits lines
  • knowing how to bypass line splitting
  • knowing how to properly flush and print accumulated logs.

Do you think we could simplify this with better helpers? Perhaps with something like:

// executeAndBatchLogs runs a command and batches its output into a single
// DEBUG log message on success or ERROR log message on failure.

func executeAndBatchLogs(ctx context.Context, env map[string]string, bufLogger *tools.BufferedLogger, prog string, args ...string) error {
...
}

// executeAndStreamLogs runs a command and streams both its standard output and
// standard error directly to the provided BufferedLogger. This allows the BufferedLogger
// to automatically trigger its periodic flush interval for potentially long-running commands.
func executeAndStreamLogs(ctx context.Context, env map[string]string, bufLogger *tools.BufferedLogger, prog string, args ...string) error {
...
}

WDYT?

bufLogger.FlushAtError(ctx)
} else {
bufLogger.FlushAtDebug(ctx)
bufLogger.Printf(ctx, "%s", stdout.String())
}
return nil
}
Expand All @@ -602,5 +606,3 @@ func logSubmissionEnvDependencies(ctx context.Context, bufLogger *tools.Buffered
bufLogger.Printf(ctx, "%s", string(content))
return nil
}


Loading