Method TestRingHash_EndpointHashKey

balancer/ringhash/ringhash_e2e_test.go:2560–2663

Tests that RPCs are routed according to endpoint hash key rather than endpoint first address if it is set in EDS endpoint metadata.

(t *testing.T)

// Tests that RPCs are routed according to endpoint hash key rather than
// endpoint first address if it is set in EDS endpoint metadata.
func (s) TestRingHash_EndpointHashKey(t *testing.T) {
	testutils.SetEnvConfig(t, &envconfig.XDSEndpointHashKeyBackwardCompat, false)

	backends := backendAddrs(startTestServiceBackends(t, 4))

	const clusterName = "cluster"
	var backendOpts []e2e.BackendOptions
	for i, addr := range backends {
		var ports []uint32
		ports = append(ports, testutils.ParsePort(t, addr))
		backendOpts = append(backendOpts, e2e.BackendOptions{
			Ports:    ports,
			Metadata: map[string]any{"hash_key": strconv.Itoa(i)},
		})
	}
	endpoints := e2e.EndpointResourceWithOptions(e2e.EndpointOptions{
		ClusterName: clusterName,
		Host:        "localhost",
		Localities: []e2e.LocalityOptions{{
			Backends: backendOpts,
			Weight:   1,
		}},
	})
	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
		ClusterName: clusterName,
		ServiceName: clusterName,
		Policy:      e2e.LoadBalancingPolicyRingHash,
	})
	route := headerHashRoute("new_route", virtualHostName, clusterName, "address_hash")
	listener := e2e.DefaultClientListener(virtualHostName, route.Name)

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	xdsServer, nodeID, xdsResolver := setupManagementServerAndResolver(t)
	if err := xdsServer.Update(ctx, xdsUpdateOpts(nodeID, endpoints, cluster, route, listener)); err != nil {
		t.Fatalf("Failed to update xDS resources: %v", err)
	}

	opts := []grpc.DialOption{
		grpc.WithResolvers(xdsResolver),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	}
	conn, err := grpc.NewClient("xds:///test.server", opts...)
	if err != nil {
		t.Fatalf("Failed to create client: %s", err)
	}
	defer conn.Close()
	client := testgrpc.NewTestServiceClient(conn)

	// Make sure RPCs are routed to backends according to the endpoint metadata
	// rather than their address. Note each type of RPC contains a header value
	// that will always be hashed to a specific backend as the header value
	// matches the endpoint metadata hash key.
	for i, backend := range backends {
		ctx := metadata.NewOutgoingContext(ctx, metadata.Pairs("address_hash", strconv.Itoa(i)+"_0"))
		numRPCs := 10
		reqPerBackend := checkRPCSendOK(ctx, t, client, numRPCs)