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) +}