grpcproxy: add support for WatchRequest_ProgressRequest

Signed-off-by: alexandre.vilain <[email protected]>

Author: alexandrevilain

server/proxy/grpcproxy/watch.go

@@ -271,6 +271,11 @@ func (wps *watchProxyStream) recvLoop() error {
 		case *pb.WatchRequest_CancelRequest:
 			wps.delete(uv.CancelRequest.WatchId)
 			wps.lg.Debug("cancel watcher", zap.Int64("watcherId", uv.CancelRequest.WatchId))
+		case *pb.WatchRequest_ProgressRequest:
+			err := wps.requestProgressAll()
+			if err != nil {
+				return err
+			}
 		default:
 			// Panic or Fatalf would allow to network clients to crash the serve remotely.
 			wps.lg.Error("not supported request type by gRPC proxy", zap.Stringer("request", req))
@@ -311,3 +316,25 @@ func (wps *watchProxyStream) delete(id int64) {
 	}
 	wps.watchCh <- resp
 }
+
+func (wps *watchProxyStream) requestProgressAll() error {
+	wps.mu.Lock()
+	defer wps.mu.Unlock()
+
+	for _, w := range wps.watchers {
+		// Enable the progress notify on the watcher, otherwise the send()
+		// method may drop progress events.
+		w.progress = true
+		wps.lg.Debug("enabled process notify on watcher", zap.Int64("watcherId", w.id))
+	}
+
+	if err := wps.ranges.wp.cw.RequestProgress(wps.stream.Context()); err != nil {
+		wps.lg.Error("failed to request progress", zap.Error(err))
+		// At this point, the error returned by the client isn't
+		// due to something wrong on the cluster but to closed conns.
+		// So we don't have to send a watch response.
+		return err
+	}
+
+	return nil
+}

tests/integration/v3_watch_test.go

@@ -1424,12 +1424,6 @@ func TestV3WatchCloseCancelRace(t *testing.T) {
 // TestV3WatchProgressWaitsForSync checks that progress notifications
 // don't get sent until the watcher is synchronised
 func TestV3WatchProgressWaitsForSync(t *testing.T) {
-	// Disable for gRPC proxy, as it does not support requesting
-	// progress notifications
-	if integration.ThroughProxy {
-		t.Skip("grpc proxy currently does not support requesting progress notifications")
-	}
-
 	integration.BeforeTest(t)
 
 	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
@@ -1490,9 +1484,6 @@ func TestV3WatchProgressWaitsForSync(t *testing.T) {
 }
 
 func TestV3WatchProgressWaitsForSyncNoEvents(t *testing.T) {
-	if integration.ThroughProxy {
-		t.Skip("grpc proxy currently does not support requesting progress notifications")
-	}
 	integration.BeforeTest(t)
 
 	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
@@ -1536,9 +1527,6 @@ func TestV3WatchProgressWaitsForSyncNoEvents(t *testing.T) {
 // TestV3NoEventsLostOnCompact verifies that slow watchers exit with compacted watch response
 // if its next revision of events are compacted and no lost events sent to client.
 func TestV3NoEventsLostOnCompact(t *testing.T) {
-	if integration.ThroughProxy {
-		t.Skip("grpc proxy currently does not support requesting progress notifications")
-	}
 	integration.BeforeTest(t)
 	if len(gofail.List()) == 0 {
 		t.Skip("please run 'make gofail-enable' before running the test")