diff --git a/test/vectors/run_test.sh b/test/vectors/run_test.sh index ab3826f580..3a44eb9009 100755 --- a/test/vectors/run_test.sh +++ b/test/vectors/run_test.sh @@ -48,4 +48,18 @@ for _ in $(seq 1 60); do done (exec 3<>/dev/tcp/127.0.0.1/9301) 2>/dev/null || { echo "vector server on :9301 never came up" >&2; exit 1; } +# A listening engine still has to land its KeepConnected registration before +# the gateway's lazy discovery can find it — poll the cluster list for the +# vector node rather than race the first request against it. +vector_registered() { + # grep -q would close the pipe on first match and SIGPIPE the shell mid-write + # under `set -o pipefail`; a full read lets cluster.ps finish. + printf 'cluster.ps\n' | ./weed shell -master 127.0.0.1:9333 2>/dev/null | grep "vector servers" >/dev/null +} +for _ in $(seq 1 60); do + vector_registered && break + sleep 1 +done +vector_registered || { echo "vector server never registered with the master" >&2; exit 1; } + python3 -m pytest -xvs s3vectors_test.py diff --git a/weed/command/s3.go b/weed/command/s3.go index cb9f4d8b33..972b83bafd 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -717,7 +717,7 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp glog.Warningf("S3 Vectors API disabled: no weed-vector registered on master") return nil } - backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption} + backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption, discover: discoverVectorWorkers} vectorsRouter := mux.NewRouter().SkipClean(true) vectorsServer := s3vectors.NewServer(s3ApiServer, s3ApiServer, backend) @@ -733,6 +733,7 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp type vectorDiscovery struct { masters []pb.ServerAddress grpcDialOption grpc.DialOption + discover func([]pb.ServerAddress, grpc.DialOption) []string mu sync.Mutex worker string conn *grpc.ClientConn @@ -765,9 +766,9 @@ func isTransportError(err error) bool { func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) { d.mu.Lock() defer d.mu.Unlock() - workers := discoverVectorWorkers(d.masters, d.grpcDialOption) + workers := d.discover(d.masters, d.grpcDialOption) if len(workers) == 0 { - return nil, fmt.Errorf("no weed-vector registered on master") + return nil, errNoVectorWorker } worker := workers[0] for _, w := range workers { @@ -786,27 +787,41 @@ func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) { } d.worker, d.conn = worker, conn d.client = vector_pb.NewVectorServiceClient(conn) + // cli() scheduled this resolve on the short recheck window while no + // client existed; a live client only refreshes every resolve interval. + d.nextResolve = time.Now().Add(vectorResolveInterval) } return d.client, nil } -const vectorResolveInterval = 5 * time.Second +// A live client refreshes membership at most every vectorResolveInterval. +// With no usable backend the recheck is much more frequent: a worker that +// registers a moment after a lookup must surface inside a caller's retry +// budget, while an absent one still cannot flood the master with lookups. +const ( + vectorResolveInterval = 5 * time.Second + vectorRecheckInterval = 500 * time.Millisecond +) + +var errNoVectorWorker = status.Error(codes.Unavailable, "no weed-vector registered on master") func (d *vectorDiscovery) cli() (vector_pb.VectorServiceClient, error) { d.mu.Lock() cli := d.client - past := !time.Now().Before(d.nextResolve) - if cli != nil && !past { + if !time.Now().Before(d.nextResolve) { + interval := vectorRecheckInterval + if cli != nil { + interval = vectorResolveInterval + } + d.nextResolve = time.Now().Add(interval) d.mu.Unlock() + return d.resolve() + } + d.mu.Unlock() + if cli != nil { 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() + return nil, errNoVectorWorker } func (d *vectorDiscovery) CreateIndex(ctx context.Context, in *vector_pb.CreateIndexRequest, opts ...grpc.CallOption) (resp *vector_pb.CreateIndexResponse, err error) { diff --git a/weed/command/s3_vector_discovery_test.go b/weed/command/s3_vector_discovery_test.go new file mode 100644 index 0000000000..3c6f561995 --- /dev/null +++ b/weed/command/s3_vector_discovery_test.go @@ -0,0 +1,61 @@ +package command + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +// A worker that registers a moment after a failed lookup must surface inside +// a caller's retry budget: the no-client recheck window is much shorter than +// the live-client membership refresh. +func TestVectorDiscoveryRechecksRegistration(t *testing.T) { + var calls int32 + d := &vectorDiscovery{ + discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + atomic.AddInt32(&calls, 1) + return nil + }, + } + + if _, err := d.cli(); status.Code(err) != codes.Unavailable { + t.Fatalf("no registered worker: expected Unavailable, got %v", err) + } + if _, err := d.cli(); status.Code(err) != codes.Unavailable { + t.Fatalf("no registered worker: expected Unavailable, got %v", err) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("lookup inside the recheck window hit master again: %d lookups", got) + } + + // Once the recheck interval passes the next call re-discovers. + d.mu.Lock() + d.nextResolve = time.Now().Add(-time.Second) + d.mu.Unlock() + d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + atomic.AddInt32(&calls, 1) + return []string{"127.0.0.1:9301"} + } + d.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) + + if _, err := d.cli(); err != nil { + t.Fatalf("worker registered: expected a client, got %v", err) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Fatalf("expired recheck window did not rediscover: %d lookups", got) + } + + // A live client rides out the refresh interval without rediscovery. + if _, err := d.cli(); err != nil { + t.Fatalf("cached client rejected: %v", err) + } + if got := atomic.LoadInt32(&calls); got != 2 { + t.Fatalf("live client rediscovered inside the refresh window: %d lookups", got) + } +} diff --git a/weed/shell/command_cluster_ps.go b/weed/shell/command_cluster_ps.go index 0c1e04fa8f..0e1d3c4dff 100644 --- a/weed/shell/command_cluster_ps.go +++ b/weed/shell/command_cluster_ps.go @@ -75,6 +75,10 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W if err != nil { return } + vectorNodes, err := listClusterNodes(cluster.VectorType) + if err != nil { + return + } if len(mqBrokerNodes) > 0 { fmt.Fprintf(writer, "* message queue brokers %d\n", len(mqBrokerNodes)) @@ -102,6 +106,19 @@ func (c *commandClusterPs) Do(args []string, commandEnv *CommandEnv, writer io.W } } + if len(vectorNodes) > 0 { + fmt.Fprintf(writer, "* vector servers %d\n", len(vectorNodes)) + for _, node := range vectorNodes { + fmt.Fprintf(writer, " * %s (%v)\n", node.Address, node.Version) + if node.DataCenter != "" { + fmt.Fprintf(writer, " DataCenter: %v\n", node.DataCenter) + } + if node.Rack != "" { + fmt.Fprintf(writer, " Rack: %v\n", node.Rack) + } + } + } + filerSignatures := make(map[*master_pb.ListClusterNodesResponse_ClusterNode]int32) fmt.Fprintf(writer, "* filers %d\n", len(filerNodes)) for _, node := range filerNodes {