roundrobin_test.go

Documentation: google.golang.org/grpc/test

     1  /*
     2   *
     3   * Copyright 2022 gRPC authors.
     4   *
     5   * Licensed under the Apache License, Version 2.0 (the "License");
     6   * you may not use this file except in compliance with the License.
     7   * You may obtain a copy of the License at
     8   *
     9   *     http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   */
    18  
    19  package test
    20  
    21  import (
    22  	"context"
    23  	"strings"
    24  	"testing"
    25  	"time"
    26  
    27  	"google.golang.org/grpc"
    28  	"google.golang.org/grpc/codes"
    29  	"google.golang.org/grpc/connectivity"
    30  	"google.golang.org/grpc/credentials/insecure"
    31  	"google.golang.org/grpc/internal/channelz"
    32  	imetadata "google.golang.org/grpc/internal/metadata"
    33  	"google.golang.org/grpc/internal/stubserver"
    34  	"google.golang.org/grpc/internal/testutils"
    35  	rrutil "google.golang.org/grpc/internal/testutils/roundrobin"
    36  	"google.golang.org/grpc/metadata"
    37  	"google.golang.org/grpc/resolver"
    38  	"google.golang.org/grpc/resolver/manual"
    39  	"google.golang.org/grpc/status"
    40  
    41  	testgrpc "google.golang.org/grpc/interop/grpc_testing"
    42  	testpb "google.golang.org/grpc/interop/grpc_testing"
    43  )
    44  
    45  const rrServiceConfig = `{"loadBalancingConfig": [{"round_robin":{}}]}`
    46  
    47  func testRoundRobinBasic(ctx context.Context, t *testing.T, opts ...grpc.DialOption) (*grpc.ClientConn, *manual.Resolver, []*stubserver.StubServer) {
    48  	t.Helper()
    49  
    50  	r := manual.NewBuilderWithScheme("whatever")
    51  
    52  	const backendCount = 5
    53  	backends := make([]*stubserver.StubServer, backendCount)
    54  	addrs := make([]resolver.Address, backendCount)
    55  	for i := 0; i < backendCount; i++ {
    56  		backend := &stubserver.StubServer{
    57  			EmptyCallF: func(ctx context.Context, in *testpb.Empty) (*testpb.Empty, error) { return &testpb.Empty{}, nil },
    58  		}
    59  		if err := backend.StartServer(); err != nil {
    60  			t.Fatalf("Failed to start backend: %v", err)
    61  		}
    62  		t.Logf("Started TestService backend at: %q", backend.Address)
    63  		t.Cleanup(func() { backend.Stop() })
    64  
    65  		backends[i] = backend
    66  		addrs[i] = resolver.Address{Addr: backend.Address}
    67  	}
    68  
    69  	dopts := []grpc.DialOption{
    70  		grpc.WithTransportCredentials(insecure.NewCredentials()),
    71  		grpc.WithResolvers(r),
    72  		grpc.WithDefaultServiceConfig(rrServiceConfig),
    73  	}
    74  	dopts = append(dopts, opts...)
    75  	cc, err := grpc.NewClient(r.Scheme()+":///test.server", dopts...)
    76  	if err != nil {
    77  		t.Fatalf("grpc.NewClient() failed: %v", err)
    78  	}
    79  	t.Cleanup(func() { cc.Close() })
    80  	client := testgrpc.NewTestServiceClient(cc)
    81  
    82  	// At this point, the resolver has not returned any addresses to the channel.
    83  	// This RPC must block until the context expires.
    84  	sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
    85  	defer sCancel()
    86  	if _, err := client.EmptyCall(sCtx, &testpb.Empty{}); status.Code(err) != codes.DeadlineExceeded {
    87  		t.Fatalf("EmptyCall() = %s, want %s", status.Code(err), codes.DeadlineExceeded)
    88  	}
    89  
    90  	r.UpdateState(resolver.State{Addresses: addrs})
    91  	if err := rrutil.CheckRoundRobinRPCs(ctx, client, addrs); err != nil {
    92  		t.Fatal(err)
    93  	}
    94  	return cc, r, backends
    95  }
    96  
    97  // TestRoundRobin_Basic tests the most basic scenario for round_robin. It brings
    98  // up a bunch of backends and verifies that RPCs are getting round robin-ed
    99  // across these backends.
   100  func (s) TestRoundRobin_Basic(t *testing.T) {
   101  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   102  	defer cancel()
   103  	testRoundRobinBasic(ctx, t)
   104  }
   105  
   106  // TestRoundRobin_AddressesRemoved tests the scenario where a bunch of backends
   107  // are brought up, and round_robin is configured as the LB policy and RPCs are
   108  // being correctly round robin-ed across these backends. We then send a resolver
   109  // update with no addresses and verify that the channel enters TransientFailure
   110  // and RPCs fail with an expected error message.
   111  func (s) TestRoundRobin_AddressesRemoved(t *testing.T) {
   112  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   113  	defer cancel()
   114  	cc, r, _ := testRoundRobinBasic(ctx, t)
   115  
   116  	// Send a resolver update with no addresses. This should push the channel into
   117  	// TransientFailure.
   118  	r.UpdateState(resolver.State{Addresses: []resolver.Address{}})
   119  	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
   120  
   121  	const msgWant = "produced zero addresses"
   122  	client := testgrpc.NewTestServiceClient(cc)
   123  	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); !strings.Contains(status.Convert(err).Message(), msgWant) {
   124  		t.Fatalf("EmptyCall() = %v, want Contains(Message(), %q)", err, msgWant)
   125  	}
   126  }
   127  
   128  // TestRoundRobin_NewAddressWhileBlocking tests the case where round_robin is
   129  // configured on a channel, things are working as expected and then a resolver
   130  // updates removes all addresses. An RPC attempted at this point in time will be
   131  // blocked because there are no valid ¡ds. This test verifies that when new
   132  // backends are added, the RPC is able to complete.
   133  func (s) TestRoundRobin_NewAddressWhileBlocking(t *testing.T) {
   134  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   135  	defer cancel()
   136  	cc, r, backends := testRoundRobinBasic(ctx, t)
   137  
   138  	// Send a resolver update with no addresses. This should push the channel into
   139  	// TransientFailure.
   140  	r.UpdateState(resolver.State{Addresses: []resolver.Address{}})
   141  	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
   142  
   143  	client := testgrpc.NewTestServiceClient(cc)
   144  	doneCh := make(chan struct{})
   145  	go func() {
   146  		// The channel is currently in TransientFailure and this RPC will block
   147  		// until the channel becomes Ready, which will only happen when we push a
   148  		// resolver update with a valid backend address.
   149  		if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
   150  			t.Errorf("EmptyCall() = %v, want <nil>", err)
   151  		}
   152  		close(doneCh)
   153  	}()
   154  
   155  	// Make sure that there is one pending RPC on the ClientConn before attempting
   156  	// to push new addresses through the name resolver. If we don't do this, the
   157  	// resolver update can happen before the above goroutine gets to make the RPC.
   158  	for {
   159  		if err := ctx.Err(); err != nil {
   160  			t.Fatal(err)
   161  		}
   162  		tcs, _ := channelz.GetTopChannels(0, 0)
   163  		if len(tcs) != 1 {
   164  			t.Fatalf("there should only be one top channel, not %d", len(tcs))
   165  		}
   166  		started := tcs[0].ChannelMetrics.CallsStarted.Load()
   167  		completed := tcs[0].ChannelMetrics.CallsSucceeded.Load() + tcs[0].ChannelMetrics.CallsFailed.Load()
   168  		if (started - completed) == 1 {
   169  			break
   170  		}
   171  		time.Sleep(defaultTestShortTimeout)
   172  	}
   173  
   174  	// Send a resolver update with a valid backend to push the channel to Ready
   175  	// and unblock the above RPC.
   176  	r.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: backends[0].Address}}})
   177  
   178  	select {
   179  	case <-ctx.Done():
   180  		t.Fatal("Timeout when waiting for blocked RPC to complete")
   181  	case <-doneCh:
   182  	}
   183  }
   184  
   185  // TestRoundRobin_OneServerDown tests the scenario where a channel is configured
   186  // to round robin across a set of backends, and things are working correctly.
   187  // One backend goes down. The test verifies that going forward, RPCs are round
   188  // robin-ed across the remaining set of backends.
   189  func (s) TestRoundRobin_OneServerDown(t *testing.T) {
   190  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   191  	defer cancel()
   192  	cc, _, backends := testRoundRobinBasic(ctx, t)
   193  
   194  	// Stop one backend. RPCs should round robin across the remaining backends.
   195  	backends[len(backends)-1].Stop()
   196  
   197  	addrs := make([]resolver.Address, len(backends)-1)
   198  	for i := 0; i < len(backends)-1; i++ {
   199  		addrs[i] = resolver.Address{Addr: backends[i].Address}
   200  	}
   201  	client := testgrpc.NewTestServiceClient(cc)
   202  	if err := rrutil.CheckRoundRobinRPCs(ctx, client, addrs); err != nil {
   203  		t.Fatalf("RPCs are not being round robined across remaining servers: %v", err)
   204  	}
   205  }
   206  
   207  // TestRoundRobin_AllServersDown tests the scenario where a channel is
   208  // configured to round robin across a set of backends, and things are working
   209  // correctly. Then, all backends go down. The test verifies that the channel
   210  // moves to TransientFailure and failfast RPCs fail with Unavailable.
   211  func (s) TestRoundRobin_AllServersDown(t *testing.T) {
   212  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   213  	defer cancel()
   214  	cc, _, backends := testRoundRobinBasic(ctx, t)
   215  
   216  	// Stop all backends.
   217  	for _, b := range backends {
   218  		b.Stop()
   219  	}
   220  
   221  	testutils.AwaitState(ctx, t, cc, connectivity.TransientFailure)
   222  
   223  	// Failfast RPCs should fail with Unavailable.
   224  	client := testgrpc.NewTestServiceClient(cc)
   225  
   226  	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); status.Code(err) != codes.Unavailable {
   227  		t.Fatalf("EmptyCall got err: %v; want Unavailable", err)
   228  	}
   229  }
   230  
   231  // TestRoundRobin_UpdateAddressAttributes tests the scenario where the addresses
   232  // returned by the resolver contain attributes. The test verifies that the
   233  // attributes contained in the addresses show up as RPC metadata in the backend.
   234  func (s) TestRoundRobin_UpdateAddressAttributes(t *testing.T) {
   235  	const (
   236  		testMDKey   = "test-md"
   237  		testMDValue = "test-md-value"
   238  	)
   239  	r := manual.NewBuilderWithScheme("whatever")
   240  
   241  	// Spin up a StubServer to serve as a backend. The implementation verifies
   242  	// that the expected metadata is received.
   243  	testMDChan := make(chan []string, 1)
   244  	backend := &stubserver.StubServer{
   245  		EmptyCallF: func(ctx context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
   246  			md, ok := metadata.FromIncomingContext(ctx)
   247  			if ok {
   248  				select {
   249  				case testMDChan <- md[testMDKey]:
   250  				case <-ctx.Done():
   251  					return nil, ctx.Err()
   252  				}
   253  			}
   254  			return &testpb.Empty{}, nil
   255  		},
   256  	}
   257  	if err := backend.StartServer(); err != nil {
   258  		t.Fatalf("Failed to start backend: %v", err)
   259  	}
   260  	t.Logf("Started TestService backend at: %q", backend.Address)
   261  	t.Cleanup(func() { backend.Stop() })
   262  
   263  	// Dial the backend with round_robin as the LB policy.
   264  	dopts := []grpc.DialOption{
   265  		grpc.WithTransportCredentials(insecure.NewCredentials()),
   266  		grpc.WithResolvers(r),
   267  		grpc.WithDefaultServiceConfig(rrServiceConfig),
   268  	}
   269  	cc, err := grpc.Dial(r.Scheme()+":///test.server", dopts...)
   270  	if err != nil {
   271  		t.Fatalf("grpc.Dial() failed: %v", err)
   272  	}
   273  	t.Cleanup(func() { cc.Close() })
   274  
   275  	// Send a resolver update with no address attributes.
   276  	addr := resolver.Address{Addr: backend.Address}
   277  	r.UpdateState(resolver.State{Addresses: []resolver.Address{addr}})
   278  
   279  	// Make an RPC and ensure it does not contain the metadata we are looking for.
   280  	client := testgrpc.NewTestServiceClient(cc)
   281  	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
   282  	defer cancel()
   283  	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
   284  		t.Fatalf("EmptyCall() = %v, want <nil>", err)
   285  	}
   286  	select {
   287  	case <-ctx.Done():
   288  		t.Fatalf("Timeout when waiting for metadata received in RPC")
   289  	case md := <-testMDChan:
   290  		if len(md) != 0 {
   291  			t.Fatalf("received metadata %v, want nil", md)
   292  		}
   293  	}
   294  
   295  	// Send a resolver update with address attributes.
   296  	addrWithAttributes := imetadata.Set(addr, metadata.Pairs(testMDKey, testMDValue))
   297  	r.UpdateState(resolver.State{Addresses: []resolver.Address{addrWithAttributes}})
   298  
   299  	// Make an RPC and ensure it contains the metadata we are looking for. The
   300  	// resolver update isn't processed synchronously, so we wait some time before
   301  	// failing if some RPCs do not contain it.
   302  Done:
   303  	for {
   304  		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
   305  			t.Fatalf("EmptyCall() = %v, want <nil>", err)
   306  		}
   307  		select {
   308  		case <-ctx.Done():
   309  			t.Fatalf("Timeout when waiting for metadata received in RPC")
   310  		case md := <-testMDChan:
   311  			if len(md) == 1 && md[0] == testMDValue {
   312  				break Done
   313  			}
   314  		}
   315  		time.Sleep(defaultTestShortTimeout)
   316  	}
   317  }
   318
View as plain text