From a5763986628fa629cb87b58f1b885426cf664d88 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 22:36:02 -0800 Subject: [PATCH] weed-vector: advertise its metrics port when registering (#12604) * master.proto: KeepConnectedRequest carries the node's metrics port Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * weed-vector: advertise --metricsPort when registering with the master Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK * master.proto: number metrics_port 9 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011UpvafvK8JpagF94MHWqwK --------- Co-authored-by: Claude Opus 5.5 --- seaweed-vector/crates/server/src/main.rs | 1 + .../crates/server/src/master_client.rs | 37 ++++++++++++------- weed/pb/master.proto | 4 ++ weed/pb/master_pb/master.pb.go | 18 +++++++-- weed/pb/master_pb/master_grpc.pb.go | 2 +- 5 files changed, 44 insertions(+), 18 deletions(-) diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index fe0e482902..8211ec3bed 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -209,6 +209,7 @@ async fn main() -> Result<()> { cluster.masters.clone(), advertise, args.filer_group.clone(), + args.metrics_port, cluster.master_tls.clone(), metrics.clone(), )); diff --git a/seaweed-vector/crates/server/src/master_client.rs b/seaweed-vector/crates/server/src/master_client.rs index 09d58aaf29..9890cb5a96 100644 --- a/seaweed-vector/crates/server/src/master_client.rs +++ b/seaweed-vector/crates/server/src/master_client.rs @@ -14,7 +14,7 @@ use crate::master_pb::{ use crate::security::GrpcTls; pub use weed_lance_worker::cluster::{CONNECT_TIMEOUT, connect, grpc_addr}; -fn registration(client_addr: &str, filer_group: &str) -> KeepConnectedRequest { +fn registration(client_addr: &str, filer_group: &str, metrics_port: u16) -> KeepConnectedRequest { KeepConnectedRequest { client_type: "vector".into(), client_address: client_addr.into(), @@ -23,7 +23,7 @@ fn registration(client_addr: &str, filer_group: &str) -> KeepConnectedRequest { data_center: String::new(), rack: String::new(), leave_lock_ring: false, - ..Default::default() + metrics_port: metrics_port.into(), } } @@ -45,16 +45,16 @@ pub async fn keep_connected( masters: Vec, client_addr: String, filer_group: String, + metrics_port: u16, tls: Option, metrics: crate::metrics::Metrics, ) { - let client_addr = client_addr.as_str(); - let filer_group = filer_group.as_str(); + let registration = ®istration(&client_addr, &filer_group, metrics_port); let tls = tls.as_ref(); let metrics = &metrics; loop { try_all_masters(&masters, |master| async move { - let result = stream_to(&master, client_addr, filer_group, tls, Some(&metrics)).await; + let result = stream_to(&master, registration, tls, Some(&metrics)).await; metrics.set_connected(false).await; result }) @@ -94,8 +94,7 @@ where async fn stream_to( master: &str, - client_addr: &str, - filer_group: &str, + registration: &KeepConnectedRequest, tls: Option<&GrpcTls>, metrics: Option<&crate::metrics::Metrics>, ) -> Result> { @@ -104,7 +103,7 @@ async fn stream_to( let answer_by = tokio::time::Instant::now() + CONNECT_TIMEOUT; let mut client = SeaweedClient::new(channel); let (tx, rx) = tokio::sync::mpsc::channel(8); - tx.send(registration(client_addr, filer_group)).await?; + tx.send(registration.clone()).await?; let mut stream = tokio::time::timeout_at( answer_by, client.keep_connected(tokio_stream::wrappers::ReceiverStream::new(rx)), @@ -141,7 +140,7 @@ async fn stream_to( anyhow::bail!("master did not answer keep-connected"); } _ = heartbeat.tick() => { - tokio::time::timeout(CONNECT_TIMEOUT, tx.send(registration(client_addr, filer_group))).await + tokio::time::timeout(CONNECT_TIMEOUT, tx.send(registration.clone())).await .context("registration send stalled")??; probe_registration(&mut client).await?; } @@ -169,11 +168,17 @@ mod tests { #[test] fn registration_preserves_filer_group() { assert_eq!( - registration("vector:8090", "tenant-a").filer_group, + registration("vector:8090", "tenant-a", 0).filer_group, "tenant-a" ); } + #[test] + fn registration_advertises_metrics_port() { + assert_eq!(registration("vector:8090", "", 9327).metrics_port, 9327); + assert_eq!(registration("vector:8090", "", 0).metrics_port, 0); + } + #[tokio::test] async fn stalled_registration_heartbeat_is_bounded() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -242,7 +247,8 @@ mod tests { #[tokio::test] async fn dial_to_a_black_holed_master_is_bounded() { // TEST-NET-1 (RFC 5737) is reserved, so nothing answers the SYN. - let dial = stream_to("192.0.2.1:9333", "127.0.0.1:8080", "", None, None); + let registration = registration("127.0.0.1:8080", "", 0); + let dial = stream_to("192.0.2.1:9333", ®istration, None, None); let result = tokio::time::timeout(CONNECT_TIMEOUT * 3, dial).await; assert!(matches!(result, Ok(Err(_))), "dial was not bounded"); } @@ -256,7 +262,8 @@ mod tests { let _ = tokio::io::copy(&mut conn, &mut tokio::io::sink()).await; }); let master = format!("127.0.0.1:9333.{port}"); - let open = stream_to(&master, "127.0.0.1:8080", "", None, None); + let registration = registration("127.0.0.1:8080", "", 0); + let open = stream_to(&master, ®istration, None, None); match tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await { Ok(Err(e)) => assert!(format!("{e:#}").contains("did not answer"), "{e:#}"), other => panic!("open was not bounded: {other:?}"), @@ -291,7 +298,8 @@ mod tests { }); let master = format!("127.0.0.1:9333.{port}"); let tls = credentials(); - let open = stream_to(&master, "127.0.0.1:8080", "", Some(&tls), None); + let registration = registration("127.0.0.1:8080", "", 0); + let open = stream_to(&master, ®istration, Some(&tls), None); match tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await { Ok(Err(e)) => assert!(format!("{e:#}").contains("did not complete"), "{e:#}"), other => panic!("handshake was not bounded: {other:?}"), @@ -317,7 +325,8 @@ mod tests { let mut tls = credentials(); tls.ca_path = fifo.display().to_string(); let master = format!("127.0.0.1:9333.{port}"); - let open = stream_to(&master, "127.0.0.1:8080", "", Some(&tls), None); + let registration = registration("127.0.0.1:8080", "", 0); + let open = stream_to(&master, ®istration, Some(&tls), None); let result = tokio::time::timeout(CONNECT_TIMEOUT * 3, open).await; use std::io::Write; let mut cleanup = std::fs::OpenOptions::new() diff --git a/weed/pb/master.proto b/weed/pb/master.proto index d1cf3dd342..51e0d5a06e 100644 --- a/weed/pb/master.proto +++ b/weed/pb/master.proto @@ -230,6 +230,10 @@ message KeepConnectedRequest { // A draining filer leaves the lock ring but stays a cluster member, so peers // keep following its metadata until the stream closes. bool leave_lock_ring = 8; + // 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 = 9; } message VolumeLocation { diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index d507d81da6..03f84d45c9 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/weed/pb/master_pb/master.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v7.35.0 +// protoc v3.21.12 // source: master.proto package master_pb @@ -1040,6 +1040,10 @@ type KeepConnectedRequest struct { // A draining filer leaves the lock ring but stays a cluster member, so peers // keep following its metadata until the stream closes. LeaveLockRing bool `protobuf:"varint,8,opt,name=leave_lock_ring,json=leaveLockRing,proto3" json:"leave_lock_ring,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,9,opt,name=metrics_port,json=metricsPort,proto3" json:"metrics_port,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -1123,6 +1127,13 @@ func (x *KeepConnectedRequest) GetLeaveLockRing() bool { return false } +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"` @@ -5303,7 +5314,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\"\xf6\x01\n" + + "volume_ids\x18\x03 \x03(\rR\tvolumeIds\"\x99\x02\n" + "\x14KeepConnectedRequest\x12\x1f\n" + "\vclient_type\x18\x01 \x01(\tR\n" + "clientType\x12%\n" + @@ -5314,7 +5325,8 @@ const file_master_proto_rawDesc = "" + "\vdata_center\x18\x06 \x01(\tR\n" + "dataCenter\x12\x12\n" + "\x04rack\x18\a \x01(\tR\x04rack\x12&\n" + - "\x0fleave_lock_ring\x18\b \x01(\bR\rleaveLockRing\"\x9e\x03\n" + + "\x0fleave_lock_ring\x18\b \x01(\bR\rleaveLockRing\x12!\n" + + "\fmetrics_port\x18\t \x01(\rR\vmetricsPort\"\x9e\x03\n" + "\x0eVolumeLocation\x12\x10\n" + "\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" + "\n" + diff --git a/weed/pb/master_pb/master_grpc.pb.go b/weed/pb/master_pb/master_grpc.pb.go index c697724512..777d6f18f1 100644 --- a/weed/pb/master_pb/master_grpc.pb.go +++ b/weed/pb/master_pb/master_grpc.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go-grpc. DO NOT EDIT. // versions: // - protoc-gen-go-grpc v1.6.2 -// - protoc v7.35.0 +// - protoc v3.21.12 // source: master.proto package master_pb