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