Skip to content

Commit

Permalink
*: fix the missing log panic (#6325)
Browse files Browse the repository at this point in the history
close #6257

Signed-off-by: Ryan Leung <[email protected]>
  • Loading branch information
rleungx authored Apr 17, 2023
1 parent 36a8e15 commit 0e42a50
Show file tree
Hide file tree
Showing 15 changed files with 48 additions and 1 deletion.
2 changes: 2 additions & 0 deletions pkg/election/lease.go
Original file line number Diff line number Diff line change
Expand Up @@ -95,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
}
Expand Down
3 changes: 3 additions & 0 deletions pkg/schedule/labeler/labeler.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
"github.com/tikv/pd/pkg/errs"
"github.com/tikv/pd/pkg/schedule/rangelist"
"github.com/tikv/pd/pkg/storage/endpoint"
"github.com/tikv/pd/pkg/utils/logutil"
"github.com/tikv/pd/pkg/utils/syncutil"
"go.uber.org/zap"
)
Expand Down Expand Up @@ -56,6 +57,8 @@ func NewRegionLabeler(ctx context.Context, storage endpoint.RuleStorage, gcInter
}

func (l *RegionLabeler) doGC(gcInterval time.Duration) {
defer logutil.LogPanic()

ticker := time.NewTicker(gcInterval)
defer ticker.Stop()
for {
Expand Down
2 changes: 2 additions & 0 deletions pkg/statistics/buckets/hot_bucket_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,8 @@ func (h *HotBucketCache) CheckAsync(task flowBucketsItemTask) bool {
}

func (h *HotBucketCache) schedule() {
defer logutil.LogPanic()

for {
select {
case <-h.ctx.Done():
Expand Down
3 changes: 3 additions & 0 deletions pkg/statistics/hot_cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (

"github.com/smallnest/chanx"
"github.com/tikv/pd/pkg/core"
"github.com/tikv/pd/pkg/utils/logutil"
)

const chanMaxLength = 6000000
Expand Down Expand Up @@ -129,6 +130,8 @@ func (w *HotCache) ResetMetrics() {
}

func (w *HotCache) updateItems(queue *chanx.UnboundedChan[FlowItemTask], runTask func(task FlowItemTask)) {
defer logutil.LogPanic()

for {
select {
case <-w.ctx.Done():
Expand Down
5 changes: 5 additions & 0 deletions pkg/storage/endpoint/gc_key_space.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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))
}
Expand Down Expand Up @@ -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))
Expand Down
5 changes: 5 additions & 0 deletions pkg/storage/hot_region_storage.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)
Expand Down Expand Up @@ -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())
Expand Down Expand Up @@ -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() {
Expand Down
3 changes: 3 additions & 0 deletions pkg/storage/leveldb_backend.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down Expand Up @@ -80,6 +81,8 @@ func newLevelDBBackend(
var dirtyFlushTick = time.Second

func (lb *levelDBBackend) backgroundFlush() {
defer logutil.LogPanic()

var (
isFlush bool
err error
Expand Down
2 changes: 2 additions & 0 deletions pkg/tso/allocator_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -940,6 +940,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
Expand Down
1 change: 1 addition & 0 deletions pkg/tso/global_allocator.go
Original file line number Diff line number Diff line change
Expand Up @@ -475,6 +475,7 @@ func (gta *GlobalTSOAllocator) Reset() {
}

func (gta *GlobalTSOAllocator) primaryElectionLoop() {
defer logutil.LogPanic()
defer gta.wg.Done()

for {
Expand Down
3 changes: 3 additions & 0 deletions pkg/tso/keyspace_group_manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ func (s *state) deinitialize() {
if am != nil {
wg.Add(1)
go func(am *AllocatorManager) {
defer logutil.LogPanic()
defer wg.Done()
am.close()
log.Info("keyspace group closed", zap.Uint32("keyspace-group-id", am.kgID))
Expand Down Expand Up @@ -274,6 +275,8 @@ func (kgm *KeyspaceGroupManager) Close() {
}

func (kgm *KeyspaceGroupManager) checkInitProgress(ctx context.Context, cancel context.CancelFunc, done chan struct{}) {
defer logutil.LogPanic()

select {
case <-done:
return
Expand Down
2 changes: 2 additions & 0 deletions pkg/tso/local_allocator.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -175,6 +176,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)
}

Expand Down
3 changes: 3 additions & 0 deletions pkg/utils/metricutil/metricutil.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/push"
"github.com/tikv/pd/pkg/errs"
"github.com/tikv/pd/pkg/utils/logutil"
"github.com/tikv/pd/pkg/utils/typeutil"
)

Expand Down Expand Up @@ -63,6 +64,8 @@ func camelCaseToSnakeCase(str string) string {

// prometheusPushClient pushes metrics to Prometheus Pushgateway.
func prometheusPushClient(job, addr string, interval time.Duration) {
defer logutil.LogPanic()

pusher := push.New(addr, job).
Gatherer(prometheus.DefaultGatherer).
Grouping("instance", instanceName())
Expand Down
5 changes: 5 additions & 0 deletions pkg/utils/tsoutil/tso_dispatcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"github.com/pingcap/log"
"github.com/prometheus/client_golang/prometheus"
"github.com/tikv/pd/pkg/errs"
"github.com/tikv/pd/pkg/utils/logutil"
"go.uber.org/zap"
"google.golang.org/grpc"
)
Expand Down Expand Up @@ -71,6 +72,7 @@ func (s *TSODispatcher) DispatchRequest(
func (s *TSODispatcher) dispatch(
ctx context.Context, tsoProtoFactory ProtoFactory, forwardedHost string, clientConn *grpc.ClientConn,
tsoRequestCh <-chan Request, tsDeadlineCh chan<- deadline, doneCh <-chan struct{}, errCh chan<- error) {
defer logutil.LogPanic()
dispatcherCtx, ctxCancel := context.WithCancel(ctx)
defer ctxCancel()
defer s.dispatchChs.Delete(forwardedHost)
Expand Down Expand Up @@ -187,6 +189,7 @@ type deadline struct {
}

func watchTSDeadline(ctx context.Context, tsDeadlineCh <-chan deadline) {
defer logutil.LogPanic()
ctx, cancel := context.WithCancel(ctx)
defer cancel()
for {
Expand All @@ -209,6 +212,8 @@ func watchTSDeadline(ctx context.Context, tsDeadlineCh <-chan deadline) {
}

func checkStream(streamCtx context.Context, cancel context.CancelFunc, done chan struct{}) {
defer logutil.LogPanic()

select {
case <-done:
return
Expand Down
1 change: 1 addition & 0 deletions server/cluster/coordinator.go
Original file line number Diff line number Diff line change
Expand Up @@ -223,6 +223,7 @@ func (c *coordinator) checkPriorityRegions() {
// The regions of new version key range and old version key range would be placed into
// the suspect regions map
func (c *coordinator) checkSuspectRanges() {
defer logutil.LogPanic()
defer c.wg.Done()
log.Info("coordinator begins to check suspect key ranges")
ticker := time.NewTicker(checkSuspectRangesInterval)
Expand Down
9 changes: 8 additions & 1 deletion server/grpc_service.go
Original file line number Diff line number Diff line change
Expand Up @@ -560,6 +560,7 @@ func (b *bucketHeartbeatServer) Send(bucket *pdpb.ReportBucketsResponse) error {
}
done := make(chan error, 1)
go func() {
defer logutil.LogPanic()
done <- b.stream.SendAndClose(bucket)
}()
select {
Expand Down Expand Up @@ -598,7 +599,10 @@ func (s *heartbeatServer) Send(m *pdpb.RegionHeartbeatResponse) error {
return io.EOF
}
done := make(chan error, 1)
go func() { done <- s.stream.Send(m) }()
go func() {
defer logutil.LogPanic()
done <- s.stream.Send(m)
}()
select {
case err := <-done:
if err != nil {
Expand Down Expand Up @@ -1711,6 +1715,7 @@ func (s *GrpcServer) createHeartbeatForwardStream(client *grpc.ClientConn) (pdpb
}

func forwardRegionHeartbeatClientToServer(forwardStream pdpb.PD_RegionHeartbeatClient, server *heartbeatServer, errCh chan error) {
defer logutil.LogPanic()
defer close(errCh)
for {
resp, err := forwardStream.Recv()
Expand All @@ -1735,6 +1740,7 @@ func (s *GrpcServer) createReportBucketsForwardStream(client *grpc.ClientConn) (
}

func forwardReportBucketClientToServer(forwardStream pdpb.PD_ReportBucketsClient, server *bucketHeartbeatServer, errCh chan error) {
defer logutil.LogPanic()
defer close(errCh)
for {
resp, err := forwardStream.CloseAndRecv()
Expand All @@ -1751,6 +1757,7 @@ func forwardReportBucketClientToServer(forwardStream pdpb.PD_ReportBucketsClient

// TODO: If goroutine here timeout when tso stream created successfully, we need to handle it correctly.
func checkStream(streamCtx context.Context, cancel context.CancelFunc, done chan struct{}) {
defer logutil.LogPanic()
select {
case <-done:
return
Expand Down

0 comments on commit 0e42a50

Please sign in to comment.