diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 9c440b281..6095815db 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -501,7 +501,7 @@ func (ma *MetaAggregator) doSubscribeToOneFiler(f *Filer, self pb.ServerAddress, return processOne(event) }) if readErr != nil { - return fmt.Errorf("read log file refs from %s: %w", peer, readErr) + return fmt.Errorf("%w from %s: %w", pb.ErrLogFileRead, peer, readErr) } if lastTs > 0 { lastTsNs = lastTs diff --git a/weed/mount/meta_cache/meta_cache_subscribe.go b/weed/mount/meta_cache/meta_cache_subscribe.go index 18b35446c..added6f46 100644 --- a/weed/mount/meta_cache/meta_cache_subscribe.go +++ b/weed/mount/meta_cache/meta_cache_subscribe.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/wdclient" ) type MetadataFollower struct { @@ -44,7 +45,7 @@ func mergeProcessors(mainProcessor func(resp *filer_pb.SubscribeMetadataResponse } } -func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, dir string, lastTsNs int64, skipSelfEvents bool, onRetry func(lastTsNs int64, err error), followers ...*MetadataFollower) error { +func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.FilerClient, lookupFn wdclient.LookupFileIdFunctionType, dir string, lastTsNs int64, skipSelfEvents bool, onRetry func(lastTsNs int64, err error), followers ...*MetadataFollower) error { var prefixes []string for _, follower := range followers { @@ -68,9 +69,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil prefix = prefix + "/" } - // Read persisted log chunks directly from volume servers, keeping the replay - // cost off the filer's heap (see LogFileReaderFn below). - lookupFn := filer.LookupFn(client) + // Replaying the persisted log chunks here keeps the cost off the filer's heap. + // The caller's lookup says where to read them from, so a mount that cannot + // reach volume servers directly still replays through its filer. metadataFollowOption := &pb.MetadataFollowOption{ ClientName: "mount", ClientId: selfSignature, diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 63f89cab9..79f897c6a 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -493,7 +493,7 @@ func (wfs *WFS) StartBackgroundTasks() error { } startTime := time.Now() - go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.option.FilerMountRootPath, startTime.UnixNano(), wfs.option.WritebackCache, func(lastTsNs int64, err error) { + go meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.LookupFn(), wfs.option.FilerMountRootPath, startTime.UnixNano(), wfs.option.WritebackCache, func(lastTsNs int64, err error) { glog.Warningf("meta events follow retry from %v: %v", time.Unix(0, lastTsNs), err) // A subscription gap may have dropped events, so distrust every cached // listing. Reset the flags first (safe — it never deletes entries), then diff --git a/weed/mount/weedfs_meta_subscribe_test.go b/weed/mount/weedfs_meta_subscribe_test.go new file mode 100644 index 000000000..241d98df1 --- /dev/null +++ b/weed/mount/weedfs_meta_subscribe_test.go @@ -0,0 +1,184 @@ +package mount + +import ( + "bytes" + "context" + "net" + "net/http" + "net/http/httptest" + "net/url" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/protobuf/proto" + + "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" +) + +// proxyFilerServer replays the previous minute's metadata as a persisted log +// chunk, the way a filer answers every fresh subscription, and points volume +// lookups at an address nothing listens on: the mount host's view of a volume +// server it is meant to reach only through -volumeServerAccess=filerProxy. +type proxyFilerServer struct { + filer_pb.UnimplementedSeaweedFilerServer + chunk *filer_pb.FileChunk + subscriptions atomic.Int32 + volumeLookups atomic.Int32 + stop chan struct{} +} + +func (s *proxyFilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error { + if s.subscriptions.Add(1) > 1 { + <-s.stop + return nil + } + return stream.Send(&filer_pb.SubscribeMetadataResponse{ + LogFileRefs: []*filer_pb.LogFileChunkRef{{ + FilerId: "5319c0e8", + FileTsNs: time.Now().UnixNano(), + Chunks: []*filer_pb.FileChunk{s.chunk}, + }}, + }) +} + +func (s *proxyFilerServer) LookupVolume(ctx context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + s.volumeLookups.Add(1) + locations := make(map[string]*filer_pb.Locations) + for _, vid := range req.VolumeIds { + locations[vid] = &filer_pb.Locations{Locations: []*filer_pb.Location{{Url: "127.0.0.1:1", PublicUrl: "127.0.0.1:1"}}} + } + return &filer_pb.LookupVolumeResponse{LocationsMap: locations}, nil +} + +// buildLogFileData writes the on-disk log format: [4-byte size | LogEntry]. +func buildLogFileData(events ...*filer_pb.SubscribeMetadataResponse) []byte { + var buf bytes.Buffer + for _, event := range events { + eventData, _ := proto.Marshal(event) + entryData, _ := proto.Marshal(&filer_pb.LogEntry{TsNs: event.TsNs, Data: eventData, Key: []byte(event.Directory)}) + sizeBuf := make([]byte, 4) + util.Uint32toBytes(sizeBuf, uint32(len(entryData))) + buf.Write(sizeBuf) + buf.Write(entryData) + } + return buf.Bytes() +} + +// TestSubscribeMetaEventsReplaysThroughFilerProxy pins the replay to the lookup +// the mount was given: resolving volume servers here fails every subscription +// of a filerProxy mount, which then resubscribes a second later, forever. +func TestSubscribeMetaEventsReplaysThroughFilerProxy(t *testing.T) { + logData := buildLogFileData(&filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: time.Now().UnixNano(), + EventNotification: &filer_pb.EventNotification{ + NewEntry: &filer_pb.Entry{Name: "file", Attributes: &filer_pb.FuseAttributes{FileSize: 7}}, + }, + }) + + var proxiedChunkIds atomic.Value + proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fileId := r.URL.Query().Get("proxyChunkId") + if fileId == "" { + http.Error(w, "not a filer proxy read", http.StatusBadRequest) + return + } + proxiedChunkIds.Store(fileId) + w.Write(logData) + })) + t.Cleanup(proxy.Close) + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + t.Cleanup(func() { _ = listener.Close() }) + + testServer := &proxyFilerServer{ + chunk: &filer_pb.FileChunk{FileId: "4471,ce598bfff8416e", Size: uint64(len(logData))}, + stop: make(chan struct{}), + } + server := pb.NewGrpcServer() + filer_pb.RegisterSeaweedFilerServer(server, testServer) + go server.Serve(listener) + t.Cleanup(func() { + close(testServer.stop) + server.Stop() + }) + + proxyUrl, err := url.Parse(proxy.URL) + if err != nil { + t.Fatalf("parse proxy url: %v", err) + } + + uidGidMapper, err := meta_cache.NewUidGidMapper("", "") + if err != nil { + t.Fatalf("create uid/gid mapper: %v", err) + } + root := util.FullPath("/") + wfs := &WFS{ + signature: 1, + inodeToPath: NewInodeToPath(root, 0), + fhMap: NewFileHandleToInode(), + fhLockTable: util.NewLockTable[FileHandleId](), + hardLinkLockTable: util.NewLockTable[string](), + option: &Option{ + ChunkSizeLimit: 1024, + ConcurrentReaders: 1, + VolumeServerAccess: "filerProxy", + FilerMountRootPath: "/", + FilerAddresses: []pb.ServerAddress{ + pb.NewServerAddressWithGrpcPort(proxyUrl.Host, listener.Addr().(*net.TCPAddr).Port), + }, + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + }, + } + wfs.metaCache = meta_cache.NewMetaCache( + filepath.Join(t.TempDir(), "meta"), + uidGidMapper, + root, + false, + func(path util.FullPath) { wfs.inodeToPath.MarkChildrenCached(path) }, + func(path util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(path) }, + func(meta_cache.EntryInvalidation) {}, + nil, + ) + t.Cleanup(wfs.metaCache.Shutdown) + wfs.inodeToPath.MarkChildrenCached(root) + wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false) + wfs.inodeToPath.MarkChildrenCached(util.FullPath("/dir")) + + followed := make(chan struct{}) + go func() { + defer close(followed) + meta_cache.SubscribeMetaEvents(wfs.metaCache, wfs.signature, wfs, wfs.LookupFn(), wfs.option.FilerMountRootPath, 0, false, nil) + }() + + select { + case <-followed: + case <-time.After(20 * time.Second): + t.Fatal("the log chunk replay never succeeded") + } + + if got := testServer.volumeLookups.Load(); got != 0 { + t.Errorf("filerProxy mount resolved %d volume locations, want 0", got) + } + if got, _ := proxiedChunkIds.Load().(string); got != testServer.chunk.FileId { + t.Errorf("filer proxy served chunk %q, want %q", got, testServer.chunk.FileId) + } + + entry, _, err := wfs.metaCache.FindEntry(context.Background(), util.FullPath("/dir/file")) + if err != nil { + t.Fatalf("replayed entry: %v", err) + } + if entry.Attr.FileSize != 7 { + t.Errorf("replayed file size %d, want 7", entry.Attr.FileSize) + } +} diff --git a/weed/pb/filer_pb_direct_read.go b/weed/pb/filer_pb_direct_read.go index 0011d82d7..ab9b58f30 100644 --- a/weed/pb/filer_pb_direct_read.go +++ b/weed/pb/filer_pb_direct_read.go @@ -25,6 +25,11 @@ const logEntryChannelSize = 512 // prefix, mirroring the filer package's unexported constant. const maxLogEntrySize = 1 << 30 +// ErrLogFileRead marks a failure to read persisted log chunks from volume +// servers. It surfaces as the metadata stream's error, but says nothing about +// the filer connection that carried the stream. +var ErrLogFileRead = errors.New("read log file refs") + // errReaderStopped signals that the entry consumer asked to stop (the merge // loop aborted or the caller hit a processing error). It is not a read failure. var errReaderStopped = errors.New("log entry reader stopped") diff --git a/weed/pb/filer_pb_tail.go b/weed/pb/filer_pb_tail.go index 4612a91d4..68538d7f4 100644 --- a/weed/pb/filer_pb_tail.go +++ b/weed/pb/filer_pb_tail.go @@ -154,7 +154,7 @@ func makeSubscribeMetadataFunc(option *MetadataFollowOption, processEventFn Proc }, processEventFn) if readErr != nil { - return fmt.Errorf("read log file refs: %w", readErr) + return fmt.Errorf("%w: %w", ErrLogFileRead, readErr) } if lastTs > 0 { option.StartTsNs = lastTs diff --git a/weed/pb/grpc_client_cascade_test.go b/weed/pb/grpc_client_cascade_test.go index 33acaa5e0..9b94d239b 100644 --- a/weed/pb/grpc_client_cascade_test.go +++ b/weed/pb/grpc_client_cascade_test.go @@ -2,9 +2,13 @@ package pb import ( "context" + "errors" + "io" "net" + "net/http" "sync" "testing" + "time" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -104,7 +108,16 @@ func TestWithGrpcClient_AbandonedRequestKeepsSharedConnection(t *testing.T) { } func (s *cascadeFilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, stream filer_pb.SeaweedFiler_SubscribeMetadataServer) error { - return nil + if !req.ClientSupportsMetadataChunks { + return nil + } + return stream.Send(&filer_pb.SubscribeMetadataResponse{ + LogFileRefs: []*filer_pb.LogFileChunkRef{{ + FilerId: "5319c0e8", + FileTsNs: time.Now().UnixNano(), + Chunks: []*filer_pb.FileChunk{{FileId: "4471,ce598bfff8416e", Size: 64}}, + }}, + }) } // TestWithGrpcClient_EndedStreamKeepsSharedConnection covers the other half of @@ -145,3 +158,52 @@ func TestWithGrpcClient_EndedStreamKeepsSharedConnection(t *testing.T) { t.Fatalf("concurrent caller must survive an unrelated stream ending: %v", concurrentErr) } } + +// TestWithGrpcClient_LogChunkReadFailureKeepsSharedConnection covers a +// subscriber replaying persisted log chunks over HTTP: an unreachable volume +// server reports "connection refused" as the subscription's error, which used +// to drop the shared filer ClientConn and cancel the assign and upload RPCs +// riding on it. +func TestWithGrpcClient_LogChunkReadFailureKeepsSharedConnection(t *testing.T) { + fake, address, dialOption := startCascadeFiler(t) + + var concurrentErr error + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + concurrentErr = WithGrpcClient(context.Background(), false, 0, func(connection *grpc.ClientConn) error { + _, err := filer_pb.NewSeaweedFilerClient(connection).LookupDirectoryEntry(context.Background(), + &filer_pb.LookupDirectoryEntryRequest{Directory: "/buckets/b", Name: "slow"}) + return err + }, address, false, dialOption) + }() + <-fake.inFlight + + option := &MetadataFollowOption{ + ClientName: "mount", + PathPrefix: "/", + EventErrorType: DontLogError, + LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + // An unreachable volume server, as a filerProxy mount sees it. + response, err := http.Get("http://127.0.0.1:1/" + chunks[0].FileId) + if err != nil { + return nil, err + } + return response.Body, nil + }, + } + subscribe := makeSubscribeMetadataFunc(option, func(resp *filer_pb.SubscribeMetadataResponse) error { return nil }) + err := WithGrpcClient(context.Background(), true, 0, func(connection *grpc.ClientConn) error { + return subscribe(filer_pb.NewSeaweedFilerClient(connection)) + }, address, false, dialOption) + if err == nil || !errors.Is(err, ErrLogFileRead) { + t.Fatalf("the replay should have failed reading the log chunk: %v", err) + } + + close(fake.release) + wg.Wait() + if concurrentErr != nil { + t.Fatalf("concurrent caller must survive a log chunk read failure: %v", concurrentErr) + } +} diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index cd1a2a81f..b508effb4 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -401,6 +401,13 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { return false } + // A metadata subscriber reads log chunks over HTTP from volume servers and + // returns what went wrong there through the stream. Those failures read like + // transport failures below, and would drop a filer channel that is fine. + if errors.Is(err, ErrLogFileRead) { + return false + } + // gRPC raises this locally, before the RPC reaches the wire, when this // process already closed the ClientConn. The caller is a bystander of // someone else's teardown, and knows nothing about the peer. Its message