From 858115137614a89799b5ee8edb4998f7d596f693 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 09:13:50 -0800 Subject: [PATCH] filer: gzip metadata log pieces before uploading them (#12580) * filer: let the fake-master test helper serve any master Co-Authored-By: Claude Opus 5.5 * filer: gzip metadata log pieces before uploading them Every metadata event carries the old and new entry with their whole chunk lists, so a file grown by small fsynced appends logs the same chunk records again and again: 2000 appends of 256 bytes through weed mount wrote 34.8 MB of log. The uploader only compresses when the first 128 bytes compress, and a log piece starts with fresh file ids and ETags, so it stored the log as is. The filer now gzips each piece itself and keeps the compressed bytes when they save at least 10%. After the same workload the volumes hold 3.05 MB instead of 35.2 MB, file data included. Readers already honor the chunk's IsCompressed flag on every path (filer-side decode, log file refs read straight from volume servers), and the assign request sizes the bytes actually stored. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/filer/filer_delete_collection_test.go | 11 +- weed/filer/filer_notify_append.go | 9 +- .../filer_notify_append_compress_test.go | 135 ++++++++++++++++++ 3 files changed, 152 insertions(+), 3 deletions(-) create mode 100644 weed/filer/filer_notify_append_compress_test.go diff --git a/weed/filer/filer_delete_collection_test.go b/weed/filer/filer_delete_collection_test.go index f44d3a0076..eae9ea6e3d 100644 --- a/weed/filer/filer_delete_collection_test.go +++ b/weed/filer/filer_delete_collection_test.go @@ -77,8 +77,15 @@ func (s *hookedStore) DeleteEntry(ctx context.Context, p util.FullPath) error { // MasterClient connected to a fake master the test can observe. func newFilerWithFakeMaster(t *testing.T) (*Filer, *hookedStore, *collectionDeleteMaster) { t.Helper() - master := &collectionDeleteMaster{calls: make(chan collectionDeleteCall, 4)} + f, store := newFilerWithMaster(t, master) + return f, store, master +} + +// newFilerWithMaster connects a filer's MasterClient to master, which must +// answer KeepConnected the way collectionDeleteMaster does. +func newFilerWithMaster(t *testing.T, master master_pb.SeaweedServer) (*Filer, *hookedStore) { + t.Helper() lis, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { @@ -129,7 +136,7 @@ func newFilerWithFakeMaster(t *testing.T) (*Filer, *hookedStore, *collectionDele f.LocalMetaLogBuffer.ShutdownLogBuffer() f.LocalMetaLogBuffer.WaitForShutdown() }) - return f, store, master + return f, store } // Deleting a bucket entry deletes its collection, and the request hanging up diff --git a/weed/filer/filer_notify_append.go b/weed/filer/filer_notify_append.go index 2b69b755ac..a769329ef4 100644 --- a/weed/filer/filer_notify_append.go +++ b/weed/filer/filer_notify_append.go @@ -96,6 +96,13 @@ func (f *Filer) metaLogReplicationFor(ruleReplication string) string { } func (f *Filer) assignAndUpload(ctx context.Context, targetFile string, data []byte) (*operation.AssignResult, *operation.UploadResult, error) { + // Consecutive events repeat whole chunk lists, but the uploader only + // samples the first bytes before deciding to compress. + isCompressed := false + if compressed, err := util.GzipData(data); err == nil && len(compressed)*10 < len(data)*9 { + data, isCompressed = compressed, true + } + // assign a volume location diskType, rule := f.resolveMetadataLogAssignDiskType(targetFile) assignRequest := &operation.VolumeAssignRequest{ @@ -121,7 +128,7 @@ func (f *Filer) assignAndUpload(ctx context.Context, targetFile string, data []b UploadUrl: targetUrl, Filename: "", Cipher: f.Cipher, - IsInputCompressed: false, + IsInputCompressed: isCompressed, MimeType: "", PairMap: nil, Jwt: assignResult.Auth, diff --git a/weed/filer/filer_notify_append_compress_test.go b/weed/filer/filer_notify_append_compress_test.go new file mode 100644 index 0000000000..0fd4bfb1a4 --- /dev/null +++ b/weed/filer/filer_notify_append_compress_test.go @@ -0,0 +1,135 @@ +package filer + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/binary" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +type assignMaster struct { + collectionDeleteMaster + volumeUrl string + requests chan *master_pb.AssignRequest +} + +func (m *assignMaster) Assign(ctx context.Context, req *master_pb.AssignRequest) (*master_pb.AssignResponse, error) { + m.requests <- req + location := &master_pb.Location{Url: m.volumeUrl, PublicUrl: m.volumeUrl} + return &master_pb.AssignResponse{Fid: "3,01637037d6", Count: 1, Location: location}, nil +} + +type receivedPiece struct { + contentEncoding string + body []byte +} + +func uploadMetaLogPiece(t *testing.T, data []byte) (*master_pb.AssignRequest, receivedPiece, *operation.UploadResult) { + t.Helper() + received := make(chan receivedPiece, 1) + volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + part, err := r.MultipartReader() + require.NoError(t, err) + p, err := part.NextPart() + require.NoError(t, err) + body, err := io.ReadAll(p) + require.NoError(t, err) + received <- receivedPiece{contentEncoding: p.Header.Get("Content-Encoding"), body: body} + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(operation.UploadResult{Size: uint32(len(body))}) + })) + t.Cleanup(volume.Close) + + master := &assignMaster{ + collectionDeleteMaster: collectionDeleteMaster{calls: make(chan collectionDeleteCall, 1)}, + volumeUrl: strings.TrimPrefix(volume.URL, "http://"), + requests: make(chan *master_pb.AssignRequest, 1), + } + f, _ := newFilerWithMaster(t, master) + f.GrpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + + _, uploadResult, err := f.assignAndUpload(context.Background(), string(appendTestLog), data) + require.NoError(t, err) + return <-master.requests, <-received, uploadResult +} + +func metaLogRecords(t *testing.T, n int) []byte { + t.Helper() + chunks := make([]*filer_pb.FileChunk, 200) + for i := range chunks { + md5 := make([]byte, 16) + _, err := rand.Read(md5) + require.NoError(t, err) + chunks[i] = &filer_pb.FileChunk{ + FileId: fmt.Sprintf("%d,%x%08x", i%7+1, i+1, binary.BigEndian.Uint32(md5)), + Offset: int64(i) * 256, + Size: 256, + ETag: base64.StdEncoding.EncodeToString(md5), + } + } + var buf []byte + for i := 0; i < n; i++ { + event, err := proto.Marshal(&filer_pb.SubscribeMetadataResponse{ + Directory: "/soak", + EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "journal.log", Chunks: chunks[:len(chunks)-1]}, + NewEntry: &filer_pb.Entry{Name: "journal.log", Chunks: chunks}, + }, + TsNs: int64(i + 1), + }) + require.NoError(t, err) + record, err := proto.Marshal(&filer_pb.LogEntry{TsNs: int64(i + 1), Data: event}) + require.NoError(t, err) + size := make([]byte, 4) + util.Uint32toBytes(size, uint32(len(record))) + buf = append(buf, size...) + buf = append(buf, record...) + } + return buf +} + +func TestMetaLogPieceUploadsCompressed(t *testing.T) { + data := metaLogRecords(t, 50) + + assign, piece, uploadResult := uploadMetaLogPiece(t, data) + + require.Equal(t, "gzip", piece.contentEncoding) + decompressed, err := util.DecompressData(piece.body) + require.NoError(t, err) + assert.Equal(t, data, decompressed) + assert.Less(t, len(piece.body)*5, len(data), "repeated chunk lists should compress well") + assert.Equal(t, uint64(len(piece.body)), assign.GetExpectedDataSize(), "assign sizes the bytes actually stored") + assert.Equal(t, uint32(1), uploadResult.Gzip) + assert.Equal(t, uint32(len(data)), uploadResult.Size, "chunk size and offsets stay in log bytes") +} + +func TestMetaLogPieceUploadsIncompressibleAsIs(t *testing.T) { + data := make([]byte, 64*1024) + _, err := rand.Read(data) + require.NoError(t, err) + + _, piece, uploadResult := uploadMetaLogPiece(t, data) + + assert.Empty(t, piece.contentEncoding) + assert.Equal(t, data, piece.body) + assert.Zero(t, uploadResult.Gzip) + assert.Equal(t, uint32(len(data)), uploadResult.Size) +}