Add support for WatchProgressRequest

go.mod

@@ -25,6 +25,7 @@ require (
 	go.etcd.io/etcd/client/v3 v3.5.9
 	go.etcd.io/etcd/server/v3 v3.5.9
 	google.golang.org/grpc v1.56.3
+	k8s.io/apimachinery v0.25.4
 	k8s.io/client-go v0.25.4
 )
 
@@ -90,7 +91,6 @@ require (
 	google.golang.org/protobuf v1.30.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	k8s.io/apimachinery v0.25.4 // indirect
 	k8s.io/klog/v2 v2.70.1 // indirect
 	k8s.io/utils v0.0.0-20220728103510-ee6ede2d64ed // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect

main.go

@@ -96,6 +96,12 @@ func main() {
 			Usage:       "Enable net/http/pprof handlers on the metrics bind address. Default is false.",
 			Destination: &metricsConfig.EnableProfiling,
 		},
+		cli.DurationFlag{
+			Name:        "watch-progress-notify-interval",
+			Usage:       "Interval between periodic watch progress notifications. Default is 10m.",
+			Destination: &config.NotifyInterval,
+			Value:       time.Minute * 10,
+		},
 		cli.BoolFlag{Name: "debug"},
 	}
 	app.Action = run

pkg/endpoint/endpoint.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/k3s-io/kine/pkg/drivers/dqlite"
 	"github.com/k3s-io/kine/pkg/drivers/generic"
@@ -44,6 +45,7 @@ type Config struct {
 	ServerTLSConfig      tls.Config
 	BackendTLSConfig     tls.Config
 	MetricsRegisterer    prometheus.Registerer
+	NotifyInterval       time.Duration
 }
 
 type ETCDConfig struct {
@@ -80,7 +82,7 @@ func Listen(ctx context.Context, config Config) (ETCDConfig, error) {
 	}
 
 	// set up GRPC server and register services
-	b := server.New(backend, endpointScheme(config))
+	b := server.New(backend, endpointScheme(config), config.NotifyInterval)
 	grpcServer, err := grpcServer(config)
 	if err != nil {
 		return ETCDConfig{}, errors.Wrap(err, "creating GRPC server")

pkg/logstructured/sqllog/sql.go

@@ -26,6 +26,7 @@ type SQLLog struct {
 	broadcaster broadcaster.Broadcaster
 	ctx         context.Context
 	notify      chan int64
+	currentRev  int64
 }
 
 func New(d server.Dialect) *SQLLog {
@@ -109,7 +110,7 @@ func (s *SQLLog) compactor(interval time.Duration) {
 	t := time.NewTicker(interval)
 	defer t.Stop()
 	compactRev, _ := s.d.GetCompactRevision(s.ctx)
-	targetCompactRev, _ := s.d.CurrentRevision(s.ctx)
+	targetCompactRev, _ := s.CurrentRevision(s.ctx)
 	logrus.Tracef("COMPACT starting compactRev=%d targetCompactRev=%d", compactRev, targetCompactRev)
 
 outer:
@@ -233,6 +234,9 @@ func (s *SQLLog) postCompact() error {
 }
 
 func (s *SQLLog) CurrentRevision(ctx context.Context) (int64, error) {
+	if s.currentRev != 0 {
+		return s.currentRev, nil
+	}
 	return s.d.CurrentRevision(ctx)
 }
 
@@ -254,7 +258,7 @@ func (s *SQLLog) After(ctx context.Context, prefix string, revision, limit int64
 
 	if revision > 0 && len(result) == 0 {
 		// a zero length result won't have the compact or current revisions so get them manually
-		rev, err = s.d.CurrentRevision(ctx)
+		rev, err = s.CurrentRevision(ctx)
 		if err != nil {
 			return 0, nil, err
 		}
@@ -305,7 +309,7 @@ func (s *SQLLog) List(ctx context.Context, prefix, startKey string, limit, revis
 
 	if revision > 0 && len(result) == 0 {
 		// a zero length result won't have the compact or current revisions so get them manually
-		rev, err = s.d.CurrentRevision(ctx)
+		rev, err = s.CurrentRevision(ctx)
 		if err != nil {
 			return 0, nil, err
 		}
@@ -386,7 +390,7 @@ func filter(events interface{}, checkPrefix bool, prefix string) ([]*server.Even
 }
 
 func (s *SQLLog) startWatch() (chan interface{}, error) {
-	pollStart, err := s.d.GetCompactRevision(s.ctx)
+	pollStart, err := s.d.CurrentRevision(s.ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -400,8 +404,9 @@ func (s *SQLLog) startWatch() (chan interface{}, error) {
 }
 
 func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
+	s.currentRev = pollStart
+
 	var (
-		last        = pollStart
 		skip        int64
 		skipTime    time.Time
 		waitForMore = true
@@ -417,15 +422,15 @@ func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
 			case <-s.ctx.Done():
 				return
 			case check := <-s.notify:
-				if check <= last {
+				if check <= s.currentRev {
 					continue
 				}
 			case <-wait.C:
 			}
 		}
 		waitForMore = true
 
-		rows, err := s.d.After(s.ctx, "%", last, pollBatchSize)
+		rows, err := s.d.After(s.ctx, "%", s.currentRev, pollBatchSize)
 		if err != nil {
 			logrus.Errorf("fail to list latest changes: %v", err)
 			continue
@@ -437,15 +442,15 @@ func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
 			continue
 		}
 
-		logrus.Tracef("POLL AFTER %d, limit=%d, events=%d", last, pollBatchSize, len(events))
+		logrus.Tracef("POLL AFTER %d, limit=%d, events=%d", s.currentRev, pollBatchSize, len(events))
 
 		if len(events) == 0 {
 			continue
 		}
 
 		waitForMore = len(events) < 100
 
-		rev := last
+		rev := s.currentRev
 		var (
 			sequential []*server.Event
 			saveLast   bool
@@ -507,7 +512,7 @@ func (s *SQLLog) poll(result chan interface{}, pollStart int64) {
 		}
 
 		if saveLast {
-			last = rev
+			s.currentRev = rev
 			if len(sequential) > 0 {
 				result <- sequential
 			}

pkg/server/limited.go

@@ -2,13 +2,15 @@ package server
 
 import (
 	"context"
+	"time"
 
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 )
 
 type LimitedServer struct {
-	backend Backend
-	scheme  string
+	notifyInterval time.Duration
+	backend        Backend
+	scheme         string
 }
 
 func (l *LimitedServer) Range(ctx context.Context, r *etcdserverpb.RangeRequest) (*RangeResponse, error) {

pkg/server/server.go

@@ -1,6 +1,8 @@
 package server
 
 import (
+	"time"
+
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health"
@@ -12,11 +14,12 @@ type KVServerBridge struct {
 	limited *LimitedServer
 }
 
-func New(backend Backend, scheme string) *KVServerBridge {
+func New(backend Backend, scheme string, notifyInterval time.Duration) *KVServerBridge {
 	return &KVServerBridge{
 		limited: &LimitedServer{
-			backend: backend,
-			scheme:  scheme,
+			notifyInterval: notifyInterval,
+			backend:        backend,
+			scheme:         scheme,
 		},
 	}
 }