diff --git a/.gitignore b/.gitignore index dc858fdb1e..91a1c58104 100644 --- a/.gitignore +++ b/.gitignore @@ -147,7 +147,6 @@ telemetry/server/telemetry-server __pycache__/ /test/s3vectors/weed /test/s3vectors/weed-vector -/test/s3vectors/weed-vector-worker /test/s3vectors/weed-vector-cli /test/s3vectors/weed-worker /test/s3vectors/weed-volume diff --git a/test/s3vectors/Makefile b/test/s3vectors/Makefile index 4ac15246a9..d1c3275b08 100644 --- a/test/s3vectors/Makefile +++ b/test/s3vectors/Makefile @@ -1,10 +1,7 @@ WEED := ./weed VEC := ./weed-vector CLI := ./weed-vector-cli -WORKER_MANIFEST := ../../rust/sw-vector-distributed/Cargo.toml -ifneq ($(wildcard $(WORKER_MANIFEST)),) -WORKER := ./weed-vector-worker -endif +ENGINE_MANIFEST := ../../rust/sw-vector-distributed/Cargo.toml VOLUME := ./weed-volume LANCE_WORKER := ./weed-worker CARGO_PROFILE ?= release @@ -21,7 +18,11 @@ $(WEED): FORCE go build -o $(WEED) ../../weed $(VEC): FORCE +ifneq ($(wildcard $(ENGINE_MANIFEST)),) + cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../rust/Cargo.toml -p sw-vector-distributed --bin weed-vector +else cd ../../seaweed-vector && cargo build --locked --profile $(CARGO_PROFILE) -p weed-vector-server +endif cp $(CARGO_OUTPUT)/weed-vector $(VEC) $(CLI): FORCE @@ -32,19 +33,13 @@ $(LANCE_WORKER): FORCE cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../seaweed-worker/Cargo.toml -p weed-lance-worker --bin weed-worker cp $(CARGO_OUTPUT)/weed-worker $(LANCE_WORKER) -ifneq ($(WORKER),) -$(WORKER): $(VEC) FORCE - cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../rust/Cargo.toml -p sw-vector-distributed --bin weed-vector - cp $(CARGO_OUTPUT)/weed-vector $(WORKER) -endif - -$(VOLUME): $(VEC) $(WORKER) FORCE +$(VOLUME): $(VEC) FORCE cargo build --locked --profile $(CARGO_PROFILE) --manifest-path ../../seaweed-volume/Cargo.toml --bin weed-volume cp $(CARGO_OUTPUT)/weed-volume $(VOLUME) FORCE: -build: $(WEED) $(VEC) $(WORKER) $(VOLUME) +build: $(WEED) $(VEC) $(VOLUME) ifneq ($(VECTOR_WORKER),) build: $(CLI) $(LANCE_WORKER) endif @@ -56,4 +51,4 @@ perf: build $(CLI) ./run_perf.sh clean: - rm -f $(WEED) $(VEC) $(CLI) $(LANCE_WORKER) ./weed-vector-worker $(VOLUME) + rm -f $(WEED) $(VEC) $(CLI) $(LANCE_WORKER) $(VOLUME) diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index a5eab9f19b..0a5269116f 100644 --- a/test/s3vectors/conftest.py +++ b/test/s3vectors/conftest.py @@ -242,7 +242,6 @@ def run_cluster(param, tmp_path_factory, identities): # VECTOR_WORKER=1 adds the admin server and the Lance maintenance worker, # for the tests that build vector indexes through a plugin job. lance_worker = binary("LANCE_WORKER_BINARY", "weed-worker") if os.environ.get("VECTOR_WORKER") else None - worker = binary("VECTOR_WORKER_BINARY", "weed-vector-worker") if mode == "distributed" else None volume_binary = binary("RUST_VOLUME_SERVER_BINARY", "weed-volume") if volume == "rust" else weed processes = {} with contextlib.ExitStack() as stack: @@ -252,13 +251,13 @@ def run_cluster(param, tmp_path_factory, identities): sock = reservations.enter_context(socket.socket()) sock.bind(("127.0.0.1", 0)) ports.append(sock.getsockname()[1]) - (master, master_grpc, vol, vol_grpc, filer, filer_grpc, s3, s3_grpc, lance, vector, worker1, worker2, + (master, master_grpc, vol, vol_grpc, filer, filer_grpc, s3, s3_grpc, lance, vector, engine1, engine2, admin, admin_grpc, vector_metrics) = ports master_addr = f"127.0.0.1:{master}.{master_grpc}" filer_addr = f"127.0.0.1:{filer}.{filer_grpc}" # VECTOR_READ_JWT=1 makes volume reads need a JWT; every weed process - # and the workers sign with the key. + # and the engines sign with the key. read_jwt_key = os.environ.get("VECTOR_READ_JWT") and uuid.uuid4().hex # The cluster's filer signing key: the gateway's catalog authenticates # the worker and the CLI by the token they derive from it. @@ -346,18 +345,19 @@ def run_cluster(param, tmp_path_factory, identities): "vector engine never registered with the master", ) - if worker: - for i, port in enumerate((worker1, worker2)): - start(f"worker-{i}", [worker, "--masters", master_addr, "--listen", f"127.0.0.1:{port}"]) - wait_for(lambda: listening(port), processes, "vector worker not ready") + # Distributed mode needs the Enterprise weed-vector, whose engines all + # join the masters' placement controller. + if mode == "distributed": + for i, port in enumerate((engine1, engine2), 1): + start(f"vector-{i}", [engine, "--masters", master_addr, "--listen", f"127.0.0.1:{port}", + "--filer-group", filer_group, "--state-dir", str(directory / f"vector-{i}"), + "--capacity-bytes", str(1 << 30), "--drain-timeout", "5s"]) + wait_for(lambda: listening(port), processes, f"vector-{i} not ready") wait_for( - lambda: all( - f"S3 Vectors: discovered search worker 127.0.0.1:{port}" - in (directory / "s3.log").read_text() - for port in (worker1, worker2) - ), + lambda: all(f"joined at 127.0.0.1:{port}" in (directory / "master.log").read_text() + for port in (engine1, engine2)), processes, - "gateway did not discover both search workers", + "the engines never joined the placement controller", ) admin_url = None if lance_worker: diff --git a/weed/command/s3.go b/weed/command/s3.go index c49f2d5004..6a6607f5a4 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -734,9 +734,9 @@ func (s3opt *S3Options) startLanceServer(listenOnLocalhost bool) *http.Server { return httpS } -// 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. +// vectorsRouter builds the S3 Vectors API router — weed-vector servers +// register to the master and the backend resolves them lazily, so a server +// 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, filerGroup string) *mux.Router { if len(masterAddresses) == 0 { glog.Warningf("S3 Vectors API disabled: no weed-vector registered on master") @@ -902,10 +902,3 @@ 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. -// discoverVectorWorkers returns every registered weed-vector address — -// callers keep their current worker when it is still listed so -// membership-order changes cannot churn live connections. diff --git a/weed/s3api/s3vectors/discovery.go b/weed/s3api/s3vectors/discovery.go index 7edd3e43ef..70963ba9f7 100644 --- a/weed/s3api/s3vectors/discovery.go +++ b/weed/s3api/s3vectors/discovery.go @@ -15,7 +15,7 @@ import ( func NewVectorDiscovery(masters func() []pb.ServerAddress, dialOption grpc.DialOption, filerGroup string) *vectorDiscovery { return &vectorDiscovery{masters: masters, grpcDialOption: dialOption, discover: func(m []pb.ServerAddress, opt grpc.DialOption) []string { - return discoverVectorWorkers(m, opt, filerGroup) + return discoverVectorServers(m, opt, filerGroup) }} } @@ -25,7 +25,7 @@ type vectorDiscovery struct { discover func([]pb.ServerAddress, grpc.DialOption) []string mu sync.Mutex closed bool - worker string + server string failed string active map[vector_pb.VectorServiceClient]int retired map[vector_pb.VectorServiceClient]*grpc.ClientConn @@ -40,7 +40,7 @@ func (d *vectorDiscovery) Close() error { defer d.mu.Unlock() d.closed = true d.retire() - d.worker, d.conn, d.client = "", nil, nil + d.server, d.conn, d.client = "", nil, nil return nil } @@ -50,9 +50,9 @@ func (d *vectorDiscovery) reset(c vector_pb.VectorServiceClient) { if c != d.client { return } - d.failed = d.worker + d.failed = d.server d.retire() - d.worker, d.conn, d.client = "", nil, nil + d.server, d.conn, d.client = "", nil, nil d.nextResolve = time.Time{} } @@ -92,7 +92,7 @@ func (d *vectorDiscovery) resolve() { if d.masters != nil { masters = d.masters() } - workers := d.discover(masters, d.grpcDialOption) + servers := d.discover(masters, d.grpcDialOption) d.mu.Lock() defer d.mu.Unlock() // Measured from lookup end so a slow lookup cannot leave the next one already due. @@ -105,30 +105,30 @@ func (d *vectorDiscovery) resolve() { close(d.resolving) d.resolving = nil }() - if d.closed || len(workers) == 0 { + if d.closed || len(servers) == 0 { return } - worker := workers[0] - for _, w := range workers { + server := servers[0] + for _, w := range servers { if w != d.failed { - worker = w + server = w break } } - for _, w := range workers { - if w == d.worker && w != d.failed { - worker = w + for _, w := range servers { + if w == d.server && w != d.failed { + server = w break } } - if worker != d.worker { - conn, err := pb.GrpcDial(context.Background(), worker, false, d.grpcDialOption) + if server != d.server { + conn, err := pb.GrpcDial(context.Background(), server, false, d.grpcDialOption) if err != nil { - glog.Warningf("S3 Vectors: dial weed-vector %s: %v", worker, err) + glog.Warningf("S3 Vectors: dial weed-vector %s: %v", server, err) return } d.retire() - d.worker, d.conn = worker, conn + d.server, d.conn = server, conn d.client = vector_pb.NewVectorServiceClient(conn) } } @@ -138,7 +138,7 @@ const ( vectorRecheckInterval = 500 * time.Millisecond ) -var errNoVectorWorker = status.Error(codes.Unavailable, "no weed-vector registered on master") +var errNoVectorServer = status.Error(codes.Unavailable, "no weed-vector registered on master") func (d *vectorDiscovery) cli(ctx context.Context) (vector_pb.VectorServiceClient, error) { if err := ctx.Err(); err != nil { @@ -168,7 +168,7 @@ func (d *vectorDiscovery) cli(ctx context.Context) (vector_pb.VectorServiceClien return nil, status.Error(codes.Canceled, "vector discovery closed") } if d.client == nil { - return nil, errNoVectorWorker + return nil, errNoVectorServer } if d.active == nil { d.active = make(map[vector_pb.VectorServiceClient]int) @@ -326,7 +326,7 @@ func (d *vectorDiscovery) QueryVectors(ctx context.Context, in *vector_pb.QueryV return nil, err } -func discoverVectorWorkers(masters []pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string) []string { +func discoverVectorServers(masters []pb.ServerAddress, grpcDialOption grpc.DialOption, filerGroup string) []string { for _, master := range masters { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) conn, err := pb.GrpcDial(ctx, master.ToGrpcAddress(), false, grpcDialOption) @@ -343,15 +343,15 @@ func discoverVectorWorkers(masters []pb.ServerAddress, grpcDialOption grpc.DialO if err != nil { continue } - workers := make([]string, 0, len(resp.ClusterNodes)) + servers := make([]string, 0, len(resp.ClusterNodes)) for _, node := range resp.ClusterNodes { glog.V(0).Infof("S3 Vectors: discovered weed-vector at %s via master %s", node.Address, master) - workers = append(workers, node.Address) + servers = append(servers, node.Address) } - // A follower can answer with an empty list — workers register with + // A follower can answer with an empty list — servers register with // the leader, so keep looking rather than accept "none." - if len(workers) > 0 { - return workers + if len(servers) > 0 { + return servers } } return nil diff --git a/weed/s3api/s3vectors/discovery_test.go b/weed/s3api/s3vectors/discovery_test.go index 008a430c4e..1d66dca6ab 100644 --- a/weed/s3api/s3vectors/discovery_test.go +++ b/weed/s3api/s3vectors/discovery_test.go @@ -15,7 +15,7 @@ import ( "google.golang.org/grpc/status" ) -// A worker that registers a moment after a failed lookup must surface inside +// A server 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) { @@ -28,10 +28,10 @@ func TestVectorDiscoveryRechecksRegistration(t *testing.T) { } if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable { - t.Fatalf("no registered worker: expected Unavailable, got %v", err) + t.Fatalf("no registered server: expected Unavailable, got %v", err) } if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable { - t.Fatalf("no registered worker: expected Unavailable, got %v", err) + t.Fatalf("no registered server: 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) @@ -48,7 +48,7 @@ func TestVectorDiscoveryRechecksRegistration(t *testing.T) { d.grpcDialOption = grpc.WithTransportCredentials(insecure.NewCredentials()) if _, err := d.cli(t.Context()); err != nil { - t.Fatalf("worker registered: expected a client, got %v", err) + t.Fatalf("server 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) @@ -79,10 +79,10 @@ func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) { } if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable { - t.Fatalf("no registered worker: expected Unavailable, got %v", err) + t.Fatalf("no registered server: expected Unavailable, got %v", err) } if _, err := d.cli(t.Context()); status.Code(err) != codes.Unavailable { - t.Fatalf("no registered worker: expected Unavailable, got %v", err) + t.Fatalf("no registered server: expected Unavailable, got %v", err) } if got := atomic.LoadInt32(&calls); got != 1 { t.Fatalf("slow lookup was repeated right away: %d lookups", got) @@ -97,7 +97,7 @@ func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) { d.nextResolve = time.Time{} d.mu.Unlock() if _, err := d.cli(t.Context()); err != nil { - t.Fatalf("worker registered: expected a client, got %v", err) + t.Fatalf("server registered: expected a client, got %v", err) } waitResolved(d) } @@ -105,7 +105,7 @@ func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) { left := time.Until(d.nextResolve) d.mu.Unlock() if left < vectorResolveInterval-lookup/2 { - t.Fatalf("refresh of a live worker is due in %v, want ~%v", left, vectorResolveInterval) + t.Fatalf("refresh of a live server is due in %v, want ~%v", left, vectorResolveInterval) } } @@ -119,7 +119,7 @@ func waitResolved(d *vectorDiscovery) { } // A lookup stuck on an unreachable master must not hold up requests that -// already have a live worker, and concurrent callers start only one lookup. +// already have a live server, and concurrent callers start only one lookup. func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) { d := &vectorDiscovery{ grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), @@ -129,7 +129,7 @@ func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) { } cached, err := d.cli(t.Context()) if err != nil { - t.Fatalf("worker registered: expected a client, got %v", err) + t.Fatalf("server registered: expected a client, got %v", err) } var calls int32 @@ -179,7 +179,7 @@ func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) { } } -// Callers with no worker wait for the in-flight lookup and share its result +// Callers with no server wait for the in-flight lookup and share its result // rather than failing while it runs. func TestVectorDiscoveryUncachedCallersShareLookup(t *testing.T) { var calls int32 @@ -212,7 +212,7 @@ func TestVectorDiscoveryUncachedCallersShareLookup(t *testing.T) { } } -// A failed dial to a new worker must leave the cached connection open. +// A failed dial to a new server must leave the cached connection open. func TestVectorDiscoveryKeepsConnOnDialError(t *testing.T) { d := &vectorDiscovery{ grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), @@ -222,7 +222,7 @@ func TestVectorDiscoveryKeepsConnOnDialError(t *testing.T) { } cached, err := d.cli(t.Context()) if err != nil { - t.Fatalf("worker registered: expected a client, got %v", err) + t.Fatalf("server registered: expected a client, got %v", err) } // Without transport credentials the dial fails synchronously. @@ -287,7 +287,7 @@ func TestVectorDiscoveryWaitingRespectsCancellation(t *testing.T) { waitResolved(d) } -func TestVectorDiscoveryRetriesAnotherWorker(t *testing.T) { +func TestVectorDiscoveryRetriesAnotherServer(t *testing.T) { d := &vectorDiscovery{grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), discover: func([]pb.ServerAddress, grpc.DialOption) []string { return []string{"127.0.0.1:9301", "127.0.0.1:9302"} }} @@ -302,8 +302,8 @@ func TestVectorDiscoveryRetriesAnotherWorker(t *testing.T) { t.Fatal(err) } defer d.release(second) - if d.worker != "127.0.0.1:9302" { - t.Fatalf("retry selected failed worker: %s", d.worker) + if d.server != "127.0.0.1:9302" { + t.Fatalf("retry selected failed server: %s", d.server) } } diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 17ffa86548..e719a44883 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -1255,7 +1255,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { vectorErr(w, r, arns, err) return } - if !errors.Is(err, errNoVectorWorker) { + if !errors.Is(err, errNoVectorServer) { switch status.Code(err) { case codes.InvalidArgument, codes.PermissionDenied, codes.Unauthenticated, codes.FailedPrecondition, codes.AlreadyExists, codes.NotFound, codes.OutOfRange, codes.Unimplemented: default: