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,23 @@ 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 {
+		// request only the next progress notification on the watcher.
+		w.requestNextProgress()
+	}
+
+	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
+}

server/proxy/grpcproxy/watcher.go

@@ -15,6 +15,7 @@
 package grpcproxy
 
 import (
+	"sync/atomic"
 	"time"
 
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
@@ -34,10 +35,12 @@ func (wr *watchRange) valid() bool {
 type watcher struct {
 	// user configuration
 
-	wr       watchRange
-	filters  []mvcc.FilterFunc
-	progress bool
-	prevKV   bool
+	wr           watchRange
+	filters      []mvcc.FilterFunc
+	progress     bool
+	nextProgress atomic.Bool
+
+	prevKV bool
 
 	// id is the id returned to the client on its watch stream.
 	id int64
@@ -50,10 +53,20 @@ type watcher struct {
 	wps *watchProxyStream
 }
 
+// requestNextProgress sets the nextProgress to true to ensure next progress notification
+// is forwarded to the client.
+func (w *watcher) requestNextProgress() {
+	if w.progress {
+		return
+	}
+
+	w.nextProgress.Store(true)
+}
+
 // send filters out repeated events by discarding revisions older
 // than the last one sent over the watch channel.
 func (w *watcher) send(wr clientv3.WatchResponse) {
-	if wr.IsProgressNotify() && !w.progress {
+	if wr.IsProgressNotify() && !w.progress && !w.nextProgress.CompareAndSwap(true, false) {
 		return
 	}
 	if w.nextrev > wr.Header.Revision && len(wr.Events) > 0 {

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