<commit_before>// Copyright 2015 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package domain

import (
	"context"
	"crypto/tls"
	"math"
	"net"
	"runtime"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/ngaut/pools"
	. "github.com/pingcap/check"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/parser/ast"
	"github.com/pingcap/parser/model"
	"github.com/pingcap/tidb/ddl"
	"github.com/pingcap/tidb/domain/infosync"
	"github.com/pingcap/tidb/errno"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/meta"
	"github.com/pingcap/tidb/metrics"
	"github.com/pingcap/tidb/sessionctx/variable"
	"github.com/pingcap/tidb/store/mockstore"
	"github.com/pingcap/tidb/store/tikv/oracle"
	"github.com/pingcap/tidb/util"
	"github.com/pingcap/tidb/util/mock"
	"github.com/pingcap/tidb/util/testleak"
	dto "github.com/prometheus/client_model/go"
	"go.etcd.io/etcd/integration"
)

func TestT(t *testing.T) {
	CustomVerboseFlag = true
	TestingT(t)
}

var _ = Suite(&testSuite{})

type testSuite struct {
}

func mockFactory() (pools.Resource, error) {
	return nil, errors.New("mock factory should not be called")
}

func sysMockFactory(dom *Domain) (pools.Resource, error) {
	return nil, nil
}

type mockEtcdBackend struct {
	kv.Storage
	pdAddrs []string
}

func (mebd *mockEtcdBackend) EtcdAddrs() []string {
	return mebd.pdAddrs
}
func (mebd *mockEtcdBackend) TLSConfig() *tls.Config { return nil }
func (mebd *mockEtcdBackend) StartGCWorker() error {
	panic("not implemented")
}

// ETCD use ip:port as unix socket address, however this address is invalid on windows.
// We have to skip some of the test in such case.
// https://github.com/etcd-io/etcd/blob/f0faa5501d936cd8c9f561bb9d1baca70eb67ab1/pkg/types/urls.go#L42
func unixSocketAvailable() bool {
	c, err := net.Listen("unix", "127.0.0.1:0")
	if err == nil {
		c.Close()
		return true
	}
	return false
}

// For debug only, will be removed later.
func interestingGoroutines() (gs []string) {
	buf := make([]byte, 2<<20)
	buf = buf[:runtime.Stack(buf, true)]
	for _, g := range strings.Split(string(buf), "\n\n") {
		sl := strings.SplitN(g, "\n", 2)
		if len(sl) != 2 {
			continue
		}
		stack := strings.TrimSpace(sl[1])
		if stack == "" ||
			strings.Contains(stack, "created by github.com/pingcap/tidb.init") ||
			strings.Contains(stack, "testing.RunTests") ||
			strings.Contains(stack, "check.(*resultTracker).start") ||
			strings.Contains(stack, "check.(*suiteRunner).runFunc") ||
			strings.Contains(stack, "check.(*suiteRunner).parallelRun") ||
			strings.Contains(stack, "localstore.(*dbStore).scheduler") ||
			strings.Contains(stack, "tikv.(*noGCHandler).Start") ||
			strings.Contains(stack, "ddl.(*ddl).start") ||
			strings.Contains(stack, "ddl.(*delRange).startEmulator") ||
			strings.Contains(stack, "domain.NewDomain") ||
			strings.Contains(stack, "testing.(*T).Run") ||
			strings.Contains(stack, "domain.(*Domain).LoadPrivilegeLoop") ||
			strings.Contains(stack, "domain.(*Domain).UpdateTableStatsLoop") ||
			strings.Contains(stack, "testing.Main(") ||
			strings.Contains(stack, "runtime.goexit") ||
			strings.Contains(stack, "created by runtime.gc") ||
			strings.Contains(stack, "interestingGoroutines") ||
			strings.Contains(stack, "runtime.MHeap_Scavenger") ||
			// these go routines are async terminated, so they may still alive after test end, thus cause
			// false positive leak failures
			strings.Contains(stack, "google.golang.org/grpc.(*addrConn).resetTransport") ||
			strings.Contains(stack, "google.golang.org/grpc.(*ccBalancerWrapper).watcher") ||
			strings.Contains(stack, "github.com/pingcap/goleveldb/leveldb/util.(*BufferPool).drain") ||
			strings.Contains(stack, "github.com/pingcap/goleveldb/leveldb.(*DB).compactionError") ||
			strings.Contains(stack, "github.com/pingcap/goleveldb/leveldb.(*DB).mpoolDrain") {
			continue
		}
		gs = append(gs, stack)
	}
	sort.Strings(gs)
	return
}

func TestInfo(t *testing.T) {
	if !unixSocketAvailable() {
		return
	}
	for _, str := range interestingGoroutines() {
		t.Logf("TestInfo: BeforeTest %s", str)
	}
	testleak.BeforeTest()
	defer func() {
		for _, str := range interestingGoroutines() {
			t.Logf("TestInfo: AfterTest %s", str)
		}
		testleak.AfterTestT(t)()
	}()
	ddlLease := 80 * time.Millisecond
	s, err := mockstore.NewMockStore()
	if err != nil {
		t.Fatal(err)
	}
	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
	defer clus.Terminate(t)
	mockStore := &mockEtcdBackend{
		Storage: s,
		pdAddrs: []string{clus.Members[0].GRPCAddr()}}
	dom := NewDomain(mockStore, ddlLease, 0, mockFactory)
	defer func() {
		dom.Close()
		s.Close()
	}()

	cli := clus.RandClient()
	dom.etcdClient = cli
	// Mock new DDL and init the schema syncer with etcd client.
	goCtx := context.Background()
	dom.ddl = ddl.NewDDL(
		goCtx,
		ddl.WithEtcdClient(dom.GetEtcdClient()),
		ddl.WithStore(s),
		ddl.WithInfoHandle(dom.infoHandle),<commit_msg>domain: print some logs to investigate the leak (#17503)<commit_after>// Copyright 2015 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package domain

import (
	"context"
	"crypto/tls"
	"math"
	"net"
	"runtime"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/ngaut/pools"
	. "github.com/pingcap/check"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/parser/ast"
	"github.com/pingcap/parser/model"
	"github.com/pingcap/tidb/ddl"
	"github.com/pingcap/tidb/domain/infosync"
	"github.com/pingcap/tidb/errno"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/meta"
	"github.com/pingcap/tidb/metrics"
	"github.com/pingcap/tidb/sessionctx/variable"
	"github.com/pingcap/tidb/store/mockstore"
	"github.com/pingcap/tidb/store/tikv/oracle"
	"github.com/pingcap/tidb/util"
	"github.com/pingcap/tidb/util/logutil"
	"github.com/pingcap/tidb/util/mock"
	"github.com/pingcap/tidb/util/testleak"
	dto "github.com/prometheus/client_model/go"
	"go.etcd.io/etcd/integration"
	"go.uber.org/zap"
)

func TestT(t *testing.T) {
	CustomVerboseFlag = true
	TestingT(t)
}

var _ = Suite(&testSuite{})

type testSuite struct {
}

func mockFactory() (pools.Resource, error) {
	return nil, errors.New("mock factory should not be called")
}

func sysMockFactory(dom *Domain) (pools.Resource, error) {
	return nil, nil
}

type mockEtcdBackend struct {
	kv.Storage
	pdAddrs []string
}

func (mebd *mockEtcdBackend) EtcdAddrs() []string {
	return mebd.pdAddrs
}
func (mebd *mockEtcdBackend) TLSConfig() *tls.Config { return nil }
func (mebd *mockEtcdBackend) StartGCWorker() error {
	panic("not implemented")
}

// ETCD use ip:port as unix socket address, however this address is invalid on windows.
// We have to skip some of the test in such case.
// https://github.com/etcd-io/etcd/blob/f0faa5501d936cd8c9f561bb9d1baca70eb67ab1/pkg/types/urls.go#L42
func unixSocketAvailable() bool {
	c, err := net.Listen("unix", "127.0.0.1:0")
	if err == nil {
		c.Close()
		return true
	}
	return false
}

// For debug only, will be removed later.
// TODO: remove these tests for investigating the leak problem.
func interestingGoroutines() (gs []string) {
	buf := make([]byte, 2<<20)
	buf = buf[:runtime.Stack(buf, true)]
	for _, g := range strings.Split(string(buf), "\n\n") {
		sl := strings.SplitN(g, "\n", 2)
		if len(sl) != 2 {
			continue
		}
		stack := strings.TrimSpace(sl[1])
		if stack == "" ||
			strings.Contains(stack, "created by github.com/pingcap/tidb.init") ||
			strings.Contains(stack, "testing.RunTests") ||
			strings.Contains(stack, "check.(*resultTracker).start") ||
			strings.Contains(stack, "check.(*suiteRunner).runFunc") ||
			strings.Contains(stack, "check.(*suiteRunner).parallelRun") ||
			strings.Contains(stack, "localstore.(*dbStore).scheduler") ||
			strings.Contains(stack, "tikv.(*noGCHandler).Start") ||
			strings.Contains(stack, "ddl.(*ddl).start") ||
			strings.Contains(stack, "ddl.(*delRange).startEmulator") ||
			strings.Contains(stack, "domain.NewDomain") ||
			strings.Contains(stack, "testing.(*T).Run") ||
			strings.Contains(stack, "domain.(*Domain).LoadPrivilegeLoop") ||
			strings.Contains(stack, "domain.(*Domain).UpdateTableStatsLoop") ||
			strings.Contains(stack, "testing.Main(") ||
			strings.Contains(stack, "runtime.goexit") ||
			strings.Contains(stack, "created by runtime.gc") ||
			strings.Contains(stack, "interestingGoroutines") ||
			strings.Contains(stack, "runtime.MHeap_Scavenger") ||
			// these go routines are async terminated, so they may still alive after test end, thus cause
			// false positive leak failures
			strings.Contains(stack, "google.golang.org/grpc.(*addrConn).resetTransport") ||
			strings.Contains(stack, "google.golang.org/grpc.(*ccBalancerWrapper).watcher") ||
			strings.Contains(stack, "github.com/pingcap/goleveldb/leveldb/util.(*BufferPool).drain") ||
			strings.Contains(stack, "github.com/pingcap/goleveldb/leveldb.(*DB).compactionError") ||
			strings.Contains(stack, "github.com/pingcap/goleveldb/leveldb.(*DB).mpoolDrain") {
			continue
		}
		gs = append(gs, stack)
	}
	sort.Strings(gs)
	return
}

var beforeTestGoroutines = map[string]bool{}
var testGoroutinesInited bool

// BeforeTest gets the current goroutines.
// It's used for check.Suite.SetUpSuite() function.
// Now it's only used in the tidb_test.go.
func BeforeTest() {
	for _, g := range interestingGoroutines() {
		beforeTestGoroutines[g] = true
	}
	testGoroutinesInited = true
}

const defaultCheckCnt = 50

func checkLeakAfterTest(errorFunc func(cnt int, g string)) func() {
	// After `BeforeTest`, `beforeTestGoroutines` may still be empty, in this case,
	// we shouldn't init it again.
	if !testGoroutinesInited && len(beforeTestGoroutines) == 0 {
		for _, g := range interestingGoroutines() {
			beforeTestGoroutines[g] = true
		}
	}

	for str := range beforeTestGoroutines {
		logutil.BgLogger().Info("checkLeakAfterTest", zap.String("L163", str))
	}

	cnt := defaultCheckCnt
	return func() {
		defer func() {
			beforeTestGoroutines = map[string]bool{}
			testGoroutinesInited = false
		}()

		var leaked []string
		for i := 0; i < cnt; i++ {
			leaked = leaked[:0]
			for _, g := range interestingGoroutines() {
				if !beforeTestGoroutines[g] {
					leaked = append(leaked, g)
					for str := range beforeTestGoroutines {
						logutil.BgLogger().Info("checkLeakAfterTest", zap.String("L180", str), zap.Int("i", i))
					}
					for _, str := range interestingGoroutines() {
						logutil.BgLogger().Info("checkLeakAfterTest", zap.String("L183", str), zap.Int("i", i))
					}
				}
			}
			// Bad stuff found, but goroutines might just still be
			// shutting down, so give it some time.
			if len(leaked) != 0 {
				time.Sleep(50 * time.Millisecond)
				continue
			}

			return
		}
		for _, g := range leaked {
			errorFunc(cnt, g)
		}
	}
}

// AfterTestT is used after all the test cases is finished.
func AfterTestT(t *testing.T) func() {
	errorFunc := func(cnt int, g string) {
		t.Errorf("Test %s check-count %d appears to have leaked: %v", t.Name(), cnt, g)
	}
	return checkLeakAfterTest(errorFunc)
}

func TestInfo(t *testing.T) {
	if !unixSocketAvailable() {
		return
	}
	for i, str := range interestingGoroutines() {
		t.Logf("TestInfo: BeforeTest %d %s", i, str)
	}
	BeforeTest()
	defer func() {
		AfterTestT(t)()
		for i, str := range interestingGoroutines() {
			t.Logf("TestInfo: AfterTest %d %s", i, str)
		}
	}()
	ddlLease := 80 * time.Millisecond
	s, err := mockstore.NewMockStore()
	if err != nil {
		t.Fatal(err)
	}
	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
	defer clus.Terminate(t)
	mockStore := &mockEtcdBackend{
		Storage: s,
		pdAddrs: []string{clus.Members[0].GRPCAddr()}}
	dom := NewDomain(mockStore, ddlLease, 0, mockFactory)
	defer func() {
		dom.Close()
		s.Close()
	}()

	cli := clus.RandClient()
	dom.etcdClient = cli
	// Mock new DDL and init the schema syncer with etcd client.
	goCtx := context.Background()
	dom.ddl = ddl.NewDDL(
		goCtx,
		ddl.WithEtcdClient(dom.GetEtcdClient()),
		ddl.WithStore(s),
		ddl.WithInfoHandle(dom.infoHandle),
{'fraction_non_alphanumeric': 0.15439981425586255, 'fraction_numerical': 0.009287206872533085, 'mean_word_length': 4.216794509487283, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '730129', 'n_tokens_mistral': 4973, 'n_tokens_neox': 4631, 'n_words': 997}