grpcproxy: add support for WatchRequest_ProgressRequest

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

Author: alexandrevilain

server/proxy/grpcproxy/watch.go

@@ -271,6 +271,22 @@ 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:
+			wps.mu.Lock()
+			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
+			}
+			wps.mu.Unlock()
 		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))

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")