bootstrapper: fix journald ram usage (#1553)

bootstrapper/internal/journald/BUILD.bazel

@@ -12,5 +12,8 @@ go_test(
     name = "journald_test",
     srcs = ["journald_test.go"],
     embed = [":journald"],
-    deps = ["@com_github_stretchr_testify//assert"],
+    deps = [
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
 )

bootstrapper/internal/journald/journald.go

@@ -11,18 +11,20 @@ package journald
 
 import (
 	"context"
-	"errors"
-	"fmt"
+	"io"
 	"os/exec"
 )
 
 type command interface {
-	Output() ([]byte, error)
+	Start() error
+	Wait() error
 }
 
 // Collector collects logs from journald.
 type Collector struct {
-	cmd command
+	cmd        command
+	stdoutPipe io.ReadCloser
+	stderrPipe io.ReadCloser
 }
 
 // NewCollector creates a new Collector for journald logs.
@@ -31,17 +33,33 @@ func NewCollector(ctx context.Context) (*Collector, error) {
 	if cmd.Err != nil {
 		return nil, cmd.Err
 	}
-	return &Collector{cmd}, nil
+	stdoutPipe, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	stderrPipe, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, err
+	}
+	collector := Collector{cmd, stdoutPipe, stderrPipe}
+	return &collector, nil
 }
 
-// Collect gets all journald logs from a service and returns a byte slice with the plain text logs.
-func (c *Collector) Collect() ([]byte, error) {
-	out, err := c.cmd.Output()
-	var exitErr *exec.ExitError
-	if errors.As(err, &exitErr) {
-		return nil, fmt.Errorf("executing %q failed: %s", c.cmd, exitErr.Stderr)
-	} else if err != nil {
-		return nil, fmt.Errorf("executing command: %w", err)
+// Start returns a pipe to read the systemd logs. This should be read with a bufio Reader.
+func (c *Collector) Start() (io.ReadCloser, error) {
+	if err := c.cmd.Start(); err != nil {
+		return nil, err
 	}
-	return out, nil
+	return c.stdoutPipe, nil
+}
+
+// Error returns output to stderr as bytes as well
+// as the exit code in form of an error.
+func (c *Collector) Error() ([]byte, error) {
+	stderr, err := io.ReadAll(c.stderrPipe)
+	if err != nil {
+		return nil, err
+	}
+	exitCode := c.cmd.Wait()
+	return stderr, exitCode
 }

bootstrapper/internal/journald/journald_test.go

@@ -7,58 +7,131 @@ SPDX-License-Identifier: AGPL-3.0-only
 package journald
 
 import (
+	"bytes"
 	"errors"
+	"io"
 	"os/exec"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
-type stubJournaldCommand struct {
-	outputReturn []byte
-	outputError  error
+func (s *stubReadCloser) Read(p []byte) (n int, err error) {
+	if s.readErr != nil {
+		return 0, s.readErr
+	}
+	return s.reader.Read(p)
 }
 
-func (j *stubJournaldCommand) Output() ([]byte, error) {
-	return j.outputReturn, j.outputError
+func (s *stubReadCloser) Close() error {
+	return s.closeErr
 }
 
-func TestCollect(t *testing.T) {
+func TestPipe(t *testing.T) {
 	someError := errors.New("failed")
 
 	testCases := map[string]struct {
-		command      *stubJournaldCommand
+		command      *stubCommand
+		stdoutPipe   io.ReadCloser
 		wantedOutput []byte
 		wantErr      bool
 	}{
 		"success": {
-			command: &stubJournaldCommand{},
+			command:      &stubCommand{},
+			wantedOutput: []byte("asdf"),
+			stdoutPipe:   &stubReadCloser{reader: bytes.NewReader([]byte("asdf"))},
 		},
 		"execution failed": {
-			command: &stubJournaldCommand{outputError: someError},
+			command: &stubCommand{startErr: someError},
 			wantErr: true,
 		},
 		"exit error": {
-			command: &stubJournaldCommand{outputError: &exec.ExitError{}},
+			command: &stubCommand{startErr: &exec.ExitError{}},
 			wantErr: true,
 		},
-		"output check": {
-			command:      &stubJournaldCommand{outputReturn: []byte("asdf")},
-			wantedOutput: []byte("asdf"),
-		},
 	}
 
 	for name, tc := range testCases {
 		t.Run(name, func(t *testing.T) {
 			assert := assert.New(t)
 
-			collector := Collector{cmd: tc.command}
+			collector := Collector{cmd: tc.command, stdoutPipe: tc.stdoutPipe}
 
-			out, err := collector.Collect()
+			pipe, err := collector.Start()
 			if tc.wantErr {
 				assert.Error(err)
+			} else {
+				stdout := make([]byte, 4)
+				_, err = io.ReadFull(pipe, stdout)
+				require.NoError(t, err)
+				assert.Equal(tc.wantedOutput, stdout)
 			}
-			assert.Equal(out, tc.wantedOutput)
 		})
 	}
 }
+
+func TestError(t *testing.T) {
+	someError := errors.New("failed")
+
+	testCases := map[string]struct {
+		stderrPipe io.ReadCloser
+		exitCode   error
+		wantErr    bool
+	}{
+		"success": {
+			stderrPipe: &stubReadCloser{readErr: io.EOF},
+		},
+		"reading error": {
+			stderrPipe: &stubReadCloser{readErr: someError},
+			wantErr:    true,
+		},
+		"close error": {
+			stderrPipe: &stubReadCloser{closeErr: someError, readErr: io.EOF},
+		},
+		"command exit": {
+			stderrPipe: &stubReadCloser{readErr: io.EOF},
+			exitCode:   someError,
+			wantErr:    true,
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			assert := assert.New(t)
+
+			collector := Collector{
+				stderrPipe: tc.stderrPipe,
+				cmd:        &stubCommand{waitErr: tc.exitCode},
+			}
+
+			stderrOut, err := collector.Error()
+			if tc.wantErr {
+				assert.Error(err)
+			} else {
+				assert.Equal(stderrOut, []byte{})
+			}
+		})
+	}
+}
+
+type stubCommand struct {
+	startCalled bool
+	startErr    error
+	waitErr     error
+}
+
+func (j *stubCommand) Start() error {
+	j.startCalled = true
+	return j.startErr
+}
+
+func (j *stubCommand) Wait() error {
+	return j.waitErr
+}
+
+type stubReadCloser struct {
+	reader   io.Reader
+	readErr  error
+	closeErr error
+}