xds: include nonce from the most recent response in new requests (#3420)

Even if the new request is for a new watch.

None and version are reset on new ADS streams.

Author: menghanl

xds/internal/client/v2client.go

@@ -67,11 +67,13 @@ type v2Client struct {
 	// these are set to nil. All accesses to the map protected and any value
 	// inside the map should be protected with the above mutex.
 	watchMap map[string]*watchInfo
-	// ackMap contains the version that was acked (the version in the ack
+	// versionMap contains the version that was acked (the version in the ack
 	// request that was sent on wire). The key is typeURL, the value is the
-	// version string, becaues the versions for different resource types
-	// should be independent.
-	ackMap map[string]string
+	// version string, becaues the versions for different resource types should
+	// be independent.
+	versionMap map[string]string
+	// nonceMap contains the nonce from the most recent received response.
+	nonceMap map[string]string
 	// rdsCache maintains a mapping of {routeConfigName --> clusterName} from
 	// validated route configurations received in RDS responses. We cache all
 	// valid route configurations, whether or not we are interested in them
@@ -107,10 +109,11 @@ func newV2Client(cc *grpc.ClientConn, nodeProto *corepb.Node, backoff func(int)
 		streamCh: make(chan adsStream, 1),
 		sendCh:   buffer.NewUnbounded(),
 
-		watchMap: make(map[string]*watchInfo),
-		ackMap:   make(map[string]string),
-		rdsCache: make(map[string]string),
-		cdsCache: make(map[string]CDSUpdate),
+		watchMap:   make(map[string]*watchInfo),
+		versionMap: make(map[string]string),
+		nonceMap:   make(map[string]string),
+		rdsCache:   make(map[string]string),
+		cdsCache:   make(map[string]CDSUpdate),
 	}
 	v2c.ctx, v2c.cancelCtx = context.WithCancel(context.Background())
 
@@ -179,7 +182,7 @@ func (v2c *v2Client) run() {
 // string
 // - If this is an ack, version will be the version from the response
 // - If this is a nack, version will be the previous acked version (from
-// ackMap). If there was no ack before, it will be an empty string
+// versionMap). If there was no ack before, it will be an empty string
 func (v2c *v2Client) sendRequest(stream adsStream, resourceNames []string, typeURL, version, nonce string) bool {
 	req := &xdspb.DiscoveryRequest{
 		Node:          v2c.nodeProto,
@@ -208,7 +211,8 @@ func (v2c *v2Client) sendExisting(stream adsStream) bool {
 	defer v2c.mu.Unlock()
 
 	// Reset the ack versions when the stream restarts.
-	v2c.ackMap = make(map[string]string)
+	v2c.versionMap = make(map[string]string)
+	v2c.nonceMap = make(map[string]string)
 
 	for typeURL, wi := range v2c.watchMap {
 		if !v2c.sendRequest(stream, wi.target, typeURL, "", "") {
@@ -239,10 +243,12 @@ func (v2c *v2Client) processWatchInfo(t *watchInfo) (target []string, typeURL, v
 	v2c.checkCacheAndUpdateWatchMap(t)
 	// TODO: if watch is called again with the same resource names,
 	// there's no need to send another request.
-	//
-	// TODO: should we reset version (for ack) when a new watch is
-	// started? Or do this only if the resource names are different
-	// (so we send a new request)?
+
+	// We don't reset version or nonce when a new watch is started. The version
+	// and nonce from previous response are carried by the request unless the
+	// stream is recreated.
+	version = v2c.versionMap[typeURL]
+	nonce = v2c.nonceMap[typeURL]
 	return
 }
 
@@ -270,13 +276,14 @@ func (v2c *v2Client) processAckInfo(t *ackInfo) (target []string, typeURL, versi
 	target = wi.target
 	if version == "" {
 		// This is a nack, get the previous acked version.
-		version = v2c.ackMap[typeURL]
+		version = v2c.versionMap[typeURL]
 		// version will still be an empty string if typeURL isn't
-		// found in ackMap, this can happen if there wasn't any ack
+		// found in versionMap, this can happen if there wasn't any ack
 		// before.
 	} else {
-		v2c.ackMap[typeURL] = version
+		v2c.versionMap[typeURL] = version
 	}
+	v2c.nonceMap[typeURL] = nonce
 	return
 }
 

xds/internal/client/v2client_ack_test.go

@@ -61,7 +61,7 @@ func sendXDSRespWithVersion(ch chan<- *fakeserver.Response, respWithoutVersion *
 
 // startXDS calls watch to send the first request. It then sends a good response
 // and checks for ack.
-func startXDS(t *testing.T, xdsname string, v2c *v2Client, reqChan *testutils.Channel, req *xdspb.DiscoveryRequest) *testutils.Channel {
+func startXDS(t *testing.T, xdsname string, v2c *v2Client, reqChan *testutils.Channel, req *xdspb.DiscoveryRequest, preVersion string, preNonce string) *testutils.Channel {
 	callbackCh := testutils.NewChannel()
 	switch xdsname {
 	case "LDS":
@@ -86,7 +86,7 @@ func startXDS(t *testing.T, xdsname string, v2c *v2Client, reqChan *testutils.Ch
 		})
 	}
 
-	if err := compareXDSRequest(reqChan, req, "", ""); err != nil {
+	if err := compareXDSRequest(reqChan, req, preVersion, preNonce); err != nil {
 		t.Fatalf("Failed to receive %s request: %v", xdsname, err)
 	}
 	t.Logf("FakeServer received %s request...", xdsname)
@@ -98,8 +98,8 @@ func startXDS(t *testing.T, xdsname string, v2c *v2Client, reqChan *testutils.Ch
 //
 // It also waits and checks that the ack request contains the given version, and
 // the generated nonce.
-func sendGoodResp(t *testing.T, xdsname string, fakeServer *fakeserver.Server, version int, goodResp *xdspb.DiscoveryResponse, wantReq *xdspb.DiscoveryRequest, callbackCh *testutils.Channel) {
-	nonce := sendXDSRespWithVersion(fakeServer.XDSResponseChan, goodResp, version)
+func sendGoodResp(t *testing.T, xdsname string, fakeServer *fakeserver.Server, version int, goodResp *xdspb.DiscoveryResponse, wantReq *xdspb.DiscoveryRequest, callbackCh *testutils.Channel) (nonce string) {
+	nonce = sendXDSRespWithVersion(fakeServer.XDSResponseChan, goodResp, version)
 	t.Logf("Good %s response pushed to fakeServer...", xdsname)
 
 	if err := compareXDSRequest(fakeServer.XDSRequestChan, wantReq, strconv.Itoa(version), nonce); err != nil {
@@ -111,6 +111,7 @@ func sendGoodResp(t *testing.T, xdsname string, fakeServer *fakeserver.Server, v
 		t.Errorf("Timeout when expecting %s update", xdsname)
 	}
 	t.Logf("Good %s response callback executed", xdsname)
+	return
 }
 
 // sendBadResp sends a bad response with the given version. This response will
@@ -159,16 +160,16 @@ func (s) TestV2ClientAck(t *testing.T) {
 	t.Log("Started xds v2Client...")
 
 	// Start the watch, send a good response, and check for ack.
-	cbLDS := startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest)
+	cbLDS := startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest, "", "")
 	sendGoodResp(t, "LDS", fakeServer, versionLDS, goodLDSResponse1, goodLDSRequest, cbLDS)
 	versionLDS++
-	cbRDS := startXDS(t, "RDS", v2c, fakeServer.XDSRequestChan, goodRDSRequest)
+	cbRDS := startXDS(t, "RDS", v2c, fakeServer.XDSRequestChan, goodRDSRequest, "", "")
 	sendGoodResp(t, "RDS", fakeServer, versionRDS, goodRDSResponse1, goodRDSRequest, cbRDS)
 	versionRDS++
-	cbCDS := startXDS(t, "CDS", v2c, fakeServer.XDSRequestChan, goodCDSRequest)
+	cbCDS := startXDS(t, "CDS", v2c, fakeServer.XDSRequestChan, goodCDSRequest, "", "")
 	sendGoodResp(t, "CDS", fakeServer, versionCDS, goodCDSResponse1, goodCDSRequest, cbCDS)
 	versionCDS++
-	cbEDS := startXDS(t, "EDS", v2c, fakeServer.XDSRequestChan, goodEDSRequest)
+	cbEDS := startXDS(t, "EDS", v2c, fakeServer.XDSRequestChan, goodEDSRequest, "", "")
 	sendGoodResp(t, "EDS", fakeServer, versionEDS, goodEDSResponse1, goodEDSRequest, cbEDS)
 	versionEDS++
 
@@ -205,7 +206,7 @@ func (s) TestV2ClientAckFirstIsNack(t *testing.T) {
 	t.Log("Started xds v2Client...")
 
 	// Start the watch, send a good response, and check for ack.
-	cbLDS := startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest)
+	cbLDS := startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest, "", "")
 
 	nonce := sendXDSRespWithVersion(fakeServer.XDSResponseChan, &xdspb.DiscoveryResponse{
 		Resources: []*anypb.Any{{}},
@@ -237,15 +238,16 @@ func (s) TestV2ClientAckNackAfterNewWatch(t *testing.T) {
 	t.Log("Started xds v2Client...")
 
 	// Start the watch, send a good response, and check for ack.
-	cbLDS := startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest)
-	sendGoodResp(t, "LDS", fakeServer, versionLDS, goodLDSResponse1, goodLDSRequest, cbLDS)
-	versionLDS++
+	cbLDS := startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest, "", "")
+	nonce := sendGoodResp(t, "LDS", fakeServer, versionLDS, goodLDSResponse1, goodLDSRequest, cbLDS)
 
-	// Start a new watch.
-	cbLDS = startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest)
+	// Start a new watch. The version in the new request should be the version
+	// from the previous response, thus versionLDS before ++.
+	cbLDS = startXDS(t, "LDS", v2c, fakeServer.XDSRequestChan, goodLDSRequest, strconv.Itoa(versionLDS), nonce)
+	versionLDS++
 
 	// This is an invalid response after the new watch.
-	nonce := sendXDSRespWithVersion(fakeServer.XDSResponseChan, &xdspb.DiscoveryResponse{
+	nonce = sendXDSRespWithVersion(fakeServer.XDSResponseChan, &xdspb.DiscoveryResponse{
 		Resources: []*anypb.Any{{}},
 		TypeUrl:   ldsURL,
 	}, versionLDS)