From 190ab32cfb22e657dc7451aa55c14cb8fd9041e0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 28 Oct 2026 00:58:31 -0700 Subject: [PATCH] vector: weed-vector registers to master; s3 discovers it there (#12146) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * vector: weed-vector registers with the weed master weed-vector-server --masters holds a KeepConnected stream to the weed master, the same mechanism filer and s3 use via wdclient — the server shows up in ListClusterNodes(client_type=vector). Resends registration and rediscovers on failure. * vector: s3 discovers weed-vector through the master With -vectors.grpc unset the vectors gateway resolves one registered weed-vector via ListClusterNodes on the masters it already derives from filer config — discovery replaces the static dial for the single-worker OSS path. The conformance harness registers the engine to mini's master and drops -vectors.grpc. * vector: default-fill registration fields ..Default::default() keeps the source compiling against proto variants that carry extra fields, so the same file works in either tree. * vector: serve the Vectors API on the main s3 port The vectors router sits on the main -port listener behind a dispatch that checks the SigV4 credential scope for /s3vectors/ — no separate port needed. -port.vectors stays for explicit dedicated-port deployments; the conformance harness now hits vectors through :8334. * vector: drop -port.vectors entirely It was new — no compat to preserve. The main -port listener serves the API whenever a weed-vector backend is reachable. * vector: drop -vectors.grpc; discovery is the only locator weed-vector always registers to master, the gateway always resolves it through ListClusterNodes — a static address just forks membership. * vector: binary is weed-vector Matches the enterprise binary; package and lib names are unchanged. * vector: review fixes — cluster type, grpc port, lazy resolve, leader hint, advertise - cluster: register 'vector' as a node type so ListClusterNodes returns weed-vector members - s3: dial masters via ToGrpcAddress (HTTP port + 10000) - s3: backend resolves lazily through vectorDiscovery — workers joining after gateway startup work on first query, and address changes re-dial - master_client: convert master HTTP addr to grpc +10000, honor http(s) scheme, follow the leader hint in KeepConnectedResponse, and --advertise covers wildcard/unreachable listen addresses * vector: no --advertise flag; derive the outbound address A wildcard listen resolves to the source IP the master sees via a route lookup, so registration always advertises something reachable without another flag. * vector: review fixes — plaintext dispatch, conn leaks, discovery backoff The two plaintext http servers now serve the vectors dispatch too — signed s3vectors calls on the main port reach the right router. discoverVectorWorker closes each master conn and bounds lookups to 5s; a dead or absent worker no longer re-hits master on every RPC — the resolver backs off for 5s between rediscovery attempts. * vector: 1 GiB grpc cap, dead-worker invalidation, CLI rename Valid PutVectors batches exceed the tonic 4 MiB default — match the gateway's 1 GiB cap. A replaced worker no longer sticks: any RPC error resets the cached address and the next call rediscovers through the master. The core CLI bin renames to weed-vector-cli so the two crates do not collide on the same output path. * vector: retry-only-transport reset, periodic re-resolve, cli rename - reset() takes the failed client and runs only on transport errors — an InvalidArgument on a healthy conn no longer interrupts unrelated requests, and a concurrent re-resolve wins the race. - cli() re-resolves once the interval passes even with a live client, so a worker that moves addresses is picked up. - verify.rs spike docs use weed-vector-cli after the bin rename. * vector: only Unavailable counts as a transport failure A request deadline or an application error can fire on a healthy connection — resetting the shared conn on those churned reconnects. --- .github/workflows/rust-vector-tests.yml | 2 +- seaweed-vector/Cargo.lock | 1 + seaweed-vector/Cargo.toml | 1 + seaweed-vector/crates/core/Cargo.toml | 2 +- seaweed-vector/crates/core/src/verify.rs | 6 +- seaweed-vector/crates/server/Cargo.toml | 3 +- seaweed-vector/crates/server/build.rs | 10 +- seaweed-vector/crates/server/src/lib.rs | 11 + seaweed-vector/crates/server/src/main.rs | 44 ++- .../crates/server/src/master_client.rs | 91 +++++ test/vectors/Makefile | 6 +- test/vectors/run_test.sh | 7 +- test/vectors/s3vectors_test.py | 6 +- weed/cluster/cluster.go | 11 + weed/command/s3.go | 338 +++++++++++++----- 15 files changed, 423 insertions(+), 116 deletions(-) create mode 100644 seaweed-vector/crates/server/src/master_client.rs diff --git a/.github/workflows/rust-vector-tests.yml b/.github/workflows/rust-vector-tests.yml index fe6b5a2fb9..9bad9cefa3 100644 --- a/.github/workflows/rust-vector-tests.yml +++ b/.github/workflows/rust-vector-tests.yml @@ -120,5 +120,5 @@ jobs: - name: Run the s3vectors conformance subset working-directory: test/vectors run: | - cp ../../seaweed-vector/target/release/weed-vector-server weed-vector-server + cp ../../seaweed-vector/target/release/weed-vector weed-vector ./run_test.sh diff --git a/seaweed-vector/Cargo.lock b/seaweed-vector/Cargo.lock index 8d2f7b6e72..1751228a87 100644 --- a/seaweed-vector/Cargo.lock +++ b/seaweed-vector/Cargo.lock @@ -6497,6 +6497,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "tokio-stream", "tonic", "tonic-prost", "tonic-prost-build", diff --git a/seaweed-vector/Cargo.toml b/seaweed-vector/Cargo.toml index 71648c1621..3632ea83a8 100644 --- a/seaweed-vector/Cargo.toml +++ b/seaweed-vector/Cargo.toml @@ -24,5 +24,6 @@ undocumented_unsafe_blocks = "warn" [workspace.dependencies] anyhow = "1" tokio = { version = "1", features = ["full"] } +tokio-stream = "0.1" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/seaweed-vector/crates/core/Cargo.toml b/seaweed-vector/crates/core/Cargo.toml index 4c645e60fe..8730650e0a 100644 --- a/seaweed-vector/crates/core/Cargo.toml +++ b/seaweed-vector/crates/core/Cargo.toml @@ -9,7 +9,7 @@ description = "Vector query engine over SeaweedFS Lance table buckets" name = "weed_vector_core" [[bin]] -name = "weed-vector" +name = "weed-vector-cli" path = "src/main.rs" [dependencies] diff --git a/seaweed-vector/crates/core/src/verify.rs b/seaweed-vector/crates/core/src/verify.rs index 33eaea498d..5ac65efd11 100644 --- a/seaweed-vector/crates/core/src/verify.rs +++ b/seaweed-vector/crates/core/src/verify.rs @@ -144,9 +144,9 @@ async fn deletes_are_not_returned() { /// $WEED_VECTOR_SPIKE_DIR for exercising the `weed-vector` CLI: /// /// WEED_VECTOR_SPIKE_DIR=/tmp/vds cargo test make_spike_dataset -- --ignored -/// weed-vector describe /tmp/vds -/// weed-vector coverage /tmp/vds -/// weed-vector query /tmp/vds --vector 0,0,0,0 --k 5 +/// weed-vector-cli describe /tmp/vds +/// weed-vector-cli coverage /tmp/vds +/// weed-vector-cli query /tmp/vds --vector 0,0,0,0 --k 5 #[tokio::test] #[ignore] async fn make_spike_dataset() { diff --git a/seaweed-vector/crates/server/Cargo.toml b/seaweed-vector/crates/server/Cargo.toml index 71f7f449a2..2727b46f26 100644 --- a/seaweed-vector/crates/server/Cargo.toml +++ b/seaweed-vector/crates/server/Cargo.toml @@ -9,7 +9,7 @@ description = "VectorService gRPC server over SeaweedFS Lance table buckets" name = "weed_vector_server" [[bin]] -name = "weed-vector-server" +name = "weed-vector" path = "src/main.rs" [dependencies] @@ -32,6 +32,7 @@ futures = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio.workspace = true +tokio-stream.workspace = true tracing.workspace = true tracing-subscriber.workspace = true diff --git a/seaweed-vector/crates/server/build.rs b/seaweed-vector/crates/server/build.rs index d49114f288..d159439de4 100644 --- a/seaweed-vector/crates/server/build.rs +++ b/seaweed-vector/crates/server/build.rs @@ -10,7 +10,15 @@ fn main() -> Result<(), Box> { tonic_prost_build::configure() .build_server(true) .build_client(true) - .compile_protos(&["../../../weed/pb/vector.proto"], &["../../../weed/pb/"])?; + .compile_protos( + &[ + "../../../weed/pb/vector.proto", + "../../../weed/pb/master.proto", + "../../../weed/pb/volume_server.proto", + "../../../weed/pb/remote.proto", + ], + &["../../../weed/pb/"], + )?; println!("cargo:rerun-if-changed=../../../weed/pb/vector.proto"); Ok(()) } diff --git a/seaweed-vector/crates/server/src/lib.rs b/seaweed-vector/crates/server/src/lib.rs index b46863fe45..322ec27dcc 100644 --- a/seaweed-vector/crates/server/src/lib.rs +++ b/seaweed-vector/crates/server/src/lib.rs @@ -2,6 +2,17 @@ pub mod service; +pub mod remote_pb { + tonic::include_proto!("remote_pb"); +} +pub mod volume_server_pb { + tonic::include_proto!("volume_server_pb"); +} +pub mod master_pb { + tonic::include_proto!("master_pb"); +} +pub mod master_client; + pub mod pb { pub mod vector { tonic::include_proto!("vector_pb"); diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index 7c0aa7a623..ca3f9ea077 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -23,6 +23,10 @@ struct Args { /// key=value, repeatable. #[arg(long = "storage-option", value_parser = parse_option)] storage_options: Vec<(String, String)>, + /// Comma-separated master addresses to register this server with; the + /// gateway discovers it via ListClusterNodes(client_type=vector). + #[arg(long, env = "WEED_VECTOR_MASTERS")] + masters: Option, } fn parse_option(s: &str) -> Result<(String, String), String> { @@ -31,16 +35,52 @@ fn parse_option(s: &str) -> Result<(String, String), String> { .ok_or_else(|| "expected key=value".to_string()) } +/// A wildcard listen address is not reachable for peers; derive the outbound +/// IP toward the first master instead — the address the master sees anyway. +fn advertised_addr(listen: &str, masters: &[String]) -> String { + let Some((host, _)) = listen.rsplit_once(':') else { + return listen.to_string(); + }; + if !matches!(host, "0.0.0.0" | "::" | "" | "[::]") { + return listen.to_string(); + } + let port = listen.rsplit_once(':').map(|(_, p)| p).unwrap_or_default(); + if let Some(master) = masters.first() { + if let Ok(sock) = std::net::UdpSocket::bind((host, 0u16)) { + if sock.connect(master).is_ok() { + if let Ok(local) = sock.local_addr() { + return format!("{}:{}", local.ip(), port); + } + } + } + } + listen.to_string() +} + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt::init(); let args = Args::parse(); + if let Some(masters) = &args.masters { + let masters: Vec = masters.split(',').map(|s| s.trim().to_string()).collect(); + let advertise = advertised_addr(&args.listen, &masters); + tokio::spawn(weed_vector_server::master_client::keep_connected( + masters, + advertise, + )); + } let fallback: FallbackOptions = args.storage_options.into_iter().collect(); let client = NamespaceClient::new(args.namespace.clone()).with_fallback(&fallback); let service = VectorServer::new(client, fallback); - tracing::info!(namespace = %args.namespace, listen = %args.listen, "weed-vector-server"); + tracing::info!(namespace = %args.namespace, listen = %args.listen, "weed-vector"); Server::builder() - .add_service(VectorServiceServer::new(service)) + // Match the gateway's 1 GiB cap — put batches over the tonic 4 MiB + // default otherwise get rejected before they reach the handler. + .add_service( + VectorServiceServer::new(service) + .max_decoding_message_size(1 << 30) + .max_encoding_message_size(1 << 30), + ) .serve(args.listen.parse()?) .await?; Ok(()) diff --git a/seaweed-vector/crates/server/src/master_client.rs b/seaweed-vector/crates/server/src/master_client.rs new file mode 100644 index 0000000000..11f983f345 --- /dev/null +++ b/seaweed-vector/crates/server/src/master_client.rs @@ -0,0 +1,91 @@ +//! Self-registration with the weed master, mirroring how filer/s3 keep +//! connected: hold a KeepConnected stream open to the leader and the node +//! shows up in ListClusterNodes(client_type="vector"). + +use std::time::Duration; + +use anyhow::Result; +use tokio::time::sleep; +use tonic::transport::Endpoint; + +use crate::master_pb::{seaweed_client::SeaweedClient, KeepConnectedRequest}; + +fn registration(client_addr: &str) -> KeepConnectedRequest { + KeepConnectedRequest { + client_type: "vector".into(), + client_address: client_addr.into(), + version: env!("CARGO_PKG_VERSION").into(), + filer_group: String::new(), + data_center: String::new(), + rack: String::new(), + leave_lock_ring: false, + ..Default::default() + } +} + +/// Masters are configured by their HTTP address; gRPC lives +10000 like the +/// weed conventions (server_address.ToGrpcAddress). +fn grpc_addr(master: &str) -> String { + let (scheme, rest) = master.split_once("://").unwrap_or(("http", master)); + if let Some((host, port)) = rest.rsplit_once(':') { + if let Ok(port) = port.parse::() { + return format!("{scheme}://{host}:{}", port + 10000); + } + } + format!("{scheme}://{rest}") +} + +/// Retry forever: try each master until a stream opens, hold it until it +/// errors, then rediscover — same shape as wdclient.KeepConnectedToMaster. +/// A follower response's leader hint takes priority on the next pass. +pub async fn keep_connected(masters: Vec, client_addr: String) { + let mut leader: Option = None; + loop { + let mut seeds: Vec = Vec::new(); + if let Some(l) = &leader { + seeds.push(l.clone()); + } + seeds.extend(masters.iter().cloned()); + for master in &seeds { + match stream_to(master, &client_addr).await { + Ok(Some(l)) => { + leader = Some(l); + break; + } + Ok(None) => break, + Err(e) => tracing::warn!("master {master}: {e:#}"), + } + } + sleep(Duration::from_secs(3)).await; + } +} + +async fn stream_to(master: &str, client_addr: &str) -> Result> { + let channel = Endpoint::new(grpc_addr(master))?.connect().await?; + let mut client = SeaweedClient::new(channel); + let (tx, rx) = tokio::sync::mpsc::channel(8); + tx.send(registration(client_addr)).await?; + let mut stream = client + .keep_connected(tokio_stream::wrappers::ReceiverStream::new(rx)) + .await? + .into_inner(); + // The master keeps the node live while the stream is open; resend the + // registration periodically so a reconnect never leaves a stale entry. + loop { + tokio::select! { + msg = stream.message() => match msg? { + Some(resp) => { + if let Some(loc) = resp.volume_location { + if !loc.leader.is_empty() { + return Ok(Some(loc.leader)); + } + } + } + None => anyhow::bail!("master closed the keep-connected stream"), + }, + _ = sleep(Duration::from_secs(5)) => { + tx.send(registration(client_addr)).await?; + } + } + } +} diff --git a/test/vectors/Makefile b/test/vectors/Makefile index 166bf720d5..fe222c9fc0 100644 --- a/test/vectors/Makefile +++ b/test/vectors/Makefile @@ -1,8 +1,8 @@ # End-to-end S3 Vectors conformance run: # weed mini (cluster + lance catalog) + weed s3 (s3vectors port) + -# weed-vector-server (VectorService gRPC), then boto3 s3vectors calls. +# weed-vector (VectorService gRPC), then boto3 s3vectors calls. WEED := ./weed -VEC := ./weed-vector-server +VEC := ./weed-vector .PHONY: all test build clean FORCE @@ -14,7 +14,7 @@ $(WEED): FORCE $(VEC): FORCE cargo build --release --manifest-path ../../seaweed-vector/Cargo.toml -p weed-vector-server - cp ../../seaweed-vector/target/release/weed-vector-server $(VEC) + cp ../../seaweed-vector/target/release/weed-vector $(VEC) FORCE: diff --git a/test/vectors/run_test.sh b/test/vectors/run_test.sh index cb2d72c7b3..ab3826f580 100755 --- a/test/vectors/run_test.sh +++ b/test/vectors/run_test.sh @@ -23,13 +23,12 @@ PIDS+=($!) -ip 127.0.0.1 \ -ip.bind 127.0.0.1 \ -port 8334 \ - -port.lance 9102 \ - -port.vectors 9201 \ - -vectors.grpc 127.0.0.1:9301 & + -port.lance 9102 & PIDS+=($!) -./weed-vector-server \ +./weed-vector \ --namespace http://127.0.0.1:9102 \ + --masters 127.0.0.1:9333 \ --storage-option aws_access_key_id=test \ --storage-option aws_secret_access_key=test \ --listen 127.0.0.1:9301 & diff --git a/test/vectors/s3vectors_test.py b/test/vectors/s3vectors_test.py index 422ab7faaa..a6857b13ef 100644 --- a/test/vectors/s3vectors_test.py +++ b/test/vectors/s3vectors_test.py @@ -1,13 +1,13 @@ """Boto3 conformance subset for the S3 Vectors API gateway. -Run via `make test` — a mini cluster, an s3 gateway with -port.vectors, and a -weed-vector-server are up by then. +Run via `make test` — a mini cluster, an s3 gateway serving the vectors API on its main port, and a +weed-vector are up by then. """ import boto3 import pytest -ENDPOINT = "http://127.0.0.1:9201" +ENDPOINT = "http://127.0.0.1:8334" BUCKET = "vectors-bucket" INDEX = "idx" diff --git a/weed/cluster/cluster.go b/weed/cluster/cluster.go index 5ee4e1c4cf..7436474dce 100644 --- a/weed/cluster/cluster.go +++ b/weed/cluster/cluster.go @@ -14,6 +14,7 @@ const ( FilerType = "filer" BrokerType = "broker" S3Type = "s3" + VectorType = "vector" ) type FilerGroupName string @@ -37,6 +38,7 @@ type Cluster struct { filerGroups *ClusterNodeGroups brokerGroups *ClusterNodeGroups s3Groups *ClusterNodeGroups + vectorGroups *ClusterNodeGroups } func newClusterNodeGroups() *ClusterNodeGroups { @@ -92,6 +94,7 @@ func NewCluster() *Cluster { filerGroups: newClusterNodeGroups(), brokerGroups: newClusterNodeGroups(), s3Groups: newClusterNodeGroups(), + vectorGroups: newClusterNodeGroups(), } } @@ -104,6 +107,8 @@ func (cluster *Cluster) AddClusterNode(ns, nodeType string, dataCenter DataCente return cluster.brokerGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version) case S3Type: return cluster.s3Groups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version) + case VectorType: + return cluster.vectorGroups.AddClusterNode(filerGroup, nodeType, dataCenter, rack, address, version) case MasterType: return buildClusterNodeUpdateMessage(true, filerGroup, nodeType, address) } @@ -119,6 +124,8 @@ func (cluster *Cluster) RemoveClusterNode(ns string, nodeType string, address pb return cluster.brokerGroups.RemoveClusterNode(filerGroup, nodeType, address) case S3Type: return cluster.s3Groups.RemoveClusterNode(filerGroup, nodeType, address) + case VectorType: + return cluster.vectorGroups.RemoveClusterNode(filerGroup, nodeType, address) case MasterType: return buildClusterNodeUpdateMessage(false, filerGroup, nodeType, address) } @@ -133,6 +140,8 @@ func (cluster *Cluster) ListClusterNode(filerGroup FilerGroupName, nodeType stri return cluster.brokerGroups.ListClusterNode(filerGroup) case S3Type: return cluster.s3Groups.ListClusterNode(filerGroup) + case VectorType: + return cluster.vectorGroups.ListClusterNode(filerGroup) case MasterType: } return @@ -160,6 +169,8 @@ func (cluster *Cluster) IsKnownNode(nodeType string, address pb.ServerAddress) b groups = cluster.brokerGroups case S3Type: groups = cluster.s3Groups + case VectorType: + groups = cluster.vectorGroups default: return false } diff --git a/weed/command/s3.go b/weed/command/s3.go index e2720518c5..6d2802246b 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -5,12 +5,12 @@ import ( "crypto/tls" "crypto/x509" "fmt" - "net" "net/http" "os" "runtime" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -20,6 +20,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_pb" "github.com/seaweedfs/seaweedfs/weed/pb/vector_pb" @@ -36,6 +37,8 @@ import ( util_http "github.com/seaweedfs/seaweedfs/weed/util/http" "github.com/seaweedfs/seaweedfs/weed/util/version" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) var ( @@ -55,8 +58,6 @@ type S3Options struct { portGrpc *int portIceberg *int portLance *int - portVectors *int - vectorsGrpc *string icebergCredentialRole *string icebergCredentialDuration *int config *string @@ -115,8 +116,6 @@ func init() { s3StandaloneOptions.portGrpc = cmdS3.Flag.Int("port.grpc", 0, "s3 server grpc listen port") s3StandaloneOptions.portIceberg = cmdS3.Flag.Int("port.iceberg", 8181, "Iceberg REST Catalog server listen port (0 to disable)") s3StandaloneOptions.portLance = cmdS3.Flag.Int("port.lance", 9101, "Lance Namespace server listen port (0 to disable); credential vending uses -iceberg.credentialRole") - s3StandaloneOptions.portVectors = cmdS3.Flag.Int("port.vectors", 0, "S3 Vectors API server listen port (0 to disable); requires -vectors.grpc") - s3StandaloneOptions.vectorsGrpc = cmdS3.Flag.String("vectors.grpc", "", "weed-vector-server gRPC address for the S3 Vectors API") s3StandaloneOptions.icebergCredentialRole = cmdS3.Flag.String("iceberg.credentialRole", "", "IAM role ARN the Iceberg catalog assumes to vend table-scoped credentials (empty disables vending)") s3StandaloneOptions.icebergCredentialDuration = cmdS3.Flag.Int("iceberg.credentialDurationSeconds", 3600, "lifetime of credentials vended by the Iceberg catalog") s3StandaloneOptions.domainName = cmdS3.Flag.String("domainName", "", "suffix of the host name in comma separated list, {bucket}.{domainName}") @@ -440,11 +439,19 @@ func (s3opt *S3Options) startS3Server() bool { } } - // Start S3 Vectors API server if enabled - if s3opt.portVectors != nil && *s3opt.portVectors > 0 { - if httpS := s3opt.startVectorsServer(s3ApiServer, grpcDialOption, listenOnLocalhost); httpS != nil { - httpServers = append(httpServers, httpS) - } + // The Vectors API rides the main -port listener whenever a weed-vector + // backend is reachable — /s3vectors/ in the SigV4 credential scope picks + // the vectors router. + vectorsHandler := http.Handler(router) + if vectorsRouter := s3opt.vectorsRouter(s3ApiServer, grpcDialOption, masterAddresses); vectorsRouter != nil { + vr, mainRouter := vectorsRouter, router + vectorsHandler = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if isVectorsRequest(r) { + vr.ServeHTTP(w, r) + return + } + mainRouter.ServeHTTP(w, r) + }) } if runtime.GOOS != "windows" { @@ -455,7 +462,7 @@ func (s3opt *S3Options) startS3Server() bool { } if s3SocketListener := pb.ListenOnLocalSocket(localSocket); s3SocketListener != nil { s3SocketListener = &util.Listener{Listener: s3SocketListener, Timeout: idleTimeoutSeconds(s3opt.idleTimeout)} - socketServer := newHttpServer(router, nil) + socketServer := newHttpServer(vectorsHandler, nil) httpServers = append(httpServers, socketServer) go func() { if err := socketServer.Serve(s3SocketListener); err != nil && err != http.ErrServerClosed { @@ -530,7 +537,7 @@ func (s3opt *S3Options) startS3Server() bool { if *s3opt.portHttps == 0 { glog.V(0).Infof("Start Seaweed S3 API Server %s at https port %d", version.Version(), *s3opt.port) if s3ApiLocalListener != nil { - localServer := newHttpServer(router, tlsConfig) + localServer := newHttpServer(vectorsHandler, tlsConfig) httpServers = append(httpServers, localServer) go func() { if err := localServer.ServeTLS(s3ApiLocalListener, "", ""); err != nil && err != http.ErrServerClosed { @@ -538,7 +545,7 @@ func (s3opt *S3Options) startS3Server() bool { } }() } - httpS := newHttpServer(router, tlsConfig) + httpS := newHttpServer(vectorsHandler, tlsConfig) httpServers = append(httpServers, httpS) shutdown := s3opt.newShutdown(stopGrpcServer, s3ApiServer, httpServers) if err = httpS.ServeTLS(s3ApiListener, "", ""); err != nil && err != http.ErrServerClosed { @@ -554,7 +561,7 @@ func (s3opt *S3Options) startS3Server() bool { glog.Fatalf("S3 API HTTPS listener on %s:%d error: %v", *s3opt.bindIp, *s3opt.portHttps, err) } if s3ApiLocalListenerHttps != nil { - localHttpsServer := newHttpServer(router, tlsConfig) + localHttpsServer := newHttpServer(vectorsHandler, tlsConfig) httpServers = append(httpServers, localHttpsServer) go func() { if err := localHttpsServer.ServeTLS(s3ApiLocalListenerHttps, "", ""); err != nil && err != http.ErrServerClosed { @@ -562,7 +569,7 @@ func (s3opt *S3Options) startS3Server() bool { } }() } - httpsServer := newHttpServer(router, tlsConfig) + httpsServer := newHttpServer(vectorsHandler, tlsConfig) httpServers = append(httpServers, httpsServer) go func() { if err := httpsServer.ServeTLS(s3ApiListenerHttps, "", ""); err != nil && err != http.ErrServerClosed { @@ -574,7 +581,7 @@ func (s3opt *S3Options) startS3Server() bool { if *s3opt.tlsPrivateKey == "" || *s3opt.portHttps > 0 { glog.V(0).Infof("Start Seaweed S3 API Server %s at http port %d", version.Version(), *s3opt.port) if s3ApiLocalListener != nil { - localServer := newHttpServer(router, nil) + localServer := newHttpServer(vectorsHandler, nil) httpServers = append(httpServers, localServer) go func() { if err := localServer.Serve(s3ApiLocalListener); err != nil && err != http.ErrServerClosed { @@ -582,7 +589,7 @@ func (s3opt *S3Options) startS3Server() bool { } }() } - httpS := newHttpServer(router, nil) + httpS := newHttpServer(vectorsHandler, nil) httpServers = append(httpServers, httpS) shutdown := s3opt.newShutdown(stopGrpcServer, s3ApiServer, httpServers) if err = httpS.Serve(s3ApiListener); err != nil && err != http.ErrServerClosed { @@ -702,107 +709,221 @@ func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer, listenO return httpS } -// startVectorsServer starts the S3 Vectors API server on a separate port. It -// fans index and data calls out to the shared VectorService over gRPC; with no -// -vectors.grpc the API is not served. -func (s3opt *S3Options) startVectorsServer(s3ApiServer *s3api.S3ApiServer, grpcDialOption grpc.DialOption, listenOnLocalhost bool) *http.Server { - if *s3opt.vectorsGrpc == "" { - glog.Warningf("port.vectors set without -vectors.grpc; S3 Vectors API disabled") - return nil - } - conn, err := pb.GrpcDial(context.Background(), *s3opt.vectorsGrpc, false, grpcDialOption) - if err != nil { - glog.Errorf("S3 Vectors: dial %s: %v", *s3opt.vectorsGrpc, err) +// vectorsRouter builds the S3 Vectors API router — workers register to the +// master and the backend resolves them lazily, so a worker that comes up +// after the gateway still gets picked up on first query. +func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOption grpc.DialOption, masterAddresses []pb.ServerAddress) *mux.Router { + if len(masterAddresses) == 0 { + glog.Warningf("S3 Vectors API disabled: no weed-vector registered on master") return nil } + backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption} vectorsRouter := mux.NewRouter().SkipClean(true) - vectorsServer := s3vectors.NewServer(s3ApiServer, s3ApiServer, vector_pb.NewVectorServiceClient(conn)) + vectorsServer := s3vectors.NewServer(s3ApiServer, s3ApiServer, backend) vectorsServer.SetRegion(s3tables.DefaultRegion) vectorsServer.SetAccountID(s3tables.DefaultAccountID) vectorsServer.RegisterRoutes(vectorsRouter) + return vectorsRouter +} - // Mirror the main API: when TLS is configured the vectors port serves - // HTTPS, not a second plaintext endpoint. Build the TLS config before - // binding listeners so a certificate failure does not leave ports bound. - var tlsConfig *tls.Config - if *s3opt.tlsPrivateKey != "" { - // The reloading provider reads the key pair asynchronously, so validate - // it once up front: a missing or mismatched pair must fail here rather - // than after the listeners bind an endpoint that cannot handshake. - if _, err := tls.LoadX509KeyPair(*s3opt.tlsCertificate, *s3opt.tlsPrivateKey); err != nil { - glog.Errorf("S3 Vectors: load HTTPS certificate: %v", err) - conn.Close() - return nil - } - getCert, _, err := security.NewReloadingServerCertificate(*s3opt.tlsCertificate, *s3opt.tlsPrivateKey) +// vectorDiscovery resolves the currently registered weed-vector through the +// master and caches the connection; workers joining after gateway startup +// are picked up on first query. +type vectorDiscovery struct { + masters []pb.ServerAddress + grpcDialOption grpc.DialOption + mu sync.Mutex + worker string + conn *grpc.ClientConn + client vector_pb.VectorServiceClient + nextResolve time.Time +} + +// reset drops the cached worker so the next call rediscovers — invoked on +// transport errors only, and only when the failed client is still the +// cached one (a concurrent request may have already re-resolved). +func (d *vectorDiscovery) reset(c vector_pb.VectorServiceClient) { + d.mu.Lock() + defer d.mu.Unlock() + if c != d.client { + return + } + if d.conn != nil { + d.conn.Close() + } + d.worker, d.conn, d.client = "", nil, nil + d.nextResolve = time.Time{} +} + +// Only Unavailable proves a transport failure — a request deadline or an +// application-level error can fire on a healthy connection. +func isTransportError(err error) bool { + return status.Code(err) == codes.Unavailable +} + +func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) { + d.mu.Lock() + defer d.mu.Unlock() + worker := discoverVectorWorker(d.masters, d.grpcDialOption) + if worker == "" { + return nil, fmt.Errorf("no weed-vector registered on master") + } + if worker != d.worker { + if d.conn != nil { + d.conn.Close() + } + conn, err := pb.GrpcDial(context.Background(), worker, false, d.grpcDialOption) if err != nil { - glog.Errorf("S3 Vectors: load HTTPS certificate: %v", err) - conn.Close() - return nil - } - clientAuth := tls.NoClientCert - var caCertPool *x509.CertPool - if *s3opt.tlsVerifyClientCert { - clientAuth = tls.RequireAndVerifyClientCert - caCertPool, err = loadClientCAPool(*s3opt.tlsCACertificate) - if err != nil { - glog.Errorf("S3 Vectors: %v", err) - conn.Close() - return nil - } - } - tlsConfig = &tls.Config{ - GetCertificate: getCert, - ClientAuth: clientAuth, - ClientCAs: caCertPool, - } - if err := security.FixTlsConfig(util.GetViper(), tlsConfig); err != nil { - glog.Errorf("S3 Vectors: tls config: %v", err) - conn.Close() - return nil + return nil, err } + d.worker, d.conn = worker, conn + d.client = vector_pb.NewVectorServiceClient(conn) } + return d.client, nil +} - listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.portVectors) - vectorsListener, vectorsLocalListener, err := util.NewIpAndLocalListeners( - *s3opt.bindIp, *s3opt.portVectors, idleTimeoutSeconds(s3opt.idleTimeout), listenOnLocalhost) - if err != nil { - glog.Errorf("S3 Vectors listener on %s error: %v", listenAddress, err) - conn.Close() - return nil +const vectorResolveInterval = 5 * time.Second + +func (d *vectorDiscovery) cli() (vector_pb.VectorServiceClient, error) { + d.mu.Lock() + cli := d.client + past := !time.Now().Before(d.nextResolve) + if cli != nil && !past { + d.mu.Unlock() + return cli, nil + } + if !past && cli == nil { + d.mu.Unlock() + return nil, fmt.Errorf("no weed-vector registered on master") } + d.nextResolve = time.Now().Add(vectorResolveInterval) + d.mu.Unlock() + return d.resolve() +} - if tlsConfig != nil { - glog.V(0).Infof("Start S3 Vectors API Server at https://%s", listenAddress) - } else { - glog.V(0).Infof("Start S3 Vectors API Server at http://%s", listenAddress) +func (d *vectorDiscovery) CreateIndex(ctx context.Context, in *vector_pb.CreateIndexRequest, opts ...grpc.CallOption) (resp *vector_pb.CreateIndexResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.CreateIndex(ctx, in, opts...); err == nil { + return resp, nil + } + if !isTransportError(err) { + return nil, err + } + d.reset(c) } - - httpS := newHttpServer(vectorsRouter, tlsConfig) - serve := func(listener net.Listener) error { - if tlsConfig != nil { - return httpS.ServeTLS(listener, "", "") + return nil, err +} +func (d *vectorDiscovery) DropIndex(ctx context.Context, in *vector_pb.DropIndexRequest, opts ...grpc.CallOption) (resp *vector_pb.DropIndexResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.DropIndex(ctx, in, opts...); err == nil { + return resp, nil + } + if !isTransportError(err) { + return nil, err + } + d.reset(c) + } + return nil, err +} +func (d *vectorDiscovery) PutVectors(ctx context.Context, in *vector_pb.PutVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.PutVectorsResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.PutVectors(ctx, in, opts...); err == nil { + return resp, nil + } + if !isTransportError(err) { + return nil, err } - return httpS.Serve(listener) + d.reset(c) } - if vectorsLocalListener != nil { - go func() { - if err := serve(vectorsLocalListener); err != nil && err != http.ErrServerClosed { - glog.V(0).Infof("S3 Vectors localhost listener error: %v", err) - } - }() + return nil, err +} +func (d *vectorDiscovery) GetVectors(ctx context.Context, in *vector_pb.GetVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.GetVectorsResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.GetVectors(ctx, in, opts...); err == nil { + return resp, nil + } + if !isTransportError(err) { + return nil, err + } + d.reset(c) } - go func() { - if err := serve(vectorsListener); err != nil && err != http.ErrServerClosed { - glog.Errorf("S3 Vectors API Server Fail to serve: %v", err) + return nil, err +} +func (d *vectorDiscovery) DeleteVectors(ctx context.Context, in *vector_pb.DeleteVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.DeleteVectorsResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.DeleteVectors(ctx, in, opts...); err == nil { + return resp, nil + } + if !isTransportError(err) { + return nil, err + } + d.reset(c) + } + return nil, err +} +func (d *vectorDiscovery) ListVectors(ctx context.Context, in *vector_pb.ListVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.ListVectorsResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.ListVectors(ctx, in, opts...); err == nil { + return resp, nil } - }() - return httpS + if !isTransportError(err) { + return nil, err + } + d.reset(c) + } + return nil, err +} +func (d *vectorDiscovery) QueryVectors(ctx context.Context, in *vector_pb.QueryVectorsRequest, opts ...grpc.CallOption) (resp *vector_pb.QueryVectorsResponse, err error) { + for attempt := 0; attempt < 2; attempt++ { + var c vector_pb.VectorServiceClient + if c, err = d.cli(); err != nil { + return nil, err + } + if resp, err = c.QueryVectors(ctx, in, opts...); err == nil { + return resp, nil + } + if !isTransportError(err) { + return nil, err + } + d.reset(c) + } + return nil, err +} + +// isVectorsRequest routes by the SigV4 credential scope — the s3vectors +// service name lives there, not in the path. +func isVectorsRequest(r *http.Request) bool { + if v := r.Header.Get("Authorization"); v != "" { + return strings.Contains(v, "/s3vectors/") + } + return strings.Contains(r.URL.Query().Get("X-Amz-Credential"), "/s3vectors/") } -// loadClientCAPool builds the client CA pool from -cacert.file, failing when -// the file reads but carries no usable certificate. func loadClientCAPool(caCertFile string) (*x509.CertPool, error) { pool := x509.NewCertPool() if caCertFile == "" { @@ -928,3 +1049,26 @@ func (v lanceCredentialVendor) VendTableCredentials(ctx context.Context, princip Expiration: credentials.Expiration, }, nil } + +// discoverVectorWorker resolves one registered weed-vector server through the +// master's cluster-nodes list; OSS fans every vector call to that single +// backend. +func discoverVectorWorker(masters []pb.ServerAddress, grpcDialOption grpc.DialOption) string { + for _, master := range masters { + conn, err := pb.GrpcDial(context.Background(), master.ToGrpcAddress(), false, grpcDialOption) + if err != nil { + continue + } + resp, err := master_pb.NewSeaweedClient(conn).ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ + ClientType: "vector", + }) + if err != nil { + continue + } + for _, node := range resp.ClusterNodes { + glog.V(0).Infof("S3 Vectors: discovered weed-vector at %s via master %s", node.Address, master) + return node.Address + } + } + return "" +}