xds/clusterimpl: Convert existing unit tests to e2e style (3/N) (#8616)

Author: Pranjali-2501

internal/xds/balancer/clusterimpl/balancer_test.go

@@ -20,9 +20,7 @@ package clusterimpl
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"strings"
 	"testing"
 	"time"
 
@@ -38,7 +36,6 @@ import (
 	"google.golang.org/grpc/internal/xds/testutils/fakeclient"
 	"google.golang.org/grpc/internal/xds/xdsclient"
 	"google.golang.org/grpc/resolver"
-	"google.golang.org/grpc/serviceconfig"
 )
 
 const (
@@ -212,108 +209,6 @@ func (s) TestClusterNameInAddressAttributes(t *testing.T) {
 	}
 }
 
-// Test verifies that child policies was updated on receipt of
-// configuration update.
-func (s) TestChildPolicyUpdatedOnConfigUpdate(t *testing.T) {
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	// Keep track of which child policy was updated
-	updatedChildPolicy := ""
-
-	// Create stub balancers to track config updates
-	const (
-		childPolicyName1 = "stubBalancer1"
-		childPolicyName2 = "stubBalancer2"
-	)
-
-	stub.Register(childPolicyName1, stub.BalancerFuncs{
-		UpdateClientConnState: func(_ *stub.BalancerData, _ balancer.ClientConnState) error {
-			updatedChildPolicy = childPolicyName1
-			return nil
-		},
-	})
-
-	stub.Register(childPolicyName2, stub.BalancerFuncs{
-		UpdateClientConnState: func(_ *stub.BalancerData, _ balancer.ClientConnState) error {
-			updatedChildPolicy = childPolicyName2
-			return nil
-		},
-	})
-
-	// Initial config update with childPolicyName1
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster: testClusterName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: childPolicyName1,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("Error updating the config: %v", err)
-	}
-
-	if updatedChildPolicy != childPolicyName1 {
-		t.Fatal("Child policy 1 was not updated on initial configuration update.")
-	}
-
-	// Second config update with childPolicyName2
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster: testClusterName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: childPolicyName2,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("Error updating the config: %v", err)
-	}
-
-	if updatedChildPolicy != childPolicyName2 {
-		t.Fatal("Child policy 2 was not updated after child policy name change.")
-	}
-}
-
-// Test verifies that config update fails if child policy config
-// failed to parse.
-func (s) TestFailedToParseChildPolicyConfig(t *testing.T) {
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	// Create a stub balancer which fails to ParseConfig.
-	const parseConfigError = "failed to parse config"
-	const childPolicyName = "stubBalancer-FailedToParseChildPolicyConfig"
-	stub.Register(childPolicyName, stub.BalancerFuncs{
-		ParseConfig: func(_ json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
-			return nil, errors.New(parseConfigError)
-		},
-	})
-
-	err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster: testClusterName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: childPolicyName,
-			},
-		},
-	})
-
-	if err == nil || !strings.Contains(err.Error(), parseConfigError) {
-		t.Fatalf("Got error: %v, want error: %s", err, parseConfigError)
-	}
-}
-
 // Test verify that the case picker is updated synchronously on receipt of
 // configuration update.
 func (s) TestPickerUpdatedSynchronouslyOnConfigUpdate(t *testing.T) {

internal/xds/balancer/clusterimpl/tests/balancer_test.go

@@ -20,21 +20,27 @@ package clusterimpl_test
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
 	"net"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/balancer/pickfirst"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal"
+	"google.golang.org/grpc/internal/balancer/stub"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
@@ -43,6 +49,7 @@ import (
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
+	"google.golang.org/grpc/serviceconfig"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/testing/protocmp"
 	"google.golang.org/protobuf/types/known/durationpb"
@@ -1113,3 +1120,186 @@ func (s) TestUpdateLRSServerToNil(t *testing.T) {
 		t.Fatalf("Expected no LRS reports after disable, got %v want %v", err, context.DeadlineExceeded)
 	}
 }
+
+// Test verifies that child policy was updated on receipt of
+// configuration update.
+func (s) TestChildPolicyChangeOnConfigUpdate(t *testing.T) {
+	// Create an xDS management server.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer mgmtServer.Stop()
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+	testutils.CreateBootstrapFileForTesting(t, bc)
+
+	// Create an xDS resolver with the above bootstrap configuration.
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
+
+	// Start a server backend exposing the test service.
+	server := stubserver.StartTestService(t, nil)
+	defer server.Stop()
+
+	const serviceName = "test-child-policy"
+
+	// Configure the xDS management server with default resources. Cluster
+	// corresponding to this resource will be configured with "round_robin"
+	// as the endpoint picking policy
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: serviceName,
+		NodeID:     nodeID,
+		Host:       "localhost",
+		Port:       testutils.ParsePort(t, server.Address),
+	})
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithResolvers(resolverBuilder), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Fatalf("client.EmptyCall() failed: %v", err)
+	}
+
+	// Register stub pickfirst LB policy so that we can catch config changes.
+	pfBuilder := balancer.Get(pickfirst.Name)
+	internal.BalancerUnregister(pfBuilder.Name())
+	lbCfgCh := make(chan serviceconfig.LoadBalancingConfig, 1)
+	var updatedChildPolicy atomic.Pointer[string]
+	stub.Register(pfBuilder.Name(), stub.BalancerFuncs{
+		ParseConfig: func(lbCfg json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
+			return pfBuilder.(balancer.ConfigParser).ParseConfig(lbCfg)
+		},
+		Init: func(bd *stub.BalancerData) {
+			bd.ChildBalancer = pfBuilder.Build(bd.ClientConn, bd.BuildOptions)
+		},
+		UpdateClientConnState: func(bd *stub.BalancerData, ccs balancer.ClientConnState) error {
+			name := pfBuilder.Name()
+			updatedChildPolicy.Store(&name)
+			select {
+			case lbCfgCh <- ccs.BalancerConfig:
+			case <-ctx.Done():
+				t.Error("Timed out while waiting for BalancerConfig, context deadline exceeded")
+			}
+			return bd.ChildBalancer.UpdateClientConnState(ccs)
+		},
+		Close: func(bd *stub.BalancerData) {
+			bd.ChildBalancer.Close()
+		},
+	})
+	defer balancer.Register(pfBuilder)
+
+	// Now update the cluster to use "pick_first" as the endpoint picking policy.
+	resources.Clusters[0].LoadBalancingPolicy = &v3clusterpb.LoadBalancingPolicy{
+		Policies: []*v3clusterpb.LoadBalancingPolicy_Policy{{
+			TypedExtensionConfig: &v3corepb.TypedExtensionConfig{
+				TypedConfig: testutils.MarshalAny(t, &v3pickfirstpb.PickFirst{}),
+			},
+		}},
+	}
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case <-ctx.Done():
+		t.Fatalf("Timeout waiting for pickfirst child policy config")
+	case <-lbCfgCh:
+	}
+
+	if p := updatedChildPolicy.Load(); p == nil || *p != pfBuilder.Name() {
+		var got string
+		if p != nil {
+			got = *p
+		}
+		t.Fatalf("Unexpected child policy after config update, got %q, want %q", got, pfBuilder.Name())
+	}
+
+	// New RPC should still be routed successfully
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Errorf("EmptyCall() failed after policy update: %v", err)
+	}
+}
+
+// Test verifies that config update fails if child policy config
+// failed to parse.
+func (s) TestFailedToParseChildPolicyConfig(t *testing.T) {
+	// Create an xDS management server.
+	mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+	defer mgmtServer.Stop()
+
+	// Create bootstrap configuration pointing to the above management server.
+	nodeID := uuid.New().String()
+	bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address)
+	testutils.CreateBootstrapFileForTesting(t, bc)
+
+	// Create an xDS resolver with the above bootstrap configuration.
+	if internal.NewXDSResolverWithConfigForTesting == nil {
+		t.Fatalf("internal.NewXDSResolverWithConfigForTesting is nil")
+	}
+	resolverBuilder, err := internal.NewXDSResolverWithConfigForTesting.(func([]byte) (resolver.Builder, error))(bc)
+	if err != nil {
+		t.Fatalf("Failed to create xDS resolver for testing: %v", err)
+	}
+
+	// Start a server backend exposing the test service.
+	server := stubserver.StartTestService(t, nil)
+	defer server.Stop()
+
+	const serviceName = "test-child-policy"
+	resources := e2e.DefaultClientResources(e2e.ResourceParams{
+		DialTarget: serviceName,
+		NodeID:     nodeID,
+		Host:       "localhost",
+		Port:       testutils.ParsePort(t, server.Address),
+	})
+	resources.Clusters[0].LoadBalancingPolicy = &v3clusterpb.LoadBalancingPolicy{
+		Policies: []*v3clusterpb.LoadBalancingPolicy_Policy{{
+			TypedExtensionConfig: &v3corepb.TypedExtensionConfig{
+				TypedConfig: testutils.MarshalAny(t, &v3pickfirstpb.PickFirst{}),
+			},
+		}},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatalf("Failed to update xDS resources: %v", err)
+	}
+
+	// Register stub pickfirst LB policy so that we can catch parsing errors.
+	pfBuilder := balancer.Get(pickfirst.Name)
+	internal.BalancerUnregister(pfBuilder.Name())
+	const parseConfigError = "failed to parse config"
+	stub.Register(pfBuilder.Name(), stub.BalancerFuncs{
+		ParseConfig: func(_ json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
+			return nil, errors.New(parseConfigError)
+		},
+	})
+	defer balancer.Register(pfBuilder)
+
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithResolvers(resolverBuilder), grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err == nil || !strings.Contains(err.Error(), parseConfigError) {
+		t.Fatal("EmptyCall RPC succeeded when expected to fail")
+	}
+}