admin: add connected Mount Clients page and dashboard section (#9968)

* admin: add connected mount clients page and dashboard section

The filer is the authority on who is subscribed to its metadata stream
(FUSE/VFS mounts, S3, peer filers, ...), but its in-memory listener
registry only tracked clientId->epoch and was not exposed.

- Enrich the filer subscriber registry with name/type/address/path/
  connected-time, populated in addClient and cleared in deleteClient so
  it reflects currently-connected clients only.
- Add a ListMetadataSubscribers filer gRPC (optional client-type filter).
- Admin server fans out to every filer, filters to mount types
  ("mount" Go weed mount, "sw-vfs" Rust VFS), and renders a new
  Cluster > Mount Clients page plus a Mount Clients dashboard section.

Read-only; no behavior change to the subscribe hot path.

* admin: address review — parallelize filer fan-out, guard nil map, robust CSV

- GetMountClients now queries filers concurrently, each under a 5s
  timeout, so a slow/unreachable filer can't stall the admin dashboard.
- Defensively initialize fs.subscribers before first write.
- Mount Clients CSV export uses a Blob with quote-escaping instead of a
  data: URI, so special characters in paths export correctly.

weed/server/filer_grpc_server_sub_meta.go

@@ -165,7 +165,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest,
 	ctx := stream.Context()
 	peerAddress := findClientAddress(ctx, 0)
 
-	isReplacing, alreadyKnown, clientName := fs.addClient("", req.ClientName, peerAddress, req.ClientId, req.ClientEpoch)
+	isReplacing, alreadyKnown, clientName := fs.addClient("", req.ClientName, peerAddress, req.PathPrefix, req.ClientId, req.ClientEpoch)
 	if isReplacing {
 		fs.filer.MetaAggregator.ListenersCond.Broadcast() // nudges the subscribers that are waiting
 	} else if alreadyKnown {
@@ -314,7 +314,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq
 	// use negative client id to differentiate from addClient()/deleteClient() used in SubscribeMetadata()
 	req.ClientId = -req.ClientId
 
-	isReplacing, alreadyKnown, clientName := fs.addClient("local", req.ClientName, peerAddress, req.ClientId, req.ClientEpoch)
+	isReplacing, alreadyKnown, clientName := fs.addClient("local", req.ClientName, peerAddress, req.PathPrefix, req.ClientId, req.ClientEpoch)
 	if isReplacing {
 		fs.listenersCond.Broadcast() // nudges the subscribers that are waiting
 	} else if alreadyKnown {
@@ -669,9 +669,9 @@ func (fs *FilerServer) eachEventNotificationFn(req *filer_pb.SubscribeMetadataRe
 	}
 }
 
-func (fs *FilerServer) addClient(prefix string, clientType string, clientAddress string, clientId int32, clientEpoch int32) (isReplacing, alreadyKnown bool, clientName string) {
+func (fs *FilerServer) addClient(scope string, clientType string, clientAddress string, pathPrefix string, clientId int32, clientEpoch int32) (isReplacing, alreadyKnown bool, clientName string) {
 	clientName = clientType + "@" + clientAddress
-	glog.V(0).Infof("+ %v listener %v clientId %v clientEpoch %v", prefix, clientName, clientId, clientEpoch)
+	glog.V(0).Infof("+ %v listener %v clientId %v clientEpoch %v", scope, clientName, clientId, clientEpoch)
 	if clientId != 0 {
 		fs.knownListenersLock.Lock()
 		defer fs.knownListenersLock.Unlock()
@@ -679,6 +679,18 @@ func (fs *FilerServer) addClient(prefix string, clientType string, clientAddress
 		if !found || epoch < clientEpoch {
 			fs.knownListeners[clientId] = clientEpoch
 			isReplacing = true
+			if fs.subscribers == nil {
+				fs.subscribers = make(map[int32]*metadataSubscriber)
+			}
+			fs.subscribers[clientId] = &metadataSubscriber{
+				clientName:    clientName,
+				clientType:    clientType,
+				address:       clientAddress,
+				pathPrefix:    pathPrefix,
+				clientId:      clientId,
+				clientEpoch:   clientEpoch,
+				connectedAtNs: time.Now().UnixNano(),
+			}
 		} else {
 			alreadyKnown = true
 		}
@@ -686,14 +698,15 @@ func (fs *FilerServer) addClient(prefix string, clientType string, clientAddress
 	return
 }
 
-func (fs *FilerServer) deleteClient(prefix string, clientName string, clientId int32, clientEpoch int32) {
-	glog.V(0).Infof("- %v listener %v clientId %v clientEpoch %v", prefix, clientName, clientId, clientEpoch)
+func (fs *FilerServer) deleteClient(scope string, clientName string, clientId int32, clientEpoch int32) {
+	glog.V(0).Infof("- %v listener %v clientId %v clientEpoch %v", scope, clientName, clientId, clientEpoch)
 	if clientId != 0 {
 		fs.knownListenersLock.Lock()
 		defer fs.knownListenersLock.Unlock()
 		epoch, found := fs.knownListeners[clientId]
 		if found && epoch <= clientEpoch {
 			delete(fs.knownListeners, clientId)
+			delete(fs.subscribers, clientId)
 		}
 	}
 }

weed/server/filer_grpc_server_subscribers.go

@@ -0,0 +1,63 @@
+package weed_server
+
+import (
+	"context"
+	"sort"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+// metadataSubscriber is the live record of one connected metadata subscriber
+// (a FUSE mount, S3, filer.sync, peer filer, ...). It is created in addClient
+// and removed in deleteClient, so the registry reflects currently-connected
+// clients only. Guarded by FilerServer.knownListenersLock.
+type metadataSubscriber struct {
+	clientName    string // "<type>@<address>"
+	clientType    string // e.g. "mount", "sw-vfs", "s3", "filer:<addr>"
+	address       string
+	pathPrefix    string
+	clientId      int32
+	clientEpoch   int32
+	connectedAtNs int64
+}
+
+// ListMetadataSubscribers returns the metadata subscribers currently connected
+// to this filer, optionally filtered by client type. Each filer only knows its
+// own subscribers, so cluster-wide callers query every filer and aggregate.
+func (fs *FilerServer) ListMetadataSubscribers(ctx context.Context, req *filer_pb.ListMetadataSubscribersRequest) (*filer_pb.ListMetadataSubscribersResponse, error) {
+	typeFilter := make(map[string]bool, len(req.ClientTypes))
+	for _, t := range req.ClientTypes {
+		typeFilter[t] = true
+	}
+
+	filerAddress := string(fs.option.Host)
+
+	fs.knownListenersLock.Lock()
+	subs := make([]*filer_pb.MetadataSubscriber, 0, len(fs.subscribers))
+	for _, s := range fs.subscribers {
+		if len(typeFilter) > 0 && !typeFilter[s.clientType] {
+			continue
+		}
+		subs = append(subs, &filer_pb.MetadataSubscriber{
+			ClientName:    s.clientName,
+			ClientType:    s.clientType,
+			Address:       s.address,
+			PathPrefix:    s.pathPrefix,
+			ClientId:      s.clientId,
+			ClientEpoch:   s.clientEpoch,
+			ConnectedAtNs: s.connectedAtNs,
+			FilerAddress:  filerAddress,
+		})
+	}
+	fs.knownListenersLock.Unlock()
+
+	// Stable ordering so the admin page doesn't shuffle rows between refreshes.
+	sort.Slice(subs, func(i, j int) bool {
+		if subs[i].ClientName != subs[j].ClientName {
+			return subs[i].ClientName < subs[j].ClientName
+		}
+		return subs[i].PathPrefix < subs[j].PathPrefix
+	})
+
+	return &filer_pb.ListMetadataSubscribersResponse{Subscribers: subs}, nil
+}

weed/server/filer_server.go

@@ -113,6 +113,9 @@ type FilerServer struct {
 	// track known metadata listeners
 	knownListenersLock sync.Mutex
 	knownListeners     map[int32]int32
+	// live metadata subscribers (FUSE mounts, S3, peer filers, ...) keyed by
+	// clientId, guarded by knownListenersLock. Exposed via ListMetadataSubscribers.
+	subscribers map[int32]*metadataSubscriber
 
 	// deduplicates concurrent remote object caching operations
 	remoteCacheGroup singleflight.Group
@@ -183,6 +186,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption)
 		option:                option,
 		grpcDialOption:        security.LoadClientTLS(util.GetViper(), "grpc.filer"),
 		knownListeners:        make(map[int32]int32),
+		subscribers:           make(map[int32]*metadataSubscriber),
 		inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)),
 		recentCopyRequests:    make(map[string]recentCopyRequest),
 		CredentialManager:     option.CredentialManager,