*: pick some fixes (tikv#76)

* fix tso inconsistent problem Signed-off-by: Ryan Leung <rleungx@gmail.com> * *: fix the missing log panic (tikv#6325) close tikv#6257 Signed-off-by: Ryan Leung <rleungx@gmail.com> * *: add defer to logs the panic reason and stack (tikv#6123) ref tikv#6099 add defer logs the panic reason and stack Signed-off-by: husharp <jinhao.hu@pingcap.com> Co-authored-by: Ti Chi Robot <ti-community-prow-bot@tidb.io> --------- Signed-off-by: Ryan Leung <rleungx@gmail.com> Co-authored-by: Hu# <jinhao.hu@pingcap.com> Co-authored-by: Ti Chi Robot <ti-community-prow-bot@tidb.io>
rleungx · Apr 21, 2023 · 5a11b83 · 5a11b83
1 parent fdbdef3
commit 5a11b83
Show file tree

Hide file tree

Showing 24 changed files with 79 additions and 4 deletions.
diff --git a/pkg/cache/ttl.go b/pkg/cache/ttl.go
@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 	"go.uber.org/zap"
 )
@@ -142,6 +143,7 @@ func (c *ttlCache) Clear() {
 }
 
 func (c *ttlCache) doGC() {
+	defer logutil.LogPanic()
 	ticker := time.NewTicker(c.gcInterval)
 	defer ticker.Stop()
 

diff --git a/pkg/election/lease.go b/pkg/election/lease.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
@@ -94,6 +95,8 @@ func (l *lease) IsExpired() bool {
 
 // KeepAlive auto renews the lease and update expireTime.
 func (l *lease) KeepAlive(ctx context.Context) {
+	defer logutil.LogPanic()
+
 	if l == nil {
 		return
 	}
@@ -129,6 +132,7 @@ func (l *lease) keepAliveWorker(ctx context.Context, interval time.Duration) <-c
 	ch := make(chan time.Time)
 
 	go func() {
+		defer logutil.LogPanic()
 		ticker := time.NewTicker(interval)
 		defer ticker.Stop()
 
@@ -137,6 +141,7 @@ func (l *lease) keepAliveWorker(ctx context.Context, interval time.Duration) <-c
 
 		for {
 			go func() {
+				defer logutil.LogPanic()
 				start := time.Now()
 				ctx1, cancel := context.WithTimeout(ctx, l.leaseTimeout)
 				defer cancel()

diff --git a/pkg/gctuner/memory_limit_tuner.go b/pkg/gctuner/memory_limit_tuner.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/log"
 	util "github.com/tikv/pd/pkg/gogc"
 	"github.com/tikv/pd/pkg/memory"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	atomicutil "go.uber.org/atomic"
 	"go.uber.org/zap"
 )
@@ -72,6 +73,7 @@ func (t *memoryLimitTuner) tuning() {
 	if float64(r.HeapInuse)*ratio > float64(setMemoryLimit(-1)) {
 		if t.nextGCTriggeredByMemoryLimit.Load() && t.waitingReset.CompareAndSwap(false, true) {
 			go func() {
+				defer logutil.LogPanic()
 				memory.MemoryLimitGCLast.Store(time.Now())
 				memory.MemoryLimitGCTotal.Add(1)
 				setMemoryLimit(t.calcMemoryLimit(fallbackPercentage))

diff --git a/pkg/mcs/discovery/register.go b/pkg/mcs/discovery/register.go
@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/pingcap/log"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
 )
@@ -70,6 +71,7 @@ func (sr *ServiceRegister) Register() error {
 		return fmt.Errorf("keepalive failed: %v", err)
 	}
 	go func() {
+		defer logutil.LogPanic()
 		for {
 			select {
 			case <-sr.ctx.Done():

diff --git a/pkg/mcs/resource_manager/server/manager.go b/pkg/mcs/resource_manager/server/manager.go
@@ -29,6 +29,7 @@ import (
 	bs "github.com/tikv/pd/pkg/basicserver"
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/storage/kv"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"go.uber.org/zap"
 )
 
@@ -121,7 +122,10 @@ func (m *Manager) Init(ctx context.Context) {
 	m.storage.LoadResourceGroupStates(tokenHandler)
 	// Start the background metrics flusher.
 	go m.backgroundMetricsFlush(ctx)
-	go m.persistLoop(ctx)
+	go func() {
+		defer logutil.LogPanic()
+		m.persistLoop(ctx)
+	}()
 	log.Info("resource group manager finishes initialization")
 }
 
@@ -249,6 +253,7 @@ func (m *Manager) persistResourceGroupRunningState() {
 
 // Receive the consumption and flush it to the metrics.
 func (m *Manager) backgroundMetricsFlush(ctx context.Context) {
+	defer logutil.LogPanic()
 	ticker := time.NewTicker(metricsCleanupInterval)
 	defer ticker.Stop()
 	for {

diff --git a/pkg/mcs/resource_manager/server/server.go b/pkg/mcs/resource_manager/server/server.go
@@ -168,6 +168,7 @@ func (s *Server) initClient() error {
 }
 
 func (s *Server) startGRPCServer(l net.Listener) {
+	defer logutil.LogPanic()
 	defer s.serverLoopWg.Done()
 
 	gs := grpc.NewServer()
@@ -179,6 +180,7 @@ func (s *Server) startGRPCServer(l net.Listener) {
 	// it doesn't happen in a reasonable amount of time.
 	done := make(chan struct{})
 	go func() {
+		defer logutil.LogPanic()
 		log.Info("try to gracefully stop the server now")
 		gs.GracefulStop()
 		close(done)
@@ -197,6 +199,7 @@ func (s *Server) startGRPCServer(l net.Listener) {
 }
 
 func (s *Server) startHTTPServer(l net.Listener) {
+	defer logutil.LogPanic()
 	defer s.serverLoopWg.Done()
 
 	handler, _ := SetUpRestHandler(s.service)
@@ -223,6 +226,7 @@ func (s *Server) startHTTPServer(l net.Listener) {
 }
 
 func (s *Server) startGRPCAndHTTPServers(l net.Listener) {
+	defer logutil.LogPanic()
 	defer s.serverLoopWg.Done()
 
 	mux := cmux.New(l)

diff --git a/pkg/mcs/tso/server/grpc_service.go b/pkg/mcs/tso/server/grpc_service.go
@@ -29,6 +29,7 @@ import (
 	"github.com/tikv/pd/pkg/mcs/registry"
 	"github.com/tikv/pd/pkg/utils/apiutil"
 	"github.com/tikv/pd/pkg/utils/grpcutil"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -191,6 +192,7 @@ func (s *Service) dispatchTSORequest(ctx context.Context, request *tsoRequest, f
 }
 
 func (s *Service) handleDispatcher(ctx context.Context, forwardedHost string, tsoRequestCh <-chan *tsoRequest, tsDeadlineCh chan<- deadline, doneCh <-chan struct{}, errCh chan<- error) {
+	defer logutil.LogPanic()
 	dispatcherCtx, ctxCancel := context.WithCancel(ctx)
 	defer ctxCancel()
 	defer s.tsoDispatcher.Delete(forwardedHost)
@@ -330,6 +332,7 @@ type deadline struct {
 }
 
 func watchTSDeadline(ctx context.Context, tsDeadlineCh <-chan deadline) {
+	defer logutil.LogPanic()
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	for {

diff --git a/pkg/mcs/tso/server/server.go b/pkg/mcs/tso/server/server.go
@@ -457,6 +457,7 @@ func (s *Server) SetExternalTS(externalTS uint64) error {
 }
 
 func checkStream(streamCtx context.Context, cancel context.CancelFunc, done chan struct{}) {
+	defer logutil.LogPanic()
 	select {
 	case <-done:
 		return
@@ -491,6 +492,7 @@ func (s *Server) initClient() error {
 }
 
 func (s *Server) startGRPCServer(l net.Listener) {
+	defer logutil.LogPanic()
 	defer s.serverLoopWg.Done()
 
 	gs := grpc.NewServer()
@@ -503,6 +505,7 @@ func (s *Server) startGRPCServer(l net.Listener) {
 	// it doesn't happen in a reasonable amount of time.
 	done := make(chan struct{})
 	go func() {
+		defer logutil.LogPanic()
 		log.Info("try to gracefully stop the server now")
 		gs.GracefulStop()
 		close(done)
@@ -522,6 +525,7 @@ func (s *Server) startGRPCServer(l net.Listener) {
 }
 
 func (s *Server) startHTTPServer(l net.Listener) {
+	defer logutil.LogPanic()
 	defer s.serverLoopWg.Done()
 
 	handler, _ := SetUpRestHandler(s.service)
@@ -548,6 +552,7 @@ func (s *Server) startHTTPServer(l net.Listener) {
 }
 
 func (s *Server) startGRPCAndHTTPServers(l net.Listener) {
+	defer logutil.LogPanic()
 	defer s.serverLoopWg.Done()
 
 	mux := cmux.New(l)

diff --git a/pkg/storage/endpoint/gc_key_space.go b/pkg/storage/endpoint/gc_key_space.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
 )
@@ -77,6 +78,8 @@ func (se *StorageEndpoint) LoadServiceSafePoint(spaceID, serviceID string) (*Ser
 	}
 	if ssp.ExpiredAt < time.Now().Unix() {
 		go func() {
+			defer logutil.LogPanic()
+
 			if err = se.Remove(key); err != nil {
 				log.Error("remove expired key meet error", zap.String("key", key), errs.ZapError(err))
 			}
@@ -124,6 +127,8 @@ func (se *StorageEndpoint) LoadMinServiceSafePoint(spaceID string, now time.Time
 	})
 	// remove expired keys asynchronously
 	go func() {
+		defer logutil.LogPanic()
+
 		for _, key := range expiredKeys {
 			if err = se.Remove(key); err != nil {
 				log.Error("remove expired key meet error", zap.String("key", key), errs.ZapError(err))

diff --git a/pkg/storage/endpoint/tso.go b/pkg/storage/endpoint/tso.go
@@ -19,6 +19,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/storage/kv"
 	"github.com/tikv/pd/pkg/utils/typeutil"
@@ -82,7 +83,7 @@ func (se *StorageEndpoint) SaveTimestamp(key string, ts time.Time) error {
 			}
 		}
 		if previousTS != typeutil.ZeroTime && typeutil.SubRealTimeByWallClock(ts, previousTS) <= 0 {
-			return nil
+			return errors.Errorf("saving timestamp %d is less than or equal to the previous one %d", ts.UnixNano(), previousTS.UnixNano())
 		}
 		data := typeutil.Uint64ToBytes(uint64(ts.UnixNano()))
 		return txn.Save(key, string(data))

diff --git a/pkg/storage/hot_region_storage.go b/pkg/storage/hot_region_storage.go
@@ -34,6 +34,7 @@ import (
 	"github.com/tikv/pd/pkg/encryption"
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/storage/kv"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 	"go.uber.org/zap"
 )
@@ -162,6 +163,8 @@ func NewHotRegionsStorage(
 
 // Delete hot region whose update_time is smaller than time.Now() minus remain day in the background.
 func (h *HotRegionStorage) backgroundDelete() {
+	defer logutil.LogPanic()
+
 	// make delete happened in defaultDeleteTime clock.
 	now := time.Now()
 	next := time.Date(now.Year(), now.Month(), now.Day(), defaultDeleteTime, 0, 0, 0, now.Location())
@@ -198,6 +201,8 @@ func (h *HotRegionStorage) backgroundDelete() {
 
 // Write hot_region info into db in the background.
 func (h *HotRegionStorage) backgroundFlush() {
+	defer logutil.LogPanic()
+
 	interval := h.getCurInterval()
 	ticker := time.NewTicker(interval)
 	defer func() {

diff --git a/pkg/storage/leveldb_backend.go b/pkg/storage/leveldb_backend.go
@@ -27,6 +27,7 @@ import (
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/storage/endpoint"
 	"github.com/tikv/pd/pkg/storage/kv"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 )
 
@@ -80,6 +81,8 @@ func newLevelDBBackend(
 var dirtyFlushTick = time.Second
 
 func (lb *levelDBBackend) backgroundFlush() {
+	defer logutil.LogPanic()
+
 	var (
 		isFlush bool
 		err     error

diff --git a/pkg/storage/storage_tso_test.go b/pkg/storage/storage_tso_test.go
@@ -116,7 +116,7 @@ func TestTimestampTxn(t *testing.T) {
 
 	globalTS2 := globalTS1.Add(-time.Millisecond).Round(0)
 	err = storage.SaveTimestamp(timestampKey, globalTS2)
-	re.NoError(err)
+	re.Error(err)
 
 	ts, err := storage.LoadTimestamp("")
 	re.NoError(err)

diff --git a/pkg/systimemon/systimemon.go b/pkg/systimemon/systimemon.go
@@ -20,11 +20,13 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"go.uber.org/zap"
 )
 
 // StartMonitor calls systimeErrHandler if system time jump backward.
 func StartMonitor(ctx context.Context, now func() time.Time, systimeErrHandler func()) {
+	defer logutil.LogPanic()
 	log.Info("start system time monitor")
 	tick := time.NewTicker(100 * time.Millisecond)
 	defer tick.Stop()

diff --git a/pkg/tso/allocator_manager.go b/pkg/tso/allocator_manager.go
@@ -35,6 +35,7 @@ import (
 	"github.com/tikv/pd/pkg/storage/kv"
 	"github.com/tikv/pd/pkg/utils/etcdutil"
 	"github.com/tikv/pd/pkg/utils/grpcutil"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 	"go.etcd.io/etcd/clientv3"
 	"go.uber.org/zap"
@@ -399,6 +400,7 @@ func (am *AllocatorManager) getLocalTSOAllocatorPath() string {
 
 // similar logic with leaderLoop in server/server.go
 func (am *AllocatorManager) allocatorLeaderLoop(ctx context.Context, allocator *LocalTSOAllocator) {
+	defer logutil.LogPanic()
 	defer log.Info("server is closed, return local tso allocator leader loop",
 		zap.String("dc-location", allocator.GetDCLocation()),
 		zap.String("local-tso-allocator-name", am.member.Name()))
@@ -650,6 +652,7 @@ func (am *AllocatorManager) allocatorUpdater() {
 
 // updateAllocator is used to update the allocator in the group.
 func (am *AllocatorManager) updateAllocator(ag *allocatorGroup) {
+	defer logutil.LogPanic()
 	defer am.wg.Done()
 	select {
 	case <-ag.ctx.Done():
@@ -700,6 +703,7 @@ func (am *AllocatorManager) allocatorPatroller(serverCtx context.Context) {
 // ClusterDCLocationChecker collects all dc-locations of a cluster, computes some related info
 // and stores them into the DCLocationInfo, then finally writes them into am.mu.clusterDCLocations.
 func (am *AllocatorManager) ClusterDCLocationChecker() {
+	defer logutil.LogPanic()
 	// Wait for the group leader to be elected out.
 	if !am.member.IsLeaderElected() {
 		return
@@ -856,6 +860,8 @@ func (am *AllocatorManager) GetLocalTSOSuffixPath(dcLocation string) string {
 // 2. If all PD servers with dc-location="dc-1" are down, then the other PD servers
 // of DC could be elected.
 func (am *AllocatorManager) PriorityChecker() {
+	defer logutil.LogPanic()
+
 	serverID := am.member.ID()
 	myServerDCLocation := am.getServerDCLocation(serverID)
 	// Check all Local TSO Allocator followers to see if their priorities is higher than the leaders

diff --git a/pkg/tso/global_allocator.go b/pkg/tso/global_allocator.go
@@ -28,6 +28,7 @@ import (
 	"github.com/tikv/pd/pkg/election"
 	"github.com/tikv/pd/pkg/errs"
 	"github.com/tikv/pd/pkg/slice"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/tsoutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
 	"go.uber.org/zap"
@@ -340,6 +341,7 @@ func (gta *GlobalTSOAllocator) SyncMaxTS(
 			// Send SyncMaxTSRequest to all allocator leaders concurrently.
 			wg.Add(1)
 			go func(ctx context.Context, conn *grpc.ClientConn, respCh chan<- *syncResp) {
+				defer logutil.LogPanic()
 				defer wg.Done()
 				syncMaxTSResp := &syncResp{}
 				syncCtx, cancel := context.WithTimeout(ctx, rpcTimeout)

diff --git a/pkg/tso/local_allocator.go b/pkg/tso/local_allocator.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/tikv/pd/pkg/election"
 	"github.com/tikv/pd/pkg/errs"
+	"github.com/tikv/pd/pkg/utils/logutil"
 	"github.com/tikv/pd/pkg/utils/tsoutil"
 	"github.com/tikv/pd/pkg/utils/typeutil"
 	"go.etcd.io/etcd/clientv3"
@@ -174,6 +175,7 @@ func (lta *LocalTSOAllocator) CampaignAllocatorLeader(leaseTimeout int64, cmps .
 
 // KeepAllocatorLeader is used to keep the PD leader's leadership.
 func (lta *LocalTSOAllocator) KeepAllocatorLeader(ctx context.Context) {
+	defer logutil.LogPanic()
 	lta.leadership.Keep(ctx)
 }