pkg/util: fix panic caused by logging grpc err (#52179) (#52187)

close #51301

util/topsql/reporter/BUILD.bazel

@@ -52,6 +52,7 @@ go_test(
         "//util/topsql/reporter/mock",
         "//util/topsql/state",
         "//util/topsql/stmtstats",
+        "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_tipb//go-tipb",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",

util/topsql/reporter/pubsub.go

@@ -19,6 +19,8 @@ import (
 	"errors"
 	"time"
 
+	tidberrors "github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/util"
 	"github.com/pingcap/tidb/util/logutil"
 	reporter_metrics "github.com/pingcap/tidb/util/topsql/reporter/metrics"
@@ -97,6 +99,11 @@ func (ds *pubSubDataSink) OnReporterClosing() {
 
 func (ds *pubSubDataSink) run() error {
 	defer func() {
+		if r := recover(); r != nil {
+			err := tidberrors.Errorf("%v", r)
+			// To catch panic when log grpc error. https://github.com/pingcap/tidb/issues/51301.
+			logutil.BgLogger().Error("[top-sql] got panic in pub sub data sink, just ignore", zap.Error(err))
+		}
 		ds.registerer.Deregister(ds)
 		ds.cancel()
 	}()
@@ -133,6 +140,7 @@ func (ds *pubSubDataSink) run() error {
 				return ctx.Err()
 			}
 
+			failpoint.Inject("mockGrpcLogPanic", nil)
 			if err != nil {
 				logutil.BgLogger().Warn(
 					"[top-sql] pubsub datasink failed to send data to subscriber",

util/topsql/reporter/pubsub_test.go

@@ -20,8 +20,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tipb/go-tipb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/metadata"
 )
 
@@ -85,6 +87,8 @@ func TestPubSubDataSink(t *testing.T) {
 		_ = ds.run()
 	}()
 
+	panicPath := "github.com/pingcap/tidb/pkg/util/topsql/reporter/mockGrpcLogPanic"
+	require.NoError(t, failpoint.Enable(panicPath, "panic"))
 	err := ds.TrySend(&ReportData{
 		DataRecords: []tipb.TopSQLRecord{{
 			SqlDigest:  []byte("S1"),
@@ -117,4 +121,5 @@ func TestPubSubDataSink(t *testing.T) {
 	mockStream.Unlock()
 
 	ds.OnReporterClosing()
+	require.NoError(t, failpoint.Disable(panicPath))
 }