mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-21 22:12:36 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Chris Lu
|
2f26d5779b | ||
|
Chris Lu
|
14bc6e5e4f | ||
|
Chris Lu
|
dfde24f3ee |
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -465,6 +465,12 @@ func peerIndex(self pb.ServerAddress, peers []pb.ServerAddress) int {
|
||||
}
|
||||
|
||||
func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOption {
|
||||
// Not every caller wires up every flag, so treat an unset metrics port as
|
||||
// disabled rather than dereferencing nil.
|
||||
metricsPort := 0
|
||||
if m.metricsHttpPort != nil {
|
||||
metricsPort = *m.metricsHttpPort
|
||||
}
|
||||
masterAddress := pb.NewServerAddress(*m.ip, *m.port, *m.portGrpc)
|
||||
return &weed_server.MasterOption{
|
||||
Master: masterAddress,
|
||||
@@ -479,6 +485,7 @@ func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOp
|
||||
WhiteList: whiteList,
|
||||
DisableHttp: *m.disableHttp,
|
||||
MetricsAddress: *m.metricsAddress,
|
||||
MetricsPort: metricsPort,
|
||||
MetricsIntervalSec: *m.metricsIntervalSec,
|
||||
TelemetryUrl: *m.telemetryUrl,
|
||||
TelemetryEnabled: *m.telemetryEnabled,
|
||||
|
||||
@@ -1308,6 +1308,13 @@ func runMini(cmd *Command, args []string) bool {
|
||||
}
|
||||
pb.RegisterLocalGrpcSocket(*miniIp, *miniAdminOptions.grpcPort, fmt.Sprintf("/tmp/seaweedfs-admin-grpc-%d.sock", *miniAdminOptions.grpcPort))
|
||||
|
||||
// One process, one shared Prometheus registry, so this single listener
|
||||
// serves every component's series. Point each component at it so they all
|
||||
// advertise the same port to the master.
|
||||
miniMasterOptions.metricsHttpPort = miniMetricsHttpPort
|
||||
miniOptions.v.metricsHttpPort = miniMetricsHttpPort
|
||||
miniFilerOptions.metricsHttpPort = miniMetricsHttpPort
|
||||
|
||||
go stats_collect.StartMetricsServer(*miniMetricsHttpIp, *miniMetricsHttpPort)
|
||||
|
||||
if *miniMasterOptions.volumeSizeLimitMB > util.MaxVolumeSizeLimitMB {
|
||||
|
||||
@@ -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, ","),
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -98,6 +98,9 @@ message Heartbeat {
|
||||
|
||||
map<string, uint32> max_volume_counts = 4;
|
||||
uint32 grpc_port = 20;
|
||||
// Port of this volume server's Prometheus /metrics listener (-metricsPort),
|
||||
// or 0 when it is not enabled.
|
||||
uint32 metrics_port = 29;
|
||||
repeated string location_uuids = 21;
|
||||
string id = 22; // volume server id, independent of ip:port for stable identification
|
||||
|
||||
@@ -213,6 +216,10 @@ message KeepConnectedRequest {
|
||||
string filer_group = 5;
|
||||
string data_center = 6;
|
||||
string rack = 7;
|
||||
// Port of this node's Prometheus /metrics listener (-metricsPort), or 0 when
|
||||
// it is not enabled. Advertised so the admin server can scrape it without
|
||||
// exposing metrics on the client-facing service port.
|
||||
uint32 metrics_port = 8;
|
||||
}
|
||||
|
||||
message VolumeLocation {
|
||||
@@ -400,6 +407,7 @@ message DataNodeInfo {
|
||||
map<string, DiskInfo> diskInfos = 2;
|
||||
uint32 grpc_port = 3;
|
||||
string address = 4; // ip:port for connecting to the volume server
|
||||
uint32 metrics_port = 5; // Prometheus /metrics port, or 0 when not enabled
|
||||
}
|
||||
message RackInfo {
|
||||
string id = 1;
|
||||
@@ -514,6 +522,10 @@ message GetMasterConfigurationResponse {
|
||||
// MIGRATION: fields 8-9 help migrate master.toml [master.maintenance] to admin script plugin. Remove after March 2027.
|
||||
string maintenance_scripts = 8;
|
||||
uint32 maintenance_sleep_minutes = 9;
|
||||
// Port of this master's Prometheus /metrics listener (-metricsPort), or 0
|
||||
// when it is not enabled. metrics_address above is unrelated: it is the
|
||||
// Prometheus push gateway that servers push to.
|
||||
uint32 metrics_port = 10;
|
||||
}
|
||||
|
||||
message ListClusterNodesRequest {
|
||||
@@ -528,6 +540,8 @@ message ListClusterNodesResponse {
|
||||
int64 created_at_ns = 4;
|
||||
string data_center = 5;
|
||||
string rack = 6;
|
||||
// Port of this node's Prometheus /metrics listener, or 0 when not enabled.
|
||||
uint32 metrics_port = 7;
|
||||
}
|
||||
repeated ClusterNode cluster_nodes = 1;
|
||||
}
|
||||
|
||||
@@ -113,8 +113,11 @@ type Heartbeat struct {
|
||||
HasNoEcShards bool `protobuf:"varint,19,opt,name=has_no_ec_shards,json=hasNoEcShards,proto3" json:"has_no_ec_shards,omitempty"`
|
||||
MaxVolumeCounts map[string]uint32 `protobuf:"bytes,4,rep,name=max_volume_counts,json=maxVolumeCounts,proto3" json:"max_volume_counts,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
|
||||
GrpcPort uint32 `protobuf:"varint,20,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
|
||||
LocationUuids []string `protobuf:"bytes,21,rep,name=location_uuids,json=locationUuids,proto3" json:"location_uuids,omitempty"`
|
||||
Id string `protobuf:"bytes,22,opt,name=id,proto3" json:"id,omitempty"` // volume server id, independent of ip:port for stable identification
|
||||
// Port of this volume server's Prometheus /metrics listener (-metricsPort),
|
||||
// or 0 when it is not enabled.
|
||||
MetricsPort uint32 `protobuf:"varint,29,opt,name=metrics_port,json=metricsPort,proto3" json:"metrics_port,omitempty"`
|
||||
LocationUuids []string `protobuf:"bytes,21,rep,name=location_uuids,json=locationUuids,proto3" json:"location_uuids,omitempty"`
|
||||
Id string `protobuf:"bytes,22,opt,name=id,proto3" json:"id,omitempty"` // volume server id, independent of ip:port for stable identification
|
||||
// state flags
|
||||
State *volume_server_pb.VolumeServerState `protobuf:"bytes,23,opt,name=state,proto3" json:"state,omitempty"`
|
||||
DiskTags []*DiskTag `protobuf:"bytes,24,rep,name=disk_tags,json=diskTags,proto3" json:"disk_tags,omitempty"`
|
||||
@@ -283,6 +286,13 @@ func (x *Heartbeat) GetGrpcPort() uint32 {
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *Heartbeat) GetMetricsPort() uint32 {
|
||||
if x != nil {
|
||||
return x.MetricsPort
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *Heartbeat) GetLocationUuids() []string {
|
||||
if x != nil {
|
||||
return x.LocationUuids
|
||||
@@ -998,6 +1008,10 @@ type KeepConnectedRequest struct {
|
||||
FilerGroup string `protobuf:"bytes,5,opt,name=filer_group,json=filerGroup,proto3" json:"filer_group,omitempty"`
|
||||
DataCenter string `protobuf:"bytes,6,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
|
||||
Rack string `protobuf:"bytes,7,opt,name=rack,proto3" json:"rack,omitempty"`
|
||||
// Port of this node's Prometheus /metrics listener (-metricsPort), or 0 when
|
||||
// it is not enabled. Advertised so the admin server can scrape it without
|
||||
// exposing metrics on the client-facing service port.
|
||||
MetricsPort uint32 `protobuf:"varint,8,opt,name=metrics_port,json=metricsPort,proto3" json:"metrics_port,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -1074,6 +1088,13 @@ func (x *KeepConnectedRequest) GetRack() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *KeepConnectedRequest) GetMetricsPort() uint32 {
|
||||
if x != nil {
|
||||
return x.MetricsPort
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type VolumeLocation struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Url string `protobuf:"bytes,1,opt,name=url,proto3" json:"url,omitempty"`
|
||||
@@ -2608,7 +2629,8 @@ type DataNodeInfo struct {
|
||||
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
|
||||
DiskInfos map[string]*DiskInfo `protobuf:"bytes,2,rep,name=diskInfos,proto3" json:"diskInfos,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
|
||||
GrpcPort uint32 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
|
||||
Address string `protobuf:"bytes,4,opt,name=address,proto3" json:"address,omitempty"` // ip:port for connecting to the volume server
|
||||
Address string `protobuf:"bytes,4,opt,name=address,proto3" json:"address,omitempty"` // ip:port for connecting to the volume server
|
||||
MetricsPort uint32 `protobuf:"varint,5,opt,name=metrics_port,json=metricsPort,proto3" json:"metrics_port,omitempty"` // Prometheus /metrics port, or 0 when not enabled
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -2671,6 +2693,13 @@ func (x *DataNodeInfo) GetAddress() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *DataNodeInfo) GetMetricsPort() uint32 {
|
||||
if x != nil {
|
||||
return x.MetricsPort
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type RackInfo struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
|
||||
@@ -3639,8 +3668,12 @@ type GetMasterConfigurationResponse struct {
|
||||
// MIGRATION: fields 8-9 help migrate master.toml [master.maintenance] to admin script plugin. Remove after March 2027.
|
||||
MaintenanceScripts string `protobuf:"bytes,8,opt,name=maintenance_scripts,json=maintenanceScripts,proto3" json:"maintenance_scripts,omitempty"`
|
||||
MaintenanceSleepMinutes uint32 `protobuf:"varint,9,opt,name=maintenance_sleep_minutes,json=maintenanceSleepMinutes,proto3" json:"maintenance_sleep_minutes,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
// Port of this master's Prometheus /metrics listener (-metricsPort), or 0
|
||||
// when it is not enabled. metrics_address above is unrelated: it is the
|
||||
// Prometheus push gateway that servers push to.
|
||||
MetricsPort uint32 `protobuf:"varint,10,opt,name=metrics_port,json=metricsPort,proto3" json:"metrics_port,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
|
||||
func (x *GetMasterConfigurationResponse) Reset() {
|
||||
@@ -3736,6 +3769,13 @@ func (x *GetMasterConfigurationResponse) GetMaintenanceSleepMinutes() uint32 {
|
||||
return 0
|
||||
}
|
||||
|
||||
func (x *GetMasterConfigurationResponse) GetMetricsPort() uint32 {
|
||||
if x != nil {
|
||||
return x.MetricsPort
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type ListClusterNodesRequest struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
ClientType string `protobuf:"bytes,1,opt,name=client_type,json=clientType,proto3" json:"client_type,omitempty"`
|
||||
@@ -4865,12 +4905,14 @@ func (x *LookupEcVolumeResponse_EcShardIdLocation) GetLocations() []*Location {
|
||||
}
|
||||
|
||||
type ListClusterNodesResponse_ClusterNode struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Address string `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
|
||||
Version string `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"`
|
||||
CreatedAtNs int64 `protobuf:"varint,4,opt,name=created_at_ns,json=createdAtNs,proto3" json:"created_at_ns,omitempty"`
|
||||
DataCenter string `protobuf:"bytes,5,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
|
||||
Rack string `protobuf:"bytes,6,opt,name=rack,proto3" json:"rack,omitempty"`
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Address string `protobuf:"bytes,1,opt,name=address,proto3" json:"address,omitempty"`
|
||||
Version string `protobuf:"bytes,2,opt,name=version,proto3" json:"version,omitempty"`
|
||||
CreatedAtNs int64 `protobuf:"varint,4,opt,name=created_at_ns,json=createdAtNs,proto3" json:"created_at_ns,omitempty"`
|
||||
DataCenter string `protobuf:"bytes,5,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
|
||||
Rack string `protobuf:"bytes,6,opt,name=rack,proto3" json:"rack,omitempty"`
|
||||
// Port of this node's Prometheus /metrics listener, or 0 when not enabled.
|
||||
MetricsPort uint32 `protobuf:"varint,7,opt,name=metrics_port,json=metricsPort,proto3" json:"metrics_port,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -4940,6 +4982,13 @@ func (x *ListClusterNodesResponse_ClusterNode) GetRack() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (x *ListClusterNodesResponse_ClusterNode) GetMetricsPort() uint32 {
|
||||
if x != nil {
|
||||
return x.MetricsPort
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
type RaftListClusterServersResponse_ClusterServers struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
|
||||
@@ -5017,7 +5066,7 @@ const file_master_proto_rawDesc = "" +
|
||||
"\adisk_id\x18\x01 \x01(\rR\x06diskId\x12\x12\n" +
|
||||
"\x04tags\x18\x02 \x03(\tR\x04tags\x12\x12\n" +
|
||||
"\x04type\x18\x03 \x01(\tR\x04type\x12(\n" +
|
||||
"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xf0\v\n" +
|
||||
"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\x93\f\n" +
|
||||
"\tHeartbeat\x12\x0e\n" +
|
||||
"\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" +
|
||||
"\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" +
|
||||
@@ -5041,7 +5090,8 @@ const file_master_proto_rawDesc = "" +
|
||||
"\x11deleted_ec_shards\x18\x12 \x03(\v2*.master_pb.VolumeEcShardInformationMessageR\x0fdeletedEcShards\x12'\n" +
|
||||
"\x10has_no_ec_shards\x18\x13 \x01(\bR\rhasNoEcShards\x12U\n" +
|
||||
"\x11max_volume_counts\x18\x04 \x03(\v2).master_pb.Heartbeat.MaxVolumeCountsEntryR\x0fmaxVolumeCounts\x12\x1b\n" +
|
||||
"\tgrpc_port\x18\x14 \x01(\rR\bgrpcPort\x12%\n" +
|
||||
"\tgrpc_port\x18\x14 \x01(\rR\bgrpcPort\x12!\n" +
|
||||
"\fmetrics_port\x18\x1d \x01(\rR\vmetricsPort\x12%\n" +
|
||||
"\x0elocation_uuids\x18\x15 \x03(\tR\rlocationUuids\x12\x0e\n" +
|
||||
"\x02id\x18\x16 \x01(\tR\x02id\x129\n" +
|
||||
"\x05state\x18\x17 \x01(\v2#.volume_server_pb.VolumeServerStateR\x05state\x12/\n" +
|
||||
@@ -5137,7 +5187,7 @@ const file_master_proto_rawDesc = "" +
|
||||
"\x04data\x18\x01 \x01(\rR\x04data\x12\x16\n" +
|
||||
"\x06parity\x18\x02 \x01(\rR\x06parity\x12\x1d\n" +
|
||||
"\n" +
|
||||
"volume_ids\x18\x03 \x03(\rR\tvolumeIds\"\xce\x01\n" +
|
||||
"volume_ids\x18\x03 \x03(\rR\tvolumeIds\"\xf1\x01\n" +
|
||||
"\x14KeepConnectedRequest\x12\x1f\n" +
|
||||
"\vclient_type\x18\x01 \x01(\tR\n" +
|
||||
"clientType\x12%\n" +
|
||||
@@ -5147,7 +5197,8 @@ const file_master_proto_rawDesc = "" +
|
||||
"filerGroup\x12\x1f\n" +
|
||||
"\vdata_center\x18\x06 \x01(\tR\n" +
|
||||
"dataCenter\x12\x12\n" +
|
||||
"\x04rack\x18\a \x01(\tR\x04rack\"\x9e\x03\n" +
|
||||
"\x04rack\x18\a \x01(\tR\x04rack\x12!\n" +
|
||||
"\fmetrics_port\x18\b \x01(\rR\vmetricsPort\"\x9e\x03\n" +
|
||||
"\x0eVolumeLocation\x12\x10\n" +
|
||||
"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
|
||||
"\n" +
|
||||
@@ -5296,12 +5347,13 @@ const file_master_proto_rawDesc = "" +
|
||||
"\x0fdisk_free_bytes\x18\r \x01(\x04R\rdiskFreeBytes\x1aG\n" +
|
||||
"\x19MaxVolumeCountByDiskEntry\x12\x10\n" +
|
||||
"\x03key\x18\x01 \x01(\rR\x03key\x12\x14\n" +
|
||||
"\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01\"\xee\x01\n" +
|
||||
"\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01\"\x91\x02\n" +
|
||||
"\fDataNodeInfo\x12\x0e\n" +
|
||||
"\x02id\x18\x01 \x01(\tR\x02id\x12D\n" +
|
||||
"\tdiskInfos\x18\x02 \x03(\v2&.master_pb.DataNodeInfo.DiskInfosEntryR\tdiskInfos\x12\x1b\n" +
|
||||
"\tgrpc_port\x18\x03 \x01(\rR\bgrpcPort\x12\x18\n" +
|
||||
"\aaddress\x18\x04 \x01(\tR\aaddress\x1aQ\n" +
|
||||
"\aaddress\x18\x04 \x01(\tR\aaddress\x12!\n" +
|
||||
"\fmetrics_port\x18\x05 \x01(\rR\vmetricsPort\x1aQ\n" +
|
||||
"\x0eDiskInfosEntry\x12\x10\n" +
|
||||
"\x03key\x18\x01 \x01(\tR\x03key\x12)\n" +
|
||||
"\x05value\x18\x02 \x01(\v2\x13.master_pb.DiskInfoR\x05value:\x028\x01\"\xf0\x01\n" +
|
||||
@@ -5385,7 +5437,7 @@ const file_master_proto_rawDesc = "" +
|
||||
" \x01(\bR\n" +
|
||||
"isReadonly\"\x1c\n" +
|
||||
"\x1aVolumeMarkReadonlyResponse\"\x1f\n" +
|
||||
"\x1dGetMasterConfigurationRequest\"\xe0\x03\n" +
|
||||
"\x1dGetMasterConfigurationRequest\"\x83\x04\n" +
|
||||
"\x1eGetMasterConfigurationResponse\x12'\n" +
|
||||
"\x0fmetrics_address\x18\x01 \x01(\tR\x0emetricsAddress\x128\n" +
|
||||
"\x18metrics_interval_seconds\x18\x02 \x01(\rR\x16metricsIntervalSeconds\x12D\n" +
|
||||
@@ -5395,22 +5447,25 @@ const file_master_proto_rawDesc = "" +
|
||||
"\x15volume_size_limit_m_b\x18\x06 \x01(\rR\x11volumeSizeLimitMB\x12-\n" +
|
||||
"\x12volume_preallocate\x18\a \x01(\bR\x11volumePreallocate\x12/\n" +
|
||||
"\x13maintenance_scripts\x18\b \x01(\tR\x12maintenanceScripts\x12:\n" +
|
||||
"\x19maintenance_sleep_minutes\x18\t \x01(\rR\x17maintenanceSleepMinutes\"q\n" +
|
||||
"\x19maintenance_sleep_minutes\x18\t \x01(\rR\x17maintenanceSleepMinutes\x12!\n" +
|
||||
"\fmetrics_port\x18\n" +
|
||||
" \x01(\rR\vmetricsPort\"q\n" +
|
||||
"\x17ListClusterNodesRequest\x12\x1f\n" +
|
||||
"\vclient_type\x18\x01 \x01(\tR\n" +
|
||||
"clientType\x12\x1f\n" +
|
||||
"\vfiler_group\x18\x02 \x01(\tR\n" +
|
||||
"filerGroup\x12\x14\n" +
|
||||
"\x05limit\x18\x04 \x01(\x05R\x05limit\"\x8d\x02\n" +
|
||||
"\x05limit\x18\x04 \x01(\x05R\x05limit\"\xb0\x02\n" +
|
||||
"\x18ListClusterNodesResponse\x12T\n" +
|
||||
"\rcluster_nodes\x18\x01 \x03(\v2/.master_pb.ListClusterNodesResponse.ClusterNodeR\fclusterNodes\x1a\x9a\x01\n" +
|
||||
"\rcluster_nodes\x18\x01 \x03(\v2/.master_pb.ListClusterNodesResponse.ClusterNodeR\fclusterNodes\x1a\xbd\x01\n" +
|
||||
"\vClusterNode\x12\x18\n" +
|
||||
"\aaddress\x18\x01 \x01(\tR\aaddress\x12\x18\n" +
|
||||
"\aversion\x18\x02 \x01(\tR\aversion\x12\"\n" +
|
||||
"\rcreated_at_ns\x18\x04 \x01(\x03R\vcreatedAtNs\x12\x1f\n" +
|
||||
"\vdata_center\x18\x05 \x01(\tR\n" +
|
||||
"dataCenter\x12\x12\n" +
|
||||
"\x04rack\x18\x06 \x01(\tR\x04rack\"\xc5\x01\n" +
|
||||
"\x04rack\x18\x06 \x01(\tR\x04rack\x12!\n" +
|
||||
"\fmetrics_port\x18\a \x01(\rR\vmetricsPort\"\xc5\x01\n" +
|
||||
"\x16LeaseAdminTokenRequest\x12%\n" +
|
||||
"\x0eprevious_token\x18\x01 \x01(\x03R\rpreviousToken\x12,\n" +
|
||||
"\x12previous_lock_time\x18\x02 \x01(\x03R\x10previousLockTime\x12\x1b\n" +
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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")},
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
+12
-7
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user