From 750710071a3639ee122043ad879743da1e992006 Mon Sep 17 00:00:00 2001 From: rootp1 Date: Mon, 13 Jul 2026 13:52:13 +0530 Subject: [PATCH] Reduce stage log Put allocation overhead Signed-off-by: rootp1 --- pkg/app/server/stagelogstore/filestore.go | 7 +- .../server/stagelogstore/filestore_test.go | 126 ++++++++++++++++++ 2 files changed, 128 insertions(+), 5 deletions(-) diff --git a/pkg/app/server/stagelogstore/filestore.go b/pkg/app/server/stagelogstore/filestore.go index 9d7f7368a2..b8f0b44e96 100644 --- a/pkg/app/server/stagelogstore/filestore.go +++ b/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 { diff --git a/pkg/app/server/stagelogstore/filestore_test.go b/pkg/app/server/stagelogstore/filestore_test.go index a3953c16ea..686c89af3b 100644 --- a/pkg/app/server/stagelogstore/filestore_test.go +++ b/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 +}