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

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

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)

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

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,

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,

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, ","),

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)

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,

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

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")

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),

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")},

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

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 {

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)

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,

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)

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()