*: add check when watching (pingcap#6671)

* *: add check when watching

ddl/mock.go

@@ -50,6 +50,9 @@ func (s *mockSchemaSyncer) GlobalVersionCh() clientv3.WatchChan {
 	return s.globalVerCh
 }
 
+// WatchGlobalSchemaVer implements SchemaSyncer.WatchGlobalSchemaVer interface.
+func (s *mockSchemaSyncer) WatchGlobalSchemaVer(context.Context) {}
+
 // UpdateSelfVersion implements SchemaSyncer.UpdateSelfVersion interface.
 func (s *mockSchemaSyncer) UpdateSelfVersion(ctx context.Context, version int64) error {
 	atomic.StoreInt64(&s.selfSchemaVersion, version)

ddl/syncer.go

@@ -17,6 +17,7 @@ import (
 	"fmt"
 	"math"
 	"strconv"
+	"sync"
 	"time"
 
 	"github.com/coreos/etcd/clientv3"
@@ -68,6 +69,8 @@ type SchemaSyncer interface {
 	OwnerUpdateGlobalVersion(ctx context.Context, version int64) error
 	// GlobalVersionCh gets the chan for watching global version.
 	GlobalVersionCh() clientv3.WatchChan
+	// WatchGlobalSchemaVer watches the global schema version.
+	WatchGlobalSchemaVer(ctx context.Context)
 	// MustGetGlobalVersion gets the global version. The only reason it fails is that ctx is done.
 	MustGetGlobalVersion(ctx context.Context) (int64, error)
 	// Done returns a channel that closes when the syncer is no longer being refreshed.
@@ -84,7 +87,10 @@ type schemaVersionSyncer struct {
 	selfSchemaVerPath string
 	etcdCli           *clientv3.Client
 	session           *concurrency.Session
-	globalVerCh       clientv3.WatchChan
+	mu                struct {
+		sync.RWMutex
+		globalVerCh clientv3.WatchChan
+	}
 }
 
 // NewSchemaSyncer creates a new SchemaSyncer.
@@ -135,7 +141,11 @@ func (s *schemaVersionSyncer) Init(ctx context.Context) error {
 	if err != nil {
 		return errors.Trace(err)
 	}
-	s.globalVerCh = s.etcdCli.Watch(ctx, DDLGlobalSchemaVersion)
+
+	s.mu.Lock()
+	s.mu.globalVerCh = s.etcdCli.Watch(ctx, DDLGlobalSchemaVersion)
+	s.mu.Unlock()
+
 	err = s.putKV(ctx, keyOpDefaultRetryCnt, s.selfSchemaVerPath, InitialVersion,
 		clientv3.WithLease(s.session.Lease()))
 	return errors.Trace(err)
@@ -172,7 +182,30 @@ func (s *schemaVersionSyncer) Restart(ctx context.Context) error {
 
 // GlobalVersionCh implements SchemaSyncer.GlobalVersionCh interface.
 func (s *schemaVersionSyncer) GlobalVersionCh() clientv3.WatchChan {
-	return s.globalVerCh
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.mu.globalVerCh
+}
+
+// WatchGlobalSchemaVer implements SchemaSyncer.WatchGlobalSchemaVer interface.
+func (s *schemaVersionSyncer) WatchGlobalSchemaVer(ctx context.Context) {
+	startTime := time.Now()
+	// Make sure the globalVerCh doesn't receive the information of 'close' before we finish the rewatch.
+	s.mu.Lock()
+	s.mu.globalVerCh = nil
+	s.mu.Unlock()
+
+	go func() {
+		defer func() {
+			metrics.DeploySyncerHistogram.WithLabelValues(metrics.SyncerRewatch, metrics.RetLabel(nil)).Observe(time.Since(startTime).Seconds())
+		}()
+		ch := s.etcdCli.Watch(ctx, DDLGlobalSchemaVersion)
+
+		s.mu.Lock()
+		s.mu.globalVerCh = ch
+		s.mu.Unlock()
+		log.Info("[syncer] watch global schema finished")
+	}()
 }
 
 // UpdateSelfVersion implements SchemaSyncer.UpdateSelfVersion interface.

domain/domain.go

@@ -338,11 +338,16 @@ func (do *Domain) loadSchemaInLoop(lease time.Duration) {
 			if err != nil {
 				log.Errorf("[ddl] reload schema in loop err %v", errors.ErrorStack(err))
 			}
-		case <-syncer.GlobalVersionCh():
+		case _, ok := <-syncer.GlobalVersionCh():
 			err := do.Reload()
 			if err != nil {
 				log.Errorf("[ddl] reload schema in loop err %v", errors.ErrorStack(err))
 			}
+			if !ok {
+				log.Warn("[ddl] reload schema in loop, schema syncer need rewatch")
+				// Make sure the rewatch doesn't affect load schema, so we watch the global schema version asynchronously.
+				syncer.WatchGlobalSchemaVer(context.Background())
+			}
 		case <-syncer.Done():
 			// The schema syncer stops, we need stop the schema validator to synchronize the schema version.
 			log.Info("[ddl] reload schema in loop, schema syncer need restart")

metrics/ddl.go

@@ -43,10 +43,10 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20),
 		}, []string{LblType})
 
-	SyncerInit    = "init"
-	SyncerRestart = "restart"
-	SyncerClear   = "clear"
-
+	SyncerInit            = "init"
+	SyncerRestart         = "restart"
+	SyncerClear           = "clear"
+	SyncerRewatch         = "rewatch"
 	DeploySyncerHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "tidb",

metrics/owner.go

@@ -28,6 +28,7 @@ var (
 			Buckets:   prometheus.ExponentialBuckets(0.0005, 2, 22),
 		}, []string{LblType, LblResult})
 
+	WatcherClosed     = "watcher_closed"
 	Cancelled         = "cancelled"
 	Deleted           = "deleted"
 	SessionDone       = "session_done"

owner/manager.go

@@ -281,10 +281,15 @@ func (m *ownerManager) watchOwner(ctx context.Context, etcdSession *concurrency.
 	watchCh := m.etcdCli.Watch(ctx, key)
 	for {
 		select {
-		case resp := <-watchCh:
+		case resp, ok := <-watchCh:
+			if !ok {
+				metrics.WatchOwnerCounter.WithLabelValues(m.prompt, metrics.WatcherClosed).Inc()
+				log.Infof("%s watcher is closed, no owner", logPrefix)
+				return
+			}
 			if resp.Canceled {
 				metrics.WatchOwnerCounter.WithLabelValues(m.prompt, metrics.Cancelled).Inc()
-				log.Infof("%s failed, no owner", logPrefix)
+				log.Infof("%s canceled, no owner", logPrefix)
 				return
 			}