From 14bc6e5e4f97370abcbdeac24559559d7e82927c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 14 Sep 2026 23:32:32 -0700 Subject: [PATCH] servers: advertise the configured metricsPort to the master Filer, S3 and broker report it on the KeepConnected registration, volume servers in their heartbeat, and masters return their own in GetMasterConfiguration. MasterClient gains SetMetricsPort so the eight callers that have no metrics listener are untouched, and the volume server takes it as a constructor argument because its heartbeat goroutine starts there. In combined "weed server" one metrics listener serves the whole shared registry, so every component advertises the same port. --- weed/cluster/cluster.go | 15 +++++++++------ weed/cluster/cluster_test.go | 8 ++++---- weed/cluster/group_members.go | 17 ++++++++++------- weed/command/filer.go | 1 + weed/command/master.go | 1 + weed/command/s3.go | 1 + weed/command/server.go | 9 +++++++++ weed/command/volume.go | 2 +- weed/s3api/s3api_server.go | 14 ++++++++++---- weed/server/filer_server.go | 6 +++++- weed/server/master_grpc_server.go | 4 +++- .../master_grpc_server_admin_ping_test.go | 2 +- weed/server/master_grpc_server_cluster.go | 1 + weed/server/master_server.go | 12 ++++++++---- weed/server/volume_server.go | 5 ++++- weed/storage/store.go | 19 ++++++++++++------- weed/topology/data_node.go | 16 ++++++++++------ weed/wdclient/masterclient.go | 9 +++++++++ 18 files changed, 99 insertions(+), 43 deletions(-) diff --git a/weed/cluster/cluster.go b/weed/cluster/cluster.go index 5ee4e1c4c..6c6ccc76d 100644 --- a/weed/cluster/cluster.go +++ b/weed/cluster/cluster.go @@ -27,6 +27,9 @@ type ClusterNode struct { CreatedTs time.Time DataCenter DataCenter Rack Rack + // MetricsPort is the node's Prometheus /metrics port, or 0 when the node + // does not run a metrics listener. + MetricsPort uint32 } type ClusterNodeGroups struct { @@ -53,11 +56,11 @@ func (g *ClusterNodeGroups) getGroupMembers(filerGroup FilerGroupName, createIfN return members } -func (g *ClusterNodeGroups) AddClusterNode(filerGroup FilerGroupName, nodeType string, dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string) []*master_pb.KeepConnectedResponse { +func (g *ClusterNodeGroups) AddClusterNode(filerGroup FilerGroupName, nodeType string, dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string, metricsPort uint32) []*master_pb.KeepConnectedResponse { g.Lock() defer g.Unlock() m := g.getGroupMembers(filerGroup, true) - if t := m.addMember(dataCenter, rack, address, version); t != nil { + if t := m.addMember(dataCenter, rack, address, version, metricsPort); t != nil { return buildClusterNodeUpdateMessage(true, filerGroup, nodeType, address) } return nil @@ -95,15 +98,15 @@ func NewCluster() *Cluster { } } -func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string) []*master_pb.KeepConnectedResponse { +func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string, metricsPort uint32) []*master_pb.KeepConnectedResponse { filerGroup := FilerGroupName(ns) switch nodeType { case FilerType: - return cluster.filerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version) + return cluster.filerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version, metricsPort) case BrokerType: - return cluster.brokerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version) + return cluster.brokerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version, metricsPort) case S3Type: - return cluster.s3Groups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version) + return cluster.s3Groups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version, metricsPort) case MasterType: return buildClusterNodeUpdateMessage(true, filerGroup, nodeType, address) } diff --git a/weed/cluster/cluster_test.go b/weed/cluster/cluster_test.go index 5a11db8cc..9d54deb68 100644 --- a/weed/cluster/cluster_test.go +++ b/weed/cluster/cluster_test.go @@ -16,7 +16,7 @@ func TestConcurrentAddRemoveNodes(t *testing.T) { go func(i int) { defer wg.Done() address := strconv.Itoa(i) - c.AddClusterNode("", "filer", "", "", pb.ServerAddress(address), "23.45") + c.AddClusterNode("", "filer", "", "", pb.ServerAddress(address), "23.45", 0) }(i) } wg.Wait() @@ -43,8 +43,8 @@ func TestConcurrentAddRemoveNodes(t *testing.T) { func TestListClusterNodeUpdates(t *testing.T) { c := NewCluster() filer := pb.ServerAddress("10.0.0.20:8888") - c.AddClusterNode("group", FilerType, "dc1", "rack1", filer, "test") - c.AddClusterNode("group", BrokerType, "dc1", "rack1", pb.ServerAddress("10.0.0.20:17777"), "test") + c.AddClusterNode("group", FilerType, "dc1", "rack1", filer, "test", 0) + c.AddClusterNode("group", BrokerType, "dc1", "rack1", pb.ServerAddress("10.0.0.20:17777"), "test", 0) updates := c.ListClusterNodeUpdates("group", FilerType) if len(updates) != 1 { @@ -64,7 +64,7 @@ func TestListClusterNodeUpdates(t *testing.T) { func TestIsKnownNode(t *testing.T) { c := NewCluster() filer := pb.ServerAddress("10.0.0.20:8888") - c.AddClusterNode("", FilerType, "dc1", "rack1", filer, "test") + c.AddClusterNode("", FilerType, "dc1", "rack1", filer, "test", 0) if !c.IsKnownNode(FilerType, filer) { t.Fatalf("registered filer %s should be known", filer) diff --git a/weed/cluster/group_members.go b/weed/cluster/group_members.go index 5c0c09977..0775bf264 100644 --- a/weed/cluster/group_members.go +++ b/weed/cluster/group_members.go @@ -16,18 +16,21 @@ func newGroupMembers() *GroupMembers { } } -func (m *GroupMembers) addMember(dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string) *ClusterNode { +func (m *GroupMembers) addMember(dataCenter DataCenter, rack Rack, address pb.ServerAddress, version string, metricsPort uint32) *ClusterNode { if existingNode, found := m.members[address]; found { existingNode.counter++ + // A restarted node may have gained or lost its metrics listener. + existingNode.MetricsPort = metricsPort return nil } t := &ClusterNode{ - Address: address, - Version: version, - counter: 1, - CreatedTs: time.Now(), - DataCenter: dataCenter, - Rack: rack, + Address: address, + Version: version, + counter: 1, + CreatedTs: time.Now(), + DataCenter: dataCenter, + Rack: rack, + MetricsPort: metricsPort, } m.members[address] = t return t diff --git a/weed/command/filer.go b/weed/command/filer.go index 0fa629996..80860e4ae 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -380,6 +380,7 @@ func (fo *FilerOptions) startFiler() { fs, nfs_err := weed_server.NewFilerServer(defaultMux, publicVolumeMux, &weed_server.FilerOption{ Masters: fo.masters, + MetricsPort: uint32(*fo.metricsHttpPort), FilerGroup: *fo.filerGroup, Collection: *fo.collection, DefaultReplication: *fo.defaultReplicaPlacement, diff --git a/weed/command/master.go b/weed/command/master.go index 61bbea730..178827550 100644 --- a/weed/command/master.go +++ b/weed/command/master.go @@ -479,6 +479,7 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp WhiteList: whiteList, DisableHttp: *m.disableHttp, MetricsAddress: *m.metricsAddress, + MetricsPort: *m.metricsHttpPort, MetricsIntervalSec: *m.metricsIntervalSec, TelemetryUrl: *m.telemetryUrl, TelemetryEnabled: *m.telemetryEnabled, diff --git a/weed/command/s3.go b/weed/command/s3.go index 6ca45f057..9512c641b 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -358,6 +358,7 @@ func (s3opt *S3Options) startS3Server() bool { Filers: filerAddresses, Masters: masterAddresses, Port: *s3opt.port, + MetricsPort: uint32(*s3opt.metricsHttpPort), Config: *s3opt.config, DomainName: *s3opt.domainName, AllowedOrigins: strings.Split(*s3opt.allowedOrigins, ","), diff --git a/weed/command/server.go b/weed/command/server.go index 6a53c4467..2b92dc93f 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -343,6 +343,15 @@ func runServer(cmd *Command, args []string) bool { webdavOptions.filer = &filerAddress mqBrokerOptions.filerGroup = filerOptions.filerGroup + // One process, one shared Prometheus registry, so a single metrics listener + // serves master, volume, filer and S3 series. Every component advertises + // that same port; the admin server deduplicates targets by address and + // attributes series by metric name. + masterOptions.metricsHttpPort = serverMetricsHttpPort + serverOptions.v.metricsHttpPort = serverMetricsHttpPort + filerOptions.metricsHttpPort = serverMetricsHttpPort + s3Options.metricsHttpPort = serverMetricsHttpPort + go stats_collect.StartMetricsServer(*serverMetricsHttpIp, *serverMetricsHttpPort) *volumeDataFolders = util.ResolveCommaSeparatedPaths(*volumeDataFolders) diff --git a/weed/command/volume.go b/weed/command/volume.go index 012222798..aab1a1d1e 100644 --- a/weed/command/volume.go +++ b/weed/command/volume.go @@ -438,7 +438,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v RecoveryCoef: *v.diskRecoveryCoef, } volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux, - *v.ip, *v.port, *v.portGrpc, *v.publicUrl, volumeServerId, + *v.ip, *v.port, *v.portGrpc, *v.metricsHttpPort, *v.publicUrl, volumeServerId, v.folders, v.folderMaxLimits, minFreeSpaces, diskTypes, folderTags, util.ResolvePath(*v.idxFolder), volumeNeedleMapKind, diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index baefc8218..8e48e38bc 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -65,10 +65,15 @@ type S3ApiServerOption struct { Ip string // address advertised to the cluster; empty falls back to BindIp BindIp string GrpcPort int - ExternalUrl string // external URL clients use, tried first during signature verification behind a reverse proxy - DefaultFileMode uint32 // default file permission mode for S3 uploads (e.g. 0660, 0644) - CacheSizeMB int64 // in-memory chunk cache capacity in MB for the shared ReaderCache; 0 disables - MaxMB int32 // filer's -maxMB, read from the filer configuration at startup + // MetricsPort is the Prometheus /metrics port this S3 server serves, + // advertised to the master so the admin server can scrape it. 0 when + // disabled. S3 metrics carry bucket labels, so they are deliberately not + // served on the client-facing S3 port. + MetricsPort uint32 + ExternalUrl string // external URL clients use, tried first during signature verification behind a reverse proxy + DefaultFileMode uint32 // default file permission mode for S3 uploads (e.g. 0660, 0644) + CacheSizeMB int64 // in-memory chunk cache capacity in MB for the shared ReaderCache; 0 disables + MaxMB int32 // filer's -maxMB, read from the filer configuration at startup // AllowUntrustedRemoteEndpoints lets a read of a remote-only object dial a // mounted endpoint that resolves to a loopback / private / metadata host. AllowUntrustedRemoteEndpoints bool @@ -209,6 +214,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl } clientHost := option.advertisedHost() masterClient = wdclient.NewMasterClient(option.GrpcDialOption, option.FilerGroup, cluster.S3Type, pb.ServerAddress(util.JoinHostPort(clientHost, option.GrpcPort)), option.DataCenter, "", *pb.NewServiceDiscoveryFromMap(masterMap)) + masterClient.SetMetricsPort(option.MetricsPort) // Build the object-write lock client and subscribe to the master's // lock-ring updates BEFORE starting the master loop, so the initial // LockRingUpdate sent on connect isn't dropped (the master only delivers diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 7d998ad8f..f7876c1a7 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -88,7 +88,10 @@ type FilerOption struct { TusMaxSize int64 TusSessionExpiry time.Duration S3ConfigFile string // optional path to static S3 identity config file - CredentialManager *credential.CredentialManager + // MetricsPort is the Prometheus /metrics port this filer serves, advertised + // to the master so the admin server can scrape it. 0 when disabled. + MetricsPort uint32 + CredentialManager *credential.CredentialManager // AllowUntrustedRemoteEndpoints lets a read of a remote-only entry dial a // mounted endpoint that resolves to a loopback / private / metadata host. AllowUntrustedRemoteEndpoints bool @@ -244,6 +247,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) fs.checkWithMaster() go stats.LoopPushingMetric("filer", string(fs.option.Host), fs.metricsAddress, fs.metricsIntervalSec) + fs.filer.MasterClient.SetMetricsPort(option.MetricsPort) go fs.filer.MasterClient.KeepConnectedToMaster(context.Background()) fs.option.recursiveDelete = v.GetBool("filer.options.recursive_delete") diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go index 976435545..b4a8e6589 100644 --- a/weed/server/master_grpc_server.go +++ b/weed/server/master_grpc_server.go @@ -188,6 +188,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ dc := ms.Topo.GetOrCreateDataCenter(dcName) rack := dc.GetOrCreateRack(rackName) dn = rack.GetOrCreateDataNode(heartbeat.Ip, int(heartbeat.Port), int(heartbeat.GrpcPort), heartbeat.PublicUrl, heartbeat.Id, heartbeat.MaxVolumeCounts) + dn.MetricsPort = int(heartbeat.MetricsPort) glog.V(0).Infof("added volume server %d: %v (id=%s, ip=%v:%d) %v", dn.Counter, dn.Id(), heartbeat.Id, heartbeat.GetIp(), heartbeat.GetPort(), heartbeat.LocationUuids) uuidlist, err := ms.RegisterUuids(heartbeat) if err != nil { @@ -418,7 +419,7 @@ func (ms *MasterServer) KeepConnected(stream master_pb.Seaweed_KeepConnectedServ stopChan := make(chan bool, 1) clientName, messageChan := ms.addClient(req.FilerGroup, req.ClientType, peerAddress) - for _, update := range ms.Cluster.AddClusterNode(req.FilerGroup, req.ClientType, cluster.DataCenter(req.DataCenter), cluster.Rack(req.Rack), peerAddress, req.Version) { + for _, update := range ms.Cluster.AddClusterNode(req.FilerGroup, req.ClientType, cluster.DataCenter(req.DataCenter), cluster.Rack(req.Rack), peerAddress, req.Version, req.MetricsPort) { glog.V(1).Infof("Cluster: %s node %s added to group '%s'", req.ClientType, peerAddress, req.FilerGroup) ms.broadcastToClients(update) } @@ -662,6 +663,7 @@ func (ms *MasterServer) GetMasterConfiguration(ctx context.Context, req *master_ resp := &master_pb.GetMasterConfigurationResponse{ MetricsAddress: ms.option.MetricsAddress, MetricsIntervalSeconds: uint32(ms.option.MetricsIntervalSec), + MetricsPort: uint32(ms.option.MetricsPort), StorageBackends: backend.ToPbStorageBackends(), DefaultReplication: ms.option.DefaultReplicaPlacement, VolumeSizeLimitMB: uint32(ms.option.VolumeSizeLimitMB), diff --git a/weed/server/master_grpc_server_admin_ping_test.go b/weed/server/master_grpc_server_admin_ping_test.go index dd31f0b0d..728578009 100644 --- a/weed/server/master_grpc_server_admin_ping_test.go +++ b/weed/server/master_grpc_server_admin_ping_test.go @@ -27,7 +27,7 @@ func TestMasterIsKnownPingTarget(t *testing.T) { c := cluster.NewCluster() filerAddr := pb.ServerAddress("10.0.0.20:8888") - c.AddClusterNode("", cluster.FilerType, "dc1", "rack1", filerAddr, "test") + c.AddClusterNode("", cluster.FilerType, "dc1", "rack1", filerAddr, "test", 0) ms := &MasterServer{ option: &MasterOption{Master: pb.ServerAddress("10.0.0.1:9333")}, diff --git a/weed/server/master_grpc_server_cluster.go b/weed/server/master_grpc_server_cluster.go index 036dd8971..75060a54f 100644 --- a/weed/server/master_grpc_server_cluster.go +++ b/weed/server/master_grpc_server_cluster.go @@ -22,6 +22,7 @@ func (ms *MasterServer) ListClusterNodes(ctx context.Context, req *master_pb.Lis CreatedAtNs: node.CreatedTs.UnixNano(), DataCenter: string(node.DataCenter), Rack: string(node.Rack), + MetricsPort: node.MetricsPort, }) } return resp, nil diff --git a/weed/server/master_server.go b/weed/server/master_server.go index 3a1364c11..2d920112e 100644 --- a/weed/server/master_server.go +++ b/weed/server/master_server.go @@ -58,10 +58,14 @@ type MasterOption struct { DisableHttp bool MetricsAddress string MetricsIntervalSec int - IsFollower bool - TelemetryUrl string - TelemetryEnabled bool - VolumeGrowthDisabled bool + // MetricsPort is this master's Prometheus /metrics port (-metricsPort), + // reported in GetMasterConfiguration so the admin server can scrape it. + // Unrelated to MetricsAddress, which is the Prometheus push gateway. + MetricsPort int + IsFollower bool + TelemetryUrl string + TelemetryEnabled bool + VolumeGrowthDisabled bool } type MasterServer struct { diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index 1b6feeac4..19def9a13 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -63,7 +63,7 @@ type VolumeServer struct { } func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, - port int, grpcPort int, publicUrl string, id string, + port int, grpcPort int, metricsPort int, publicUrl string, id string, folders []string, maxCounts []int32, minFreeSpaces []util.MinFreeSpace, diskTypes []types.DiskType, diskTags [][]string, idxFolder string, needleMapKind storage.NeedleMapKind, @@ -136,6 +136,9 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string, vs.checkWithMaster() vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig) + // Set before the heartbeat goroutine starts below, since the heartbeat is + // what advertises this port to the master. + vs.store.MetricsPort = metricsPort vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec) handleStaticResources(adminMux) diff --git a/weed/storage/store.go b/weed/storage/store.go index a8983bb36..1076b97f8 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -60,13 +60,17 @@ type ReadOption struct { * A VolumeServer contains one Store */ type Store struct { - MasterAddress pb.ServerAddress - grpcDialOption grpc.DialOption - volumeSizeLimit uint64 // read from the master - preallocate atomic.Bool // read from the master - Ip string - Port int - GrpcPort int + MasterAddress pb.ServerAddress + grpcDialOption grpc.DialOption + volumeSizeLimit uint64 // read from the master + preallocate atomic.Bool // read from the master + Ip string + Port int + GrpcPort int + // MetricsPort is the Prometheus /metrics port this volume server serves, + // advertised to the master so the admin server can scrape it. 0 when + // disabled. + MetricsPort int PublicUrl string Id string // volume server id, independent of ip:port for stable identification Locations []*DiskLocation @@ -684,6 +688,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { Ip: s.Ip, Port: uint32(s.Port), GrpcPort: uint32(s.GrpcPort), + MetricsPort: uint32(s.MetricsPort), PublicUrl: s.PublicUrl, Id: s.Id, MaxVolumeCounts: maxVolumeCounts, diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 00325efb7..875ba16b5 100644 --- a/weed/topology/data_node.go +++ b/weed/topology/data_node.go @@ -15,9 +15,12 @@ import ( type DataNode struct { NodeImpl - Ip string - Port int - GrpcPort int + Ip string + Port int + GrpcPort int + // MetricsPort is the volume server's Prometheus /metrics port as reported in + // its heartbeat, or 0 when it does not run a metrics listener. + MetricsPort int PublicUrl string LastSeen int64 // unix time in seconds Counter int // in race condition, the previous dataNode was not dead @@ -404,9 +407,10 @@ func (dn *DataNode) ToDataNodeInfo(filter VolumeFilter) *master_pb.DataNodeInfo Id: string(dn.Id()), // Start from disk usage counters so empty disks are still represented // even when there are no volumes/EC shards on this data node yet. - DiskInfos: dn.diskUsages.ToDiskInfo(), - GrpcPort: uint32(dn.GrpcPort), - Address: dn.Url(), // ip:port for connecting to the volume server + DiskInfos: dn.diskUsages.ToDiskInfo(), + GrpcPort: uint32(dn.GrpcPort), + MetricsPort: uint32(dn.MetricsPort), + Address: dn.Url(), // ip:port for connecting to the volume server } if m.DiskInfos == nil { m.DiskInfos = make(map[string]*master_pb.DiskInfo) diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go index beced07cf..3e6a0a019 100644 --- a/weed/wdclient/masterclient.go +++ b/weed/wdclient/masterclient.go @@ -149,6 +149,7 @@ type MasterClient struct { clientType string clientHost pb.ServerAddress rack string + metricsPort uint32 currentMaster pb.ServerAddress currentMasterLock sync.RWMutex masters pb.ServerDiscovery @@ -180,6 +181,13 @@ func NewMasterClient(grpcDialOption grpc.DialOption, filerGroup string, clientTy return mc } +// SetMetricsPort advertises this node's Prometheus /metrics port to the master, +// so a central scraper can discover it. Must be called before +// KeepConnectedToMaster starts, which is what publishes the value. +func (mc *MasterClient) SetMetricsPort(port uint32) { + mc.metricsPort = port +} + func (mc *MasterClient) SetOnPeerUpdateFn(onPeerUpdate func(update *master_pb.ClusterNodeUpdate, startFrom time.Time)) { mc.OnPeerUpdateLock.Lock() mc.OnPeerUpdate = onPeerUpdate @@ -246,6 +254,7 @@ func (mc *MasterClient) tryConnectToMaster(ctx context.Context, master pb.Server ClientType: mc.clientType, ClientAddress: string(mc.clientHost), Version: version.Version(), + MetricsPort: mc.metricsPort, }); err != nil { glog.V(0).Infof("%s.%s masterClient failed to send to %s: %v", mc.FilerGroup, mc.clientType, master, err) stats.MasterClientConnectCounter.WithLabelValues(stats.FailedToSend).Inc()