diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index bf728771f..b8beefd71 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -136,6 +136,10 @@ message ListEntriesRequest { bool inclusiveStartFrom = 4; uint32 limit = 5; int64 snapshot_ts_ns = 6; + // Leave the chunk list out of every entry in the response. The attributes + // still carry the file size, so a listing that only reads attributes can + // ask for this and skip the largest part of the payload. + bool omit_chunks = 7; } message ListEntriesResponse { diff --git a/weed/filer/entry_codec_attributes.go b/weed/filer/entry_codec_attributes.go new file mode 100644 index 000000000..71146fd79 --- /dev/null +++ b/weed/filer/entry_codec_attributes.go @@ -0,0 +1,213 @@ +package filer + +import ( + "context" + "errors" + "fmt" + "sync" + "unicode/utf8" + + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// Field numbers from filer.proto. Only the ones this decoder has to recognise +// are named; every other Entry field is handed to the generated unmarshaller +// as-is. +const ( + entryChunksField = 3 // Entry.chunks + fileChunkOffsetField = 2 // FileChunk.offset + fileChunkSizeField = 3 // FileChunk.size +) + +// The chunk bytes are the one part of the blob the generated unmarshaller never +// sees, so the checks it would have made are made here instead: a submessage +// has to parse, and a proto3 string has to be valid UTF-8. Anything else in a +// FileChunk is a scalar, which walking it already validates. +// TestChunkValidationCoversEveryField fails if FileChunk gains a field of +// either kind that is missing from these. +var ( + fileChunkMessageFields = map[protowire.Number]bool{ + 7: true, // fid, a FileId of scalars only, so walking it is a full check + 8: true, // source_fid + } + fileChunkStringFields = map[protowire.Number]bool{ + 1: true, // file_id + 5: true, // e_tag + 6: true, // source_file_id + } +) + +// attributesScratchPool holds the re-encoded entry, which is the blob minus its +// chunks and so much smaller than what came in. +var attributesScratchPool = sync.Pool{ + New: func() any { + b := make([]byte, 0, 256) + return &b + }, +} + +// DecodeListedEntry decodes one listed entry, dropping the chunk list when the +// listing asked for attributes only. +func DecodeListedEntry(ctx context.Context, entry *Entry, blob []byte) error { + if filer_pb.ChunksOmitted(ctx) { + return entry.DecodeAttributesOnly(blob) + } + return entry.DecodeAttributesAndChunks(blob) +} + +// DecodeAttributesOnly fills entry from blob without building its chunk list, +// which is the bulk of the work for anything but a tiny file. Chunks are still +// measured, because an entry whose stored FileSize is zero takes its size from +// them, but no FileChunk is allocated. +// +// entry.Chunks is left nil. The one exception is a hard link, whose attributes +// and chunks are replaced wholesale by a full decode of its own record in +// FilerStoreWrapper.maybeReadHardLink straight after the store listing. Only a +// caller that reads attributes and nothing else may use this. +func (entry *Entry) DecodeAttributesOnly(blob []byte) error { + // The scratch buffer is only taken once a chunk is actually found, so an + // entry with none — every directory, for one — neither re-encodes nor + // touches the pool, and is unmarshalled where it lies. + var scratchPtr *[]byte + var scratch []byte + defer func() { + if scratchPtr != nil { + *scratchPtr = scratch + attributesScratchPool.Put(scratchPtr) + } + }() + + var chunkExtent uint64 + attributes := blob + for pos := 0; pos < len(blob); { + rest := blob[pos:] + num, typ, tagLen := protowire.ConsumeTag(rest) + if tagLen < 0 { + return fmt.Errorf("decoding value blob for %s: %w", entry.FullPath, protowire.ParseError(tagLen)) + } + var valLen int + if num == entryChunksField && typ == protowire.BytesType { + chunk, n := protowire.ConsumeBytes(rest[tagLen:]) + if n < 0 { + return fmt.Errorf("decoding value blob for %s: %w", entry.FullPath, protowire.ParseError(n)) + } + valLen = n + end, err := chunkExtentEnd(chunk) + if err != nil { + return fmt.Errorf("decoding value blob for %s: %w", entry.FullPath, err) + } + if end > chunkExtent { + chunkExtent = end + } + if scratchPtr == nil { + scratchPtr = attributesScratchPool.Get().(*[]byte) + scratch = append((*scratchPtr)[:0], blob[:pos]...) + } + } else { + valLen = protowire.ConsumeFieldValue(num, typ, rest[tagLen:]) + if valLen < 0 { + return fmt.Errorf("decoding value blob for %s: %w", entry.FullPath, protowire.ParseError(valLen)) + } + if scratchPtr != nil { + scratch = append(scratch, rest[:tagLen+valLen]...) + } + } + pos += tagLen + valLen + } + if scratchPtr != nil { + attributes = scratch + } + + message := pbEntryPool.Get().(*filer_pb.Entry) + defer func() { + resetPbEntry(message) + pbEntryPool.Put(message) + }() + + if err := proto.Unmarshal(attributes, message); err != nil { + return fmt.Errorf("decoding value blob for %s: %v", entry.FullPath, err) + } + + FromPbEntryToExistingEntry(message, entry) + + // FromPbEntryToExistingEntry took the size over a chunk list that is not + // there, so fold in what the chunks actually reached. This is TotalSize. + if chunkExtent > entry.FileSize { + entry.FileSize = chunkExtent + } + + return nil +} + +// chunkExtentEnd reports where one encoded FileChunk ends, the offset plus size +// that TotalSize maximises over, without building the chunk. A chunk this +// rejects is one the full decoder rejects too, so a listing never reports a +// size for an entry that cannot be opened. +// +// A manifest chunk needs no special handling: it carries the offset and size of +// the whole range it stands for, and TotalSize does not resolve it either. +func chunkExtentEnd(chunk []byte) (uint64, error) { + var offset int64 + var size uint64 + for len(chunk) > 0 { + num, typ, tagLen := protowire.ConsumeTag(chunk) + if tagLen < 0 { + return 0, protowire.ParseError(tagLen) + } + chunk = chunk[tagLen:] + if typ == protowire.VarintType && (num == fileChunkOffsetField || num == fileChunkSizeField) { + v, n := protowire.ConsumeVarint(chunk) + if n < 0 { + return 0, protowire.ParseError(n) + } + if num == fileChunkOffsetField { + offset = int64(v) + } else { + size = v + } + chunk = chunk[n:] + continue + } + if typ == protowire.BytesType { + v, n := protowire.ConsumeBytes(chunk) + if n < 0 { + return 0, protowire.ParseError(n) + } + if fileChunkMessageFields[num] { + if err := validateMessage(v); err != nil { + return 0, err + } + } else if fileChunkStringFields[num] && !utf8.Valid(v) { + return 0, errors.New("invalid UTF-8 in string field") + } + chunk = chunk[n:] + continue + } + n := protowire.ConsumeFieldValue(num, typ, chunk) + if n < 0 { + return 0, protowire.ParseError(n) + } + chunk = chunk[n:] + } + return uint64(offset + int64(size)), nil +} + +// validateMessage walks an encoded message to check it parses, which is all the +// generated unmarshaller would do for one whose fields are scalars. +func validateMessage(b []byte) error { + for len(b) > 0 { + num, typ, tagLen := protowire.ConsumeTag(b) + if tagLen < 0 { + return protowire.ParseError(tagLen) + } + valLen := protowire.ConsumeFieldValue(num, typ, b[tagLen:]) + if valLen < 0 { + return protowire.ParseError(valLen) + } + b = b[tagLen+valLen:] + } + return nil +} diff --git a/weed/filer/entry_codec_attributes_test.go b/weed/filer/entry_codec_attributes_test.go new file mode 100644 index 000000000..34842a71e --- /dev/null +++ b/weed/filer/entry_codec_attributes_test.go @@ -0,0 +1,444 @@ +package filer + +import ( + "fmt" + "math/rand" + "os" + "reflect" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/reflect/protoreflect" +) + +// decodeBothWays round-trips entry and returns what each decoder made of it. +func decodeBothWays(t *testing.T, entry *Entry) (full, attrsOnly Entry) { + t.Helper() + blob, err := entry.EncodeAttributesAndChunks() + if err != nil { + t.Fatalf("encode: %v", err) + } + full.FullPath = entry.FullPath + if err := full.DecodeAttributesAndChunks(blob); err != nil { + t.Fatalf("full decode: %v", err) + } + attrsOnly.FullPath = entry.FullPath + if err := attrsOnly.DecodeAttributesOnly(blob); err != nil { + t.Fatalf("attributes-only decode: %v", err) + } + return full, attrsOnly +} + +// assertSameButChunks checks the two decoders agree on everything a listing +// reads. Chunks are the deliberate exception; size is not, because an entry can +// carry a zero FileSize and take its size from the chunks. +func assertSameButChunks(t *testing.T, full, attrsOnly Entry) { + t.Helper() + if attrsOnly.Chunks != nil { + t.Errorf("attributes-only decode built %d chunks, want none", len(attrsOnly.Chunks)) + } + if full.FileSize != attrsOnly.FileSize { + t.Errorf("FileSize = %d, want %d", attrsOnly.FileSize, full.FileSize) + } + if full.Size() != attrsOnly.Size() { + t.Errorf("Size() = %d, want %d", attrsOnly.Size(), full.Size()) + } + if !reflect.DeepEqual(full.Attr, attrsOnly.Attr) { + t.Errorf("Attr = %+v, want %+v", attrsOnly.Attr, full.Attr) + } + if string(full.HardLinkId) != string(attrsOnly.HardLinkId) { + t.Errorf("HardLinkId = %x, want %x", attrsOnly.HardLinkId, full.HardLinkId) + } + if full.HardLinkCounter != attrsOnly.HardLinkCounter { + t.Errorf("HardLinkCounter = %d, want %d", attrsOnly.HardLinkCounter, full.HardLinkCounter) + } + if string(full.Content) != string(attrsOnly.Content) { + t.Errorf("Content = %q, want %q", attrsOnly.Content, full.Content) + } + if full.Quota != attrsOnly.Quota { + t.Errorf("Quota = %d, want %d", attrsOnly.Quota, full.Quota) + } + if full.WORMEnforcedAtTsNs != attrsOnly.WORMEnforcedAtTsNs { + t.Errorf("WORMEnforcedAtTsNs = %d, want %d", attrsOnly.WORMEnforcedAtTsNs, full.WORMEnforcedAtTsNs) + } + if len(full.Extended) != len(attrsOnly.Extended) { + t.Errorf("Extended has %d keys, want %d", len(attrsOnly.Extended), len(full.Extended)) + } + for k, v := range full.Extended { + if string(attrsOnly.Extended[k]) != string(v) { + t.Errorf("Extended[%q] = %q, want %q", k, attrsOnly.Extended[k], v) + } + } + if (full.Remote == nil) != (attrsOnly.Remote == nil) { + t.Errorf("Remote presence differs: %v vs %v", attrsOnly.Remote != nil, full.Remote != nil) + } else if full.Remote != nil && full.Remote.RemoteSize != attrsOnly.Remote.RemoteSize { + t.Errorf("Remote.RemoteSize = %d, want %d", attrsOnly.Remote.RemoteSize, full.Remote.RemoteSize) + } +} + +func chunkAt(offset int64, size uint64, i int) *filer_pb.FileChunk { + return &filer_pb.FileChunk{ + FileId: fmt.Sprintf("3,01637037d6%04d", i), + Offset: offset, + Size: size, + ModifiedTsNs: int64(1700000000+i) * 1e9, + ETag: "1a2b3c4d5e6f7890", + Fid: &filer_pb.FileId{VolumeId: uint32(3 + i), FileKey: uint64(i), Cookie: 0x1637037d}, + CipherKey: []byte{1, 2, 3, 4}, + } +} + +func TestDecodeAttributesOnlyMatchesFullDecode(t *testing.T) { + now := time.Unix(1700000000, 123456789) + + cases := []struct { + name string + entry *Entry + }{ + {"no chunks", &Entry{ + FullPath: util.FullPath("/d/plain"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, Ctime: now, Uid: 99, Gid: 100, FileSize: 12}, + }}, + {"directory", &Entry{ + FullPath: util.FullPath("/d/sub"), + Attr: Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now, Uid: 99, Gid: 100}, + }}, + {"one chunk", &Entry{ + FullPath: util.FullPath("/d/one"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, Uid: 99, Gid: 100, FileSize: 4 << 20}, + Chunks: []*filer_pb.FileChunk{chunkAt(0, 4<<20, 0)}, + }}, + // The S3 copy and multipart paths deliberately store a zero FileSize and + // let the chunks define it, so this is the case that forces the extent + // walk rather than just skipping the field. + {"zero FileSize, size comes from chunks", &Entry{ + FullPath: util.FullPath("/d/zerosize"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, Uid: 99, Gid: 100, FileSize: 0}, + Chunks: []*filer_pb.FileChunk{ + chunkAt(0, 4<<20, 0), chunkAt(4<<20, 4<<20, 1), chunkAt(8<<20, 1234, 2), + }, + }}, + {"chunks out of order", &Entry{ + FullPath: util.FullPath("/d/unordered"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, Uid: 99, Gid: 100}, + Chunks: []*filer_pb.FileChunk{ + chunkAt(8<<20, 99, 0), chunkAt(0, 4<<20, 1), chunkAt(4<<20, 4<<20, 2), + }, + }}, + {"stored FileSize larger than chunks", &Entry{ + FullPath: util.FullPath("/d/sparse"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, Uid: 99, Gid: 100, FileSize: 1 << 30}, + Chunks: []*filer_pb.FileChunk{chunkAt(0, 16, 0)}, + }}, + {"symlink", &Entry{ + FullPath: util.FullPath("/d/link"), + Attr: Attr{Mode: os.ModeSymlink | 0o777, Mtime: now, Crtime: now, SymlinkTarget: "../target"}, + }}, + {"hard link", &Entry{ + FullPath: util.FullPath("/d/hard"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now}, + HardLinkId: HardLinkId([]byte{9, 8, 7, 6}), + HardLinkCounter: 3, + }}, + {"inline content", &Entry{ + FullPath: util.FullPath("/d/inline"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now}, + Content: []byte("hello world"), + }}, + {"extended attributes", &Entry{ + FullPath: util.FullPath("/d/xattr"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now}, + Extended: map[string][]byte{"a": []byte("1"), "b": []byte("2"), "Seaweed-X": []byte("y")}, + }}, + {"remote entry", &Entry{ + FullPath: util.FullPath("/d/remote"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 5}, + Remote: &filer_pb.RemoteEntry{RemoteSize: 4096, RemoteMtime: now.Unix() + 60, StorageName: "s3"}, + }}, + {"quota and worm", &Entry{ + FullPath: util.FullPath("/d/bucket"), + Attr: Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now}, + Quota: 1 << 40, + WORMEnforcedAtTsNs: now.UnixNano(), + }}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + full, attrsOnly := decodeBothWays(t, tc.entry) + assertSameButChunks(t, full, attrsOnly) + }) + } +} + +// TestDecodeAttributesOnlyRandomEntries fuzzes the field combinations, since the +// decoder walks the wire format by hand and has to stay in step with the +// generated one as filer.proto grows. +func TestDecodeAttributesOnlyRandomEntries(t *testing.T) { + rnd := rand.New(rand.NewSource(1)) + for i := 0; i < 2000; i++ { + now := time.Unix(1600000000+rnd.Int63n(1e8), rnd.Int63n(1e9)) + entry := &Entry{ + FullPath: util.FullPath(fmt.Sprintf("/d/f%d", i)), + Attr: Attr{ + Mode: os.FileMode(rnd.Intn(0o777)), + Mtime: now, Crtime: now, Ctime: now, + Uid: uint32(rnd.Intn(70000)), Gid: uint32(rnd.Intn(70000)), + FileSize: uint64(rnd.Int63n(1 << 34)), + Inode: rnd.Uint64(), + Rdev: uint32(rnd.Intn(1 << 20)), + TtlSec: int32(rnd.Intn(1000)), + }, + } + if rnd.Intn(2) == 0 { + entry.Attr.FileSize = 0 + } + if rnd.Intn(4) == 0 { + entry.Content = make([]byte, rnd.Intn(64)) + rnd.Read(entry.Content) + } + if rnd.Intn(4) == 0 { + entry.Extended = map[string][]byte{} + for k := 0; k < rnd.Intn(4); k++ { + entry.Extended[fmt.Sprintf("k%d", k)] = []byte(fmt.Sprintf("v%d", rnd.Intn(1000))) + } + } + if rnd.Intn(8) == 0 { + entry.Remote = &filer_pb.RemoteEntry{RemoteSize: rnd.Int63n(1 << 30), RemoteMtime: now.Unix() + int64(rnd.Intn(120)) - 60} + } + for c := 0; c < rnd.Intn(20); c++ { + entry.Chunks = append(entry.Chunks, chunkAt(rnd.Int63n(1<<30), uint64(rnd.Int63n(1<<22)), c)) + } + full, attrsOnly := decodeBothWays(t, entry) + assertSameButChunks(t, full, attrsOnly) + } +} + +func TestDecodeAttributesOnlyRejectsGarbage(t *testing.T) { + var entry Entry + entry.FullPath = util.FullPath("/d/bad") + if err := entry.DecodeAttributesOnly([]byte{0xff, 0xff, 0xff, 0xff}); err == nil { + t.Fatal("expected an error for a malformed blob") + } +} + +func BenchmarkDecode(b *testing.B) { + now := time.Unix(1700000000, 0) + for _, n := range []int{0, 1, 4, 16, 64} { + entry := &Entry{ + FullPath: util.FullPath("/images/image-00000001.jpg"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, Ctime: now, Uid: 99, Gid: 100, FileSize: uint64(n) * 4 << 20}, + } + for i := 0; i < n; i++ { + entry.Chunks = append(entry.Chunks, chunkAt(int64(i)*4<<20, 4<<20, i)) + } + blob, err := entry.EncodeAttributesAndChunks() + if err != nil { + b.Fatal(err) + } + b.Run(fmt.Sprintf("chunks=%d/decoder=full", n), func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + var out Entry + if err := out.DecodeAttributesAndChunks(blob); err != nil { + b.Fatal(err) + } + } + }) + b.Run(fmt.Sprintf("chunks=%d/decoder=attrsonly", n), func(b *testing.B) { + b.ReportAllocs() + for i := 0; i < b.N; i++ { + var out Entry + if err := out.DecodeAttributesOnly(blob); err != nil { + b.Fatal(err) + } + } + }) + } +} + +// TestProtoEntryWithoutChunksKeepsSize covers the wire contract the filer's +// omit_chunks listing relies on: the size a client needs survives in the +// attributes once the chunk list is dropped, including for the entries that +// store a zero FileSize and take their size from the chunks. +func TestProtoEntryWithoutChunksKeepsSize(t *testing.T) { + now := time.Unix(1700000000, 0) + for _, storedSize := range []uint64{0, 7, 1 << 30} { + stored := &Entry{ + FullPath: util.FullPath("/d/obj"), + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: storedSize}, + Chunks: []*filer_pb.FileChunk{chunkAt(0, 4<<20, 0), chunkAt(4<<20, 1234, 1)}, + } + blob, err := stored.EncodeAttributesAndChunks() + if err != nil { + t.Fatalf("encode: %v", err) + } + // What the filer holds after reading the entry out of its store. + var loaded Entry + loaded.FullPath = stored.FullPath + if err := loaded.DecodeAttributesAndChunks(blob); err != nil { + t.Fatalf("decode: %v", err) + } + want := FileSize(loaded.ToProtoEntry()) + + pbEntry := loaded.ToProtoEntry() + pbEntry.Chunks = nil + if got := FileSize(pbEntry); got != want { + t.Errorf("stored FileSize %d: size over the wire = %d, want %d", storedSize, got, want) + } + if got := FromPbEntry("/d", pbEntry).Size(); got != want { + t.Errorf("stored FileSize %d: size at the client = %d, want %d", storedSize, got, want) + } + } +} + +// TestChunkValidationCoversEveryField keeps the hand-rolled chunk walk honest as +// filer.proto grows. The generated unmarshaller never sees the chunk bytes, so +// every FileChunk field whose contents it would have checked -- a submessage, or +// a proto3 string's UTF-8 -- has to be listed for the walk to check instead. +func TestChunkValidationCoversEveryField(t *testing.T) { + fields := (&filer_pb.FileChunk{}).ProtoReflect().Descriptor().Fields() + for i := 0; i < fields.Len(); i++ { + f := fields.Get(i) + switch f.Kind() { + case protoreflect.MessageKind, protoreflect.GroupKind: + if !fileChunkMessageFields[protowire.Number(f.Number())] { + t.Errorf("FileChunk.%s (field %d) is a message but is not in fileChunkMessageFields, so a corrupt one would pass the listing decoder and fail the full one", f.Name(), f.Number()) + } + case protoreflect.StringKind: + if !fileChunkStringFields[protowire.Number(f.Number())] { + t.Errorf("FileChunk.%s (field %d) is a string but is not in fileChunkStringFields, so invalid UTF-8 would pass the listing decoder and fail the full one", f.Name(), f.Number()) + } + } + } +} + +// corruptChunkBlobs builds entry blobs whose chunk bytes are damaged in ways the +// generated unmarshaller rejects. +func corruptChunkBlobs(t *testing.T) map[string][]byte { + t.Helper() + now := time.Unix(1700000000, 0) + base := func() *filer_pb.FileChunk { + return &filer_pb.FileChunk{Offset: 0, Size: 1024, Fid: &filer_pb.FileId{VolumeId: 3, FileKey: 7, Cookie: 9}} + } + blobFor := func(mangle func(raw []byte) []byte) []byte { + chunk := base() + chunkBytes, err := proto.Marshal(chunk) + if err != nil { + t.Fatalf("marshal chunk: %v", err) + } + chunkBytes = mangle(chunkBytes) + var blob []byte + blob = protowire.AppendTag(blob, entryChunksField, protowire.BytesType) + blob = protowire.AppendBytes(blob, chunkBytes) + attrs, err := proto.Marshal(&filer_pb.FuseAttributes{FileSize: 0, Mtime: now.Unix(), FileMode: 0o644}) + if err != nil { + t.Fatalf("marshal attrs: %v", err) + } + blob = protowire.AppendTag(blob, 4, protowire.BytesType) + return protowire.AppendBytes(blob, attrs) + } + + out := map[string][]byte{} + // The exact probe from review: a nested fid whose payload is not a message. + out["corrupt nested fid"] = blobFor(func(raw []byte) []byte { + var b []byte + b = protowire.AppendTag(b, 2, protowire.VarintType) + b = protowire.AppendVarint(b, 0) + b = protowire.AppendTag(b, 3, protowire.VarintType) + b = protowire.AppendVarint(b, 1024) + b = protowire.AppendTag(b, 7, protowire.BytesType) + return protowire.AppendBytes(b, []byte{0xff, 0xff, 0xff, 0xff}) + }) + out["invalid utf8 in file_id"] = blobFor(func(raw []byte) []byte { + var b []byte + b = protowire.AppendTag(b, 1, protowire.BytesType) + b = protowire.AppendBytes(b, []byte{0xff, 0xfe, 0xfd}) + b = protowire.AppendTag(b, 3, protowire.VarintType) + return protowire.AppendVarint(b, 1024) + }) + out["truncated chunk"] = blobFor(func(raw []byte) []byte { return raw[:len(raw)-1] }) + return out +} + +// TestDecodeAttributesOnlyRejectsWhatFullDecodeRejects is the invariant that +// keeps a listing from showing a file that cannot then be opened: the fast path +// must never accept a blob the full decoder turns away. +func TestDecodeAttributesOnlyRejectsWhatFullDecodeRejects(t *testing.T) { + for name, blob := range corruptChunkBlobs(t) { + t.Run(name, func(t *testing.T) { + var full, attrsOnly Entry + full.FullPath = util.FullPath("/d/corrupt") + attrsOnly.FullPath = full.FullPath + fullErr := full.DecodeAttributesAndChunks(blob) + attrsErr := attrsOnly.DecodeAttributesOnly(blob) + if fullErr == nil { + t.Skip("full decoder accepts this blob, nothing to match") + } + if attrsErr == nil { + t.Errorf("full decode rejected the blob (%v) but attributes-only accepted it with FileSize=%d", fullErr, attrsOnly.FileSize) + } + }) + } +} + +// TestDecodeAttributesOnlyManifestChunk pins that a manifest chunk needs no +// resolving: it carries the offset and size of the range it stands for, and +// TotalSize does not resolve it either, so both decoders see one number. +func TestDecodeAttributesOnlyManifestChunk(t *testing.T) { + now := time.Unix(1700000000, 0) + manifest := chunkAt(0, 64<<20, 0) + manifest.IsChunkManifest = true + entry := &Entry{ + FullPath: util.FullPath("/d/big"), + // Zero stored size, so the manifest's extent is the only source. + Attr: Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 0}, + Chunks: []*filer_pb.FileChunk{manifest}, + } + full, attrsOnly := decodeBothWays(t, entry) + assertSameButChunks(t, full, attrsOnly) + if attrsOnly.FileSize != 64<<20 { + t.Errorf("FileSize = %d, want %d from the manifest extent", attrsOnly.FileSize, 64<<20) + } +} + +// TestDecodeAttributesOnlyFieldBeforeChunks exercises the prefix copy, which no +// other case reaches: EncodeAttributesAndChunks emits chunks before every field +// the other tests set, so `dropping` always turns on at pos 0 there. +func TestDecodeAttributesOnlyFieldBeforeChunks(t *testing.T) { + now := time.Unix(1700000000, 0) + attrs, err := proto.Marshal(&filer_pb.FuseAttributes{FileSize: 0, Mtime: now.Unix(), FileMode: 0o755}) + if err != nil { + t.Fatalf("marshal attrs: %v", err) + } + chunkBytes, err := proto.Marshal(chunkAt(0, 4<<20, 0)) + if err != nil { + t.Fatalf("marshal chunk: %v", err) + } + // is_directory (2) ahead of chunks (3), so the walk has a prefix to copy. + var blob []byte + blob = protowire.AppendTag(blob, 2, protowire.VarintType) + blob = protowire.AppendVarint(blob, 1) + blob = protowire.AppendTag(blob, entryChunksField, protowire.BytesType) + blob = protowire.AppendBytes(blob, chunkBytes) + blob = protowire.AppendTag(blob, 4, protowire.BytesType) + blob = protowire.AppendBytes(blob, attrs) + + var full, attrsOnly Entry + full.FullPath = util.FullPath("/d/dirwithchunks") + attrsOnly.FullPath = full.FullPath + if err := full.DecodeAttributesAndChunks(blob); err != nil { + t.Fatalf("full decode: %v", err) + } + if err := attrsOnly.DecodeAttributesOnly(blob); err != nil { + t.Fatalf("attributes-only decode: %v", err) + } + assertSameButChunks(t, full, attrsOnly) + if attrsOnly.FileSize != 4<<20 { + t.Errorf("FileSize = %d, want %d", attrsOnly.FileSize, 4<<20) + } +} diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index 3e001b6df..a6d699d8b 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/weed/filer/leveldb/leveldb_store.go @@ -237,7 +237,7 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir entry := &filer.Entry{ FullPath: weed_util.NewFullPath(string(dirPath), fileName), } - if decodeErr := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(iter.Value())); decodeErr != nil { + if decodeErr := filer.DecodeListedEntry(ctx, entry, weed_util.MaybeDecompressData(iter.Value())); decodeErr != nil { err = decodeErr glog.V(0).InfofCtx(ctx, "list %s : %v", entry.FullPath, err) break diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index 0949a2e0f..06a637047 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/weed/mount/weedfs_dir_read.go @@ -20,6 +20,11 @@ const ( batchSize = 1000 ) +// readdirContext marks the meta cache listing as reading attributes only. A +// readdir never looks at a chunk list, and building one per child is most of +// the cost of decoding a wide directory. +var readdirContext = filer_pb.WithChunksOmitted(context.Background()) + // DirectoryHandle represents an open directory handle. // It maintains state for directory listing pagination and is protected by a mutex // to handle concurrent readdir operations from NFS-Ganesha and other multi-threaded clients. @@ -247,7 +252,7 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode } // Load entries from beginning to fill cache up to the requested offset - loadErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) { + loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) { dh.entryStream = append(dh.entryStream, entry) return true, nil }) @@ -284,7 +289,7 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode // Batch loading: fetch batchSize entries starting from lastEntryName loadedCount := 0 bufferFull := false - loadErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) { + loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) { currentIndex := int64(len(dh.entryStream)) dh.entryStream = append(dh.entryStream, entry) loadedCount++ @@ -315,7 +320,7 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di if input.Offset >= dh.entryStreamOffset { if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset { skipCount := uint32(input.Offset-dh.entryStreamOffset) + batchSize - entries, snapshotTs, err := loadDirectoryEntriesDirect(context.Background(), wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) + entries, snapshotTs, err := loadDirectoryEntriesDirect(readdirContext, wfs, wfs.option.UidGidMapper, dirPath, "", false, skipCount, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) if err != nil { glog.Errorf("list filer directory: %v", err) return fuse.EIO @@ -344,7 +349,7 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di } } - entries, snapshotTs, err := loadDirectoryEntriesDirect(context.Background(), wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) + entries, snapshotTs, err := loadDirectoryEntriesDirect(readdirContext, wfs, wfs.option.UidGidMapper, dirPath, lastEntryName, false, batchSize, dh.snapshotTsNs, wfs.option.IncludeSystemEntries) if err != nil { glog.Errorf("list filer directory: %v", err) return fuse.EIO diff --git a/weed/mount/weedfs_dir_read_bench_test.go b/weed/mount/weedfs_dir_read_bench_test.go index 0681f72d1..a78da820d 100644 --- a/weed/mount/weedfs_dir_read_bench_test.go +++ b/weed/mount/weedfs_dir_read_bench_test.go @@ -12,6 +12,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -126,7 +127,17 @@ func newBenchWFS(tb testing.TB, dir util.FullPath, n int) *WFS { FullPath: child, // The filer stamps an inode on every entry it stores, so a listing // arrives with one and never has to derive its own. - Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, Uid: 99, Gid: 100, FileSize: 4096, Inode: child.AsInode(now.Unix())}, + Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, Uid: 99, Gid: 100, FileSize: 4 << 20, Inode: child.AsInode(now.Unix())}, + // A real file has chunks, and building them is most of what decoding + // an entry costs. + // No FileId: BeforeEntrySerialization reparses that legacy string + // over Fid on the way in, which would make every entry's chunk + // byte-identical instead of varying per file. + Chunks: []*filer_pb.FileChunk{{ + Size: 4 << 20, ModifiedTsNs: now.UnixNano(), + ETag: "1a2b3c4d5e6f7890", + Fid: &filer_pb.FileId{VolumeId: 3, FileKey: uint64(i), Cookie: 0x1637037d}, + }}, }, 0); err != nil { tb.Fatalf("insert entry %d: %v", i, err) } diff --git a/weed/mount/weedfs_dir_read_chunks_test.go b/weed/mount/weedfs_dir_read_chunks_test.go new file mode 100644 index 000000000..bf12e0e35 --- /dev/null +++ b/weed/mount/weedfs_dir_read_chunks_test.go @@ -0,0 +1,55 @@ +package mount + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// TestListDirectoryEntriesOmitsChunks covers the wiring the readdir speedup +// rests on: that the context marker actually reaches the store's decode. Every +// other test exercises the decoder directly, so a refactor that stopped +// threading the context would revert the optimisation silently. +func TestListDirectoryEntriesOmitsChunks(t *testing.T) { + dir := util.FullPath("/images") + wfs := newBenchWFS(t, dir, 4) + + for _, tc := range []struct { + name string + ctx context.Context + wantChunks bool + }{ + {"plain listing keeps chunks", context.Background(), true}, + {"marked listing drops chunks", filer_pb.WithChunksOmitted(context.Background()), false}, + } { + t.Run(tc.name, func(t *testing.T) { + var seen int + err := wfs.metaCache.ListDirectoryEntries(tc.ctx, dir, "", false, 100, func(entry *filer.Entry) (bool, error) { + seen++ + if got := len(entry.Chunks) > 0; got != tc.wantChunks { + t.Errorf("%s: has chunks = %v, want %v", entry.Name(), got, tc.wantChunks) + } + // The size has to survive either way, since that is what the + // readdir reports. + if entry.FileSize != 4<<20 { + t.Errorf("%s: FileSize = %d, want %d", entry.Name(), entry.FileSize, 4<<20) + } + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if seen != 4 { + t.Fatalf("listed %d entries, want 4", seen) + } + }) + } + + // readdirContext is what weedfs_dir_read.go actually passes. + if !filer_pb.ChunksOmitted(readdirContext) { + t.Error("readdirContext does not carry the chunks-omitted marker") + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index bf728771f..b8beefd71 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -136,6 +136,10 @@ message ListEntriesRequest { bool inclusiveStartFrom = 4; uint32 limit = 5; int64 snapshot_ts_ns = 6; + // Leave the chunk list out of every entry in the response. The attributes + // still carry the file size, so a listing that only reads attributes can + // ask for this and skip the largest part of the payload. + bool omit_chunks = 7; } message ListEntriesResponse { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 0ec399f48..a00719891 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -440,8 +440,12 @@ type ListEntriesRequest struct { InclusiveStartFrom bool `protobuf:"varint,4,opt,name=inclusiveStartFrom,proto3" json:"inclusiveStartFrom,omitempty"` Limit uint32 `protobuf:"varint,5,opt,name=limit,proto3" json:"limit,omitempty"` SnapshotTsNs int64 `protobuf:"varint,6,opt,name=snapshot_ts_ns,json=snapshotTsNs,proto3" json:"snapshot_ts_ns,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Leave the chunk list out of every entry in the response. The attributes + // still carry the file size, so a listing that only reads attributes can + // ask for this and skip the largest part of the payload. + OmitChunks bool `protobuf:"varint,7,opt,name=omit_chunks,json=omitChunks,proto3" json:"omit_chunks,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ListEntriesRequest) Reset() { @@ -516,6 +520,13 @@ func (x *ListEntriesRequest) GetSnapshotTsNs() int64 { return 0 } +func (x *ListEntriesRequest) GetOmitChunks() bool { + if x != nil { + return x.OmitChunks + } + return false +} + type ListEntriesResponse struct { state protoimpl.MessageState `protogen:"open.v1"` Entry *Entry `protobuf:"bytes,1,opt,name=entry,proto3" json:"entry,omitempty"` @@ -6939,14 +6950,16 @@ const file_filer_proto_rawDesc = "" + "\x1cLookupDirectoryEntryResponse\x12%\n" + "\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12\x1a\n" + "\tlog_ts_ns\x18\x02 \x01(\x03R\alogTsNs\x12#\n" + - "\rlog_signature\x18\x03 \x01(\x05R\flogSignature\"\xe4\x01\n" + + "\rlog_signature\x18\x03 \x01(\x05R\flogSignature\"\x85\x02\n" + "\x12ListEntriesRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12\x16\n" + "\x06prefix\x18\x02 \x01(\tR\x06prefix\x12,\n" + "\x11startFromFileName\x18\x03 \x01(\tR\x11startFromFileName\x12.\n" + "\x12inclusiveStartFrom\x18\x04 \x01(\bR\x12inclusiveStartFrom\x12\x14\n" + "\x05limit\x18\x05 \x01(\rR\x05limit\x12$\n" + - "\x0esnapshot_ts_ns\x18\x06 \x01(\x03R\fsnapshotTsNs\"b\n" + + "\x0esnapshot_ts_ns\x18\x06 \x01(\x03R\fsnapshotTsNs\x12\x1f\n" + + "\vomit_chunks\x18\a \x01(\bR\n" + + "omitChunks\"b\n" + "\x13ListEntriesResponse\x12%\n" + "\x05entry\x18\x01 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x12$\n" + "\x0esnapshot_ts_ns\x18\x02 \x01(\x03R\fsnapshotTsNs\"\xa1\x02\n" + diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go index 93f86a2ff..839c8099a 100644 --- a/weed/pb/filer_pb/filer_client.go +++ b/weed/pb/filer_pb/filer_client.go @@ -139,6 +139,7 @@ func DoSeaweedListWithSnapshot(ctx context.Context, client SeaweedFilerClient, f Limit: redLimit, InclusiveStartFrom: inclusive, SnapshotTsNs: snapshotTsNs, + OmitChunks: ChunksOmitted(ctx), } // Preserve the caller-requested snapshot so pagination uses the same diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index f4771b211..efcfa0819 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -149,6 +149,16 @@ func (m *ListEntriesRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.OmitChunks { + i-- + if m.OmitChunks { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x38 + } if m.SnapshotTsNs != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.SnapshotTsNs)) i-- @@ -6285,6 +6295,9 @@ func (m *ListEntriesRequest) SizeVT() (n int) { if m.SnapshotTsNs != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.SnapshotTsNs)) } + if m.OmitChunks { + n += 2 + } n += len(m.unknownFields) return n } @@ -9140,6 +9153,26 @@ func (m *ListEntriesRequest) UnmarshalVT(dAtA []byte) error { break } } + case 7: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field OmitChunks", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.OmitChunks = bool(v != 0) default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/pb/filer_pb/listing_options.go b/weed/pb/filer_pb/listing_options.go new file mode 100644 index 000000000..b25ba9e42 --- /dev/null +++ b/weed/pb/filer_pb/listing_options.go @@ -0,0 +1,22 @@ +package filer_pb + +import "context" + +type omitChunksKey struct{} + +// WithChunksOmitted marks a listing as wanting attributes only. A listing over +// gRPC then asks the filer to leave the chunk lists out, and a listing served +// from a local store skips building them. The file size is carried in the +// attributes either way. +// +// Only a caller that reads attributes and nothing else may use this. In +// particular a listing that populates a cache, or one whose entries can be +// written back or deleted, needs the chunks. +func WithChunksOmitted(ctx context.Context) context.Context { + return context.WithValue(ctx, omitChunksKey{}, true) +} + +// ChunksOmitted reports whether the listing wants attributes only. +func ChunksOmitted(ctx context.Context) bool { + return ctx.Value(omitChunksKey{}) != nil +} diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index ed15fabcc..0e78b7046 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -86,8 +86,21 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file var hasEntries bool lastFileName, listErr = fs.filer.StreamListDirectoryEntries(stream.Context(), util.FullPath(req.Directory), lastFileName, includeLastFile, int64(paginationLimit), req.Prefix, "", "", func(entry *filer.Entry) (bool, error) { hasEntries = true + pbEntry := entry.ToProtoEntry() + if req.OmitChunks { + // Stamp the size before dropping the only other thing carrying + // it. Most stores fold the chunk extents into FileSize when they + // decode, but one that keeps entries as JSON rather than as an + // encoded Entry never re-derives it, and the caller would be + // left with a zero. The entries are still read whole, because + // expiring one here deletes its data and that needs the chunks. + if pbEntry.Attributes != nil { + pbEntry.Attributes.FileSize = filer.FileSize(pbEntry) + } + pbEntry.Chunks = nil + } resp := &filer_pb.ListEntriesResponse{ - Entry: entry.ToProtoEntry(), + Entry: pbEntry, } if !sentSnapshot { resp.SnapshotTsNs = snapshotTsNs