VReplication: Auto renew lease on Keyspace Lock

go/vt/topo/etcd2topo/lock.go

@@ -22,7 +22,6 @@ import (
 	"path"
 
 	"github.com/spf13/pflag"
-
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 

go/vt/topo/keyspace_lock.go

@@ -19,6 +19,16 @@ package topo
 import (
 	"context"
 	"path"
+	"time"
+
+	"vitess.io/vitess/go/vt/vterrors"
+
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+)
+
+const (
+	MaxKeyspaceLockLeaseTTL = 10 * time.Minute
+	leaseRenewalInterval    = 10 * time.Second
 )
 
 type keyspaceLock struct {
@@ -56,3 +66,52 @@ func CheckKeyspaceLocked(ctx context.Context, keyspace string) error {
 		keyspace: keyspace,
 	})
 }
+
+// LockKeyspaceWithLeaseRenewal locks the keyspace and starts a goroutine which runs
+// until the MaxKeyspaceLockLeaseTTL is reached -- exiting if the context is
+// cancelled, the unlock function is called, or an error is encountered -- refreshing
+// the lock's lease every leaseRenewal until it ends.
+// It returns a read-only error channel that you should regularly check to ensure that
+// you have not lost the lock or encountered any other non-recoverable errors related
+// to your lease.
+func (ts *Server) LockKeyspaceWithLeaseRenewal(ctx context.Context, keyspace, action string) (context.Context, func(*error), <-chan error, error) {
+	ksLock := &keyspaceLock{keyspace: keyspace}
+	lockCtx, unlockF, err := ts.internalLock(ctx, ksLock, action, true)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	done := make(chan struct{})  // Our work is done and we should exit
+	errCh := make(chan error, 1) // Communicate any errors encountered during renewals
+	go func() {
+		defer close(errCh)
+		maxRenewals := int((MaxKeyspaceLockLeaseTTL.Seconds() - leaseRenewalInterval.Seconds()) / leaseRenewalInterval.Seconds())
+		for i := 0; i < maxRenewals; i++ {
+			time.Sleep(leaseRenewalInterval)
+			select {
+			case <-lockCtx.Done():
+				return
+			case <-done:
+				return
+			default:
+				// Attempt to renew the lease.
+				if err := checkLocked(lockCtx, ksLock); err != nil {
+					errCh <- vterrors.Wrapf(err, "failed to renew keyspace %s lock lease", keyspace)
+					return
+				}
+			}
+		}
+		time.Sleep(leaseRenewalInterval)
+		select {
+		case <-done:
+		default:
+			errCh <- vterrors.Errorf(vtrpcpb.Code_DEADLINE_EXCEEDED, "cannot renew keyspace %s lock lease as we've hit the time limit of %v",
+				keyspace, MaxKeyspaceLockLeaseTTL)
+		}
+	}()
+	// Add to the unlock function to end the lease renewal work.
+	newUnlockF := func(err *error) {
+		close(done)
+		unlockF(err)
+	}
+	return lockCtx, newUnlockF, errCh, nil
+}

go/vt/vtctl/workflow/resharder.go

@@ -99,6 +99,9 @@ func (s *Server) buildResharder(ctx context.Context, keyspace, workflow string,
 		if err != nil {
 			return nil, vterrors.Wrapf(err, "GetShard(%s) failed", shard)
 		}
+		if si.PrimaryAlias == nil {
+			return nil, fmt.Errorf("target shard %v has no primary tablet", shard)
+		}
 		if si.IsPrimaryServing {
 			return nil, fmt.Errorf("target shard %v is in serving state", shard)
 		}