integration: add goleak (#173)

integration_tests/1pc_test.go

@@ -58,6 +58,10 @@ func (s *testOnePCSuite) SetupTest() {
 	s.bo = tikv.NewBackofferWithVars(context.Background(), 5000, nil)
 }
 
+func (s *testOnePCSuite) TearDownTest() {
+	s.testAsyncCommitCommon.tearDownTest()
+}
+
 func (s *testOnePCSuite) Test1PC() {
 	ctx := context.WithValue(context.Background(), util.SessionID, uint64(1))
 

integration_tests/2pc_test.go

@@ -83,6 +83,11 @@ func (s *testCommitterSuite) SetupSuite() {
 	atomic.StoreUint64(&tikv.VeryLongMaxBackoff, 1000)
 }
 
+func (s *testCommitterSuite) TearDownSuite() {
+	atomic.StoreUint64(&tikv.CommitMaxBackoff, 20000)
+	atomic.StoreUint64(&tikv.VeryLongMaxBackoff, 600000)
+}
+
 func (s *testCommitterSuite) SetupTest() {
 	mvccStore, err := mocktikv.NewMVCCLevelDB("")
 	s.Require().Nil(err)
@@ -99,9 +104,7 @@ func (s *testCommitterSuite) SetupTest() {
 	s.store = tikv.StoreProbe{KVStore: store}
 }
 
-func (s *testCommitterSuite) TearDownSuite() {
-	atomic.StoreUint64(&tikv.CommitMaxBackoff, 20000)
-	atomic.StoreUint64(&tikv.VeryLongMaxBackoff, 600000)
+func (s *testCommitterSuite) TearDownTest() {
 	s.store.Close()
 }
 

integration_tests/async_commit_fail_test.go

@@ -61,6 +61,10 @@ func (s *testAsyncCommitFailSuite) SetupTest() {
 	s.testAsyncCommitCommon.setUpTest()
 }
 
+func (s *testAsyncCommitFailSuite) TearDownTest() {
+	s.testAsyncCommitCommon.tearDownTest()
+}
+
 // TestFailCommitPrimaryRpcErrors tests rpc errors are handled properly when
 // committing primary region task.
 func (s *testAsyncCommitFailSuite) TestFailAsyncCommitPrewriteRpcErrors() {

integration_tests/async_commit_test.go

@@ -74,6 +74,7 @@ func (s *testAsyncCommitCommon) setUpTest() {
 
 	client, pdClient, cluster, err := unistore.New("")
 	s.Require().Nil(err)
+
 	unistore.BootstrapWithSingleStore(cluster)
 	s.cluster = cluster
 	store, err := tikv.NewTestTiKVStore(fpClient{Client: client}, pdClient, nil, nil, 0)
@@ -82,6 +83,10 @@ func (s *testAsyncCommitCommon) setUpTest() {
 	s.store = store
 }
 
+func (s *testAsyncCommitCommon) tearDownTest() {
+	s.store.Close()
+}
+
 func (s *testAsyncCommitCommon) putAlphabets(enableAsyncCommit bool) {
 	for ch := byte('a'); ch <= byte('z'); ch++ {
 		s.putKV([]byte{ch}, []byte{ch}, enableAsyncCommit)
@@ -180,6 +185,10 @@ func (s *testAsyncCommitSuite) SetupTest() {
 	s.bo = tikv.NewBackofferWithVars(context.Background(), 5000, nil)
 }
 
+func (s *testAsyncCommitSuite) TearDownTest() {
+	s.testAsyncCommitCommon.tearDownTest()
+}
+
 func (s *testAsyncCommitSuite) lockKeysWithAsyncCommit(keys, values [][]byte, primaryKey, primaryValue []byte, commitPrimary bool) (uint64, uint64) {
 	txn, err := s.store.Begin()
 	s.Nil(err)

integration_tests/go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/tikv/client-go/v2 v2.0.0
 	github.com/tikv/pd v1.1.0-beta.0.20210323121136-78679e5e209d
+	go.uber.org/goleak v1.1.10
 	go.uber.org/zap v1.17.0
 )
 

integration_tests/main_test.go

@@ -17,9 +17,16 @@ import (
 	"testing"
 
 	"github.com/tikv/client-go/v2/tikv"
+	"go.uber.org/goleak"
 )
 
 func TestMain(m *testing.M) {
 	tikv.EnableFailpoints()
-	m.Run()
+
+	opts := []goleak.Option{
+		goleak.IgnoreTopFunction("github.com/pingcap/goleveldb/leveldb.(*DB).mpoolDrain"),
+		goleak.IgnoreTopFunction("github.com/tikv/client-go/v2/tikv.(*ttlManager).keepAlive"), // TODO: fix ttlManager goroutine leak
+	}
+
+	goleak.VerifyTestMain(m, opts...)
 }

integration_tests/prewrite_test.go

@@ -50,6 +50,7 @@ func TestSetMinCommitTSInAsyncCommit(t *testing.T) {
 	unistore.BootstrapWithSingleStore(cluster)
 	store, err := tikv.NewTestTiKVStore(client, pdClient, nil, nil, 0)
 	require.Nil(err)
+	defer store.Close()
 
 	tx, err := store.Begin()
 	require.Nil(err)

integration_tests/snapshot_fail_test.go

@@ -64,6 +64,10 @@ func (s *testSnapshotFailSuite) SetupSuite() {
 	s.store = tikv.StoreProbe{KVStore: store}
 }
 
+func (s *testSnapshotFailSuite) TearDownSuite() {
+	s.store.Close()
+}
+
 func (s *testSnapshotFailSuite) TearDownTest() {
 	txn, err := s.store.Begin()
 	s.Require().Nil(err)

integration_tests/split_test.go

@@ -79,6 +79,10 @@ func (s *testSplitSuite) SetupTest() {
 	s.bo = tikv.NewBackofferWithVars(context.Background(), 5000, nil)
 }
 
+func (s *testSplitSuite) TearDownTest() {
+	s.store.Close()
+}
+
 func (s *testSplitSuite) begin() tikv.TxnProbe {
 	txn, err := s.store.Begin()
 	s.Require().Nil(err)

integration_tests/store_test.go

@@ -65,6 +65,7 @@ func (s *testStoreSuite) TearDownTest() {
 }
 
 func (s *testStoreSuite) TestOracle() {
+	s.store.GetOracle().Close()
 	o := &oracles.MockOracle{}
 	s.store.SetOracle(o)
 

tikv/test_probe.go

@@ -82,8 +82,11 @@ func (s StoreProbe) SetRegionCachePDClient(client pd.Client) {
 
 // ClearTxnLatches clears store's txn latch scheduler.
 func (s StoreProbe) ClearTxnLatches() {
+	if s.txnLatches != nil {
+		s.txnLatches.Close()
 		s.txnLatches = nil
 	}
+}
 
 // SendTxnHeartbeat renews a txn's ttl.
 func (s StoreProbe) SendTxnHeartbeat(ctx context.Context, key []byte, startTS uint64, ttl uint64) (uint64, error) {