Reduce stage log Put allocation overhead

pkg/app/server/stagelogstore/filestore.go

@@ -70,14 +70,11 @@ func (f *stageLogFileStore) Get(ctx context.Context, deploymentID, stageID strin
 func (f *stageLogFileStore) Put(ctx context.Context, deploymentID, stageID string, retriedCount int32, lf *logFragment) error {
 	path := stageLogPath(deploymentID, stageID, retriedCount)
 	var buf bytes.Buffer
+	encoder := json.NewEncoder(&buf)
 	for _, lb := range lf.Blocks {
-		// TODO: Reduce the number of marshaling log blocks for improving performance
-		raw, err := json.Marshal(lb)
-		if err != nil {
+		if err := encoder.Encode(lb); err != nil {
 			return err
 		}
-		buf.Write(raw)
-		buf.WriteString("\n")
 	}
 
 	if lf.Completed {

pkg/app/server/stagelogstore/filestore_test.go

@@ -15,8 +15,11 @@
 package stagelogstore
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"io"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -25,6 +28,7 @@ import (
 
 	"github.com/pipe-cd/pipecd/pkg/filestore"
 	"github.com/pipe-cd/pipecd/pkg/filestore/filestoretest"
+	"github.com/pipe-cd/pipecd/pkg/model"
 )
 
 func TestFileStoreGet(t *testing.T) {
@@ -121,3 +125,125 @@ EOL`,
 		})
 	}
 }
+
+func TestFileStorePut(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	store := filestoretest.NewMockStore(ctrl)
+
+	fs := stageLogFileStore{
+		filestore: store,
+	}
+
+	lf := &logFragment{
+		Blocks: []*model.LogBlock{
+			{
+				Index:     1,
+				Log:       "Hello 1",
+				Severity:  model.LogSeverity_INFO,
+				CreatedAt: 1590499431,
+			},
+			{
+				Index:     2,
+				Log:       "Hello 2\nWorld",
+				Severity:  model.LogSeverity_ERROR,
+				CreatedAt: 1590499432,
+			},
+		},
+		Completed: true,
+	}
+
+	store.EXPECT().
+		Put(context.TODO(), "log/deployment-id/stage-id/0.txt", []byte("{\"index\":1,\"log\":\"Hello 1\",\"created_at\":1590499431}\n{\"index\":2,\"log\":\"Hello 2\\nWorld\",\"severity\":2,\"created_at\":1590499432}\nEOL")).
+		Return(nil)
+
+	assert.NoError(t, fs.Put(context.TODO(), "deployment-id", "stage-id", 0, lf))
+}
+
+func BenchmarkStageLogFileStorePut(b *testing.B) {
+	for _, blockCount := range []int{10, 100, 1000} {
+		lf := benchmarkLogFragment(blockCount, true)
+		b.Run("legacy/"+strconv.Itoa(blockCount), func(b *testing.B) {
+			store := stageLogFileStore{filestore: benchmarkFilestore{}}
+			ctx := context.Background()
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				if err := benchmarkLegacyPut(&store, ctx, "deployment-id", "stage-id", 0, lf); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+		b.Run("current/"+strconv.Itoa(blockCount), func(b *testing.B) {
+			store := stageLogFileStore{filestore: benchmarkFilestore{}}
+			ctx := context.Background()
+			b.ReportAllocs()
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				if err := store.Put(ctx, "deployment-id", "stage-id", 0, lf); err != nil {
+					b.Fatal(err)
+				}
+			}
+		})
+	}
+}
+
+func benchmarkLegacyPut(f *stageLogFileStore, ctx context.Context, deploymentID, stageID string, retriedCount int32, lf *logFragment) error {
+	path := stageLogPath(deploymentID, stageID, retriedCount)
+	var buf bytes.Buffer
+	for _, lb := range lf.Blocks {
+		raw, err := json.Marshal(lb)
+		if err != nil {
+			return err
+		}
+		buf.Write(raw)
+		buf.WriteString("\n")
+	}
+
+	if lf.Completed {
+		buf.Write(eol)
+	}
+	return f.filestore.Put(ctx, path, buf.Bytes())
+}
+
+func benchmarkLogFragment(blockCount int, completed bool) *logFragment {
+	blocks := make([]*model.LogBlock, 0, blockCount)
+	for i := 0; i < blockCount; i++ {
+		blocks = append(blocks, &model.LogBlock{
+			Index:     int64(i),
+			Log:       strings.Repeat("benchmark-stage-log-line-", 4) + strconv.Itoa(i),
+			Severity:  model.LogSeverity_INFO,
+			CreatedAt: 1590499431 + int64(i),
+		})
+	}
+	return &logFragment{
+		Blocks:    blocks,
+		Completed: completed,
+	}
+}
+
+type benchmarkFilestore struct{}
+
+func (benchmarkFilestore) Get(context.Context, string) ([]byte, error) {
+	panic("not implemented")
+}
+
+func (benchmarkFilestore) GetReader(context.Context, string) (io.ReadCloser, error) {
+	panic("not implemented")
+}
+
+func (benchmarkFilestore) Put(context.Context, string, []byte) error {
+	return nil
+}
+
+func (benchmarkFilestore) List(context.Context, string) ([]filestore.ObjectAttrs, error) {
+	panic("not implemented")
+}
+
+func (benchmarkFilestore) Delete(context.Context, string) error {
+	panic("not implemented")
+}
+
+func (benchmarkFilestore) Close() error {
+	return nil
+}