Remove expiration_interval from grpclb message (#1477)

Author: menghanl

grpclb.go

@@ -97,19 +97,18 @@ type grpclbAddrInfo struct {
 }
 
 type balancer struct {
-	r        naming.Resolver
-	target   string
-	mu       sync.Mutex
-	seq      int // a sequence number to make sure addrCh does not get stale addresses.
-	w        naming.Watcher
-	addrCh   chan []Address
-	rbs      []remoteBalancerInfo
-	addrs    []*grpclbAddrInfo
-	next     int
-	waitCh   chan struct{}
-	done     bool
-	expTimer *time.Timer
-	rand     *rand.Rand
+	r      naming.Resolver
+	target string
+	mu     sync.Mutex
+	seq    int // a sequence number to make sure addrCh does not get stale addresses.
+	w      naming.Watcher
+	addrCh chan []Address
+	rbs    []remoteBalancerInfo
+	addrs  []*grpclbAddrInfo
+	next   int
+	waitCh chan struct{}
+	done   bool
+	rand   *rand.Rand
 
 	clientStats lbmpb.ClientStats
 }
@@ -181,21 +180,6 @@ func (b *balancer) watchAddrUpdates(w naming.Watcher, ch chan []remoteBalancerIn
 	return nil
 }
 
-func (b *balancer) serverListExpire(seq int) {
-	b.mu.Lock()
-	defer b.mu.Unlock()
-	// TODO: gRPC interanls do not clear the connections when the server list is stale.
-	// This means RPCs will keep using the existing server list until b receives new
-	// server list even though the list is expired. Revisit this behavior later.
-	if b.done || seq < b.seq {
-		return
-	}
-	b.next = 0
-	b.addrs = nil
-	// Ask grpc internals to close all the corresponding connections.
-	b.addrCh <- nil
-}
-
 func convertDuration(d *lbmpb.Duration) time.Duration {
 	if d == nil {
 		return 0
@@ -208,7 +192,6 @@ func (b *balancer) processServerList(l *lbmpb.ServerList, seq int) {
 		return
 	}
 	servers := l.GetServers()
-	expiration := convertDuration(l.GetExpirationInterval())
 	var (
 		sl    []*grpclbAddrInfo
 		addrs []Address
@@ -243,15 +226,6 @@ func (b *balancer) processServerList(l *lbmpb.ServerList, seq int) {
 		b.next = 0
 		b.addrs = sl
 		b.addrCh <- addrs
-		if b.expTimer != nil {
-			b.expTimer.Stop()
-			b.expTimer = nil
-		}
-		if expiration > 0 {
-			b.expTimer = time.AfterFunc(expiration, func() {
-				b.serverListExpire(seq)
-			})
-		}
 	}
 	return
 }
@@ -721,9 +695,6 @@ func (b *balancer) Close() error {
 		return errBalancerClosed
 	}
 	b.done = true
-	if b.expTimer != nil {
-		b.expTimer.Stop()
-	}
 	if b.waitCh != nil {
 		close(b.waitCh)
 	}

grpclb/grpc_lb_v1/messages/messages.pb.go

@@ -32,7 +32,6 @@ message Duration {
 }
 
 message Timestamp {
-
   // Represents seconds of UTC time since Unix epoch
   // 1970-01-01T00:00:00Z. Must be from 0001-01-01T00:00:00Z to
   // 9999-12-31T23:59:59Z inclusive.
@@ -122,11 +121,8 @@ message ServerList {
   // unless instructed otherwise via the client_config.
   repeated Server servers = 1;
 
-  // Indicates the amount of time that the client should consider this server
-  // list as valid. It may be considered stale after waiting this interval of
-  // time after receiving the list. If the interval is not positive, the
-  // client can assume the list is valid until the next list is received.
-  Duration expiration_interval = 3;
+  // Was google.protobuf.Duration expiration_interval.
+  reserved 3;
 }
 
 // Contains server information. When none of the [drop_for_*] fields are true,

grpclb/grpc_lb_v1/messages/messages.proto

@@ -501,66 +501,6 @@ func TestDropRequestFailedNonFailFast(t *testing.T) {
 	cc.Close()
 }
 
-func TestServerExpiration(t *testing.T) {
-	tss, cleanup, err := newLoadBalancer(1)
-	if err != nil {
-		t.Fatalf("failed to create new load balancer: %v", err)
-	}
-	defer cleanup()
-	be := &lbmpb.Server{
-		IpAddress:        tss.beIPs[0],
-		Port:             int32(tss.bePorts[0]),
-		LoadBalanceToken: lbToken,
-	}
-	var bes []*lbmpb.Server
-	bes = append(bes, be)
-	exp := &lbmpb.Duration{
-		Seconds: 0,
-		Nanos:   100000000, // 100ms
-	}
-	var sls []*lbmpb.ServerList
-	sl := &lbmpb.ServerList{
-		Servers:            bes,
-		ExpirationInterval: exp,
-	}
-	sls = append(sls, sl)
-	sl = &lbmpb.ServerList{
-		Servers: bes,
-	}
-	sls = append(sls, sl)
-	var intervals []time.Duration
-	intervals = append(intervals, 0)
-	intervals = append(intervals, 500*time.Millisecond)
-	tss.ls.sls = sls
-	tss.ls.intervals = intervals
-	creds := serverNameCheckCreds{
-		expected: besn,
-	}
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	cc, err := grpc.DialContext(ctx, besn,
-		grpc.WithBalancer(grpc.NewGRPCLBBalancer(&testNameResolver{addrs: []string{tss.lbAddr}})),
-		grpc.WithBlock(), grpc.WithTransportCredentials(&creds), grpc.WithDialer(fakeNameDialer))
-	if err != nil {
-		t.Fatalf("Failed to dial to the backend %v", err)
-	}
-	testC := testpb.NewTestServiceClient(cc)
-	if _, err := testC.EmptyCall(context.Background(), &testpb.Empty{}); err != nil {
-		t.Fatalf("%v.EmptyCall(_, _) = _, %v, want _, <nil>", testC, err)
-	}
-	// Sleep and wake up when the first server list gets expired.
-	time.Sleep(150 * time.Millisecond)
-	if _, err := testC.EmptyCall(context.Background(), &testpb.Empty{}); grpc.Code(err) != codes.Unavailable {
-		t.Fatalf("%v.EmptyCall(_, _) = _, %v, want _, %s", testC, err, codes.Unavailable)
-	}
-	// A non-failfast rpc should be succeeded after the second server list is received from
-	// the remote load balancer.
-	if _, err := testC.EmptyCall(context.Background(), &testpb.Empty{}, grpc.FailFast(false)); err != nil {
-		t.Fatalf("%v.EmptyCall(_, _) = _, %v, want _, <nil>", testC, err)
-	}
-	cc.Close()
-}
-
 // When the balancer in use disconnects, grpclb should connect to the next address from resolved balancer address list.
 func TestBalancerDisconnects(t *testing.T) {
 	var (