You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

464 lines
14 KiB

// 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"
"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/parser/terror"
"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, error) {
return mebd.pdAddrs, nil
}
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
}
func TestInfo(t *testing.T) {
if !unixSocketAvailable() {
return
}
testleak.BeforeTest()
defer testleak.AfterTestT(t)()
ddlLease := 80 * time.Millisecond
s, err := mockstore.NewMockTikvStore()
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),
ddl.WithLease(ddlLease),
)
err = dom.ddl.Start(nil)
if err != nil {
t.Fatal(err)
}
err = failpoint.Enable("github.com/pingcap/tidb/domain/MockReplaceDDL", `return(true)`)
if err != nil {
t.Fatal(err)
}
err = dom.Init(ddlLease, sysMockFactory)
if err != nil {
t.Fatal(err)
}
err = failpoint.Disable("github.com/pingcap/tidb/domain/MockReplaceDDL")
if err != nil {
t.Fatal(err)
}
// Test for GetServerInfo and GetServerInfoByID.
ddlID := dom.ddl.GetID()
serverInfo, err := infosync.GetServerInfo()
if err != nil {
t.Fatal(err)
}
info, err := infosync.GetServerInfoByID(goCtx, ddlID)
if err != nil {
t.Fatal(err)
}
if serverInfo.ID != info.ID {
t.Fatalf("server self info %v, info %v", serverInfo, info)
}
_, err = infosync.GetServerInfoByID(goCtx, "not_exist_id")
if err == nil || (err != nil && err.Error() != "[info-syncer] get /tidb/server/info/not_exist_id failed") {
t.Fatal(err)
}
// Test for GetAllServerInfo.
infos, err := infosync.GetAllServerInfo(goCtx)
if err != nil {
t.Fatal(err)
}
if len(infos) != 1 || infos[ddlID].ID != info.ID {
t.Fatalf("server one info %v, info %v", infos[ddlID], info)
}
// Test the scene where syncer.Done() gets the information.
err = failpoint.Enable("github.com/pingcap/tidb/ddl/util/ErrorMockSessionDone", `return(true)`)
if err != nil {
t.Fatal(err)
}
<-dom.ddl.SchemaSyncer().Done()
err = failpoint.Disable("github.com/pingcap/tidb/ddl/util/ErrorMockSessionDone")
if err != nil {
t.Fatal(err)
}
time.Sleep(15 * time.Millisecond)
syncerStarted := false
for i := 0; i < 1000; i++ {
if dom.SchemaValidator.IsStarted() {
syncerStarted = true
break
}
time.Sleep(5 * time.Millisecond)
}
if !syncerStarted {
t.Fatal("start syncer failed")
}
// Make sure loading schema is normal.
cs := &ast.CharsetOpt{
Chs: "utf8",
Col: "utf8_bin",
}
ctx := mock.NewContext()
err = dom.ddl.CreateSchema(ctx, model.NewCIStr("aaa"), cs)
if err != nil {
t.Fatal(err)
}
err = dom.Reload()
if err != nil {
t.Fatal(err)
}
if dom.InfoSchema().SchemaMetaVersion() != 1 {
t.Fatalf("update schema version failed, ver %d", dom.InfoSchema().SchemaMetaVersion())
}
// Test for RemoveServerInfo.
dom.info.RemoveServerInfo()
infos, err = infosync.GetAllServerInfo(goCtx)
if err != nil || len(infos) != 0 {
t.Fatalf("err %v, infos %v", err, infos)
}
}
type mockSessionManager struct {
PS []*util.ProcessInfo
}
func (msm *mockSessionManager) ShowProcessList() map[uint64]*util.ProcessInfo {
ret := make(map[uint64]*util.ProcessInfo)
for _, item := range msm.PS {
ret[item.ID] = item
}
return ret
}
func (msm *mockSessionManager) GetProcessInfo(id uint64) (*util.ProcessInfo, bool) {
for _, item := range msm.PS {
if item.ID == id {
return item, true
}
}
return &util.ProcessInfo{}, false
}
func (msm *mockSessionManager) Kill(cid uint64, query bool) {}
func (msm *mockSessionManager) UpdateTLSConfig(cfg *tls.Config) {}
func (*testSuite) TestT(c *C) {
defer testleak.AfterTest(c)()
store, err := mockstore.NewMockTikvStore()
c.Assert(err, IsNil)
ddlLease := 80 * time.Millisecond
dom := NewDomain(store, ddlLease, 0, mockFactory)
err = dom.Init(ddlLease, sysMockFactory)
c.Assert(err, IsNil)
ctx := mock.NewContext()
ctx.Store = dom.Store()
dd := dom.DDL()
c.Assert(dd, NotNil)
c.Assert(dd.GetLease(), Equals, 80*time.Millisecond)
snapTS := oracle.EncodeTSO(oracle.GetPhysical(time.Now()))
cs := &ast.CharsetOpt{
Chs: "utf8",
Col: "utf8_bin",
}
err = dd.CreateSchema(ctx, model.NewCIStr("aaa"), cs)
c.Assert(err, IsNil)
// Test for fetchSchemasWithTables when "tables" isn't nil.
err = dd.CreateTable(ctx, &ast.CreateTableStmt{Table: &ast.TableName{
Schema: model.NewCIStr("aaa"),
Name: model.NewCIStr("tbl")}})
c.Assert(err, IsNil)
is := dom.InfoSchema()
c.Assert(is, NotNil)
// for updating the self schema version
goCtx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
err = dd.SchemaSyncer().OwnerCheckAllVersions(goCtx, is.SchemaMetaVersion())
cancel()
c.Assert(err, IsNil)
snapIs, err := dom.GetSnapshotInfoSchema(snapTS)
c.Assert(snapIs, NotNil)
c.Assert(err, IsNil)
// Make sure that the self schema version doesn't be changed.
goCtx, cancel = context.WithTimeout(context.Background(), 100*time.Millisecond)
err = dd.SchemaSyncer().OwnerCheckAllVersions(goCtx, is.SchemaMetaVersion())
cancel()
c.Assert(err, IsNil)
// for GetSnapshotInfoSchema
currSnapTS := oracle.EncodeTSO(oracle.GetPhysical(time.Now()))
currSnapIs, err := dom.GetSnapshotInfoSchema(currSnapTS)
c.Assert(err, IsNil)
c.Assert(currSnapIs, NotNil)
c.Assert(currSnapIs.SchemaMetaVersion(), Equals, is.SchemaMetaVersion())
// for GetSnapshotMeta
dbInfo, ok := currSnapIs.SchemaByName(model.NewCIStr("aaa"))
c.Assert(ok, IsTrue)
tbl, err := currSnapIs.TableByName(model.NewCIStr("aaa"), model.NewCIStr("tbl"))
c.Assert(err, IsNil)
m, err := dom.GetSnapshotMeta(snapTS)
c.Assert(err, IsNil)
tblInfo1, err := m.GetTable(dbInfo.ID, tbl.Meta().ID)
c.Assert(meta.ErrDBNotExists.Equal(err), IsTrue)
c.Assert(tblInfo1, IsNil)
m, err = dom.GetSnapshotMeta(currSnapTS)
c.Assert(err, IsNil)
tblInfo2, err := m.GetTable(dbInfo.ID, tbl.Meta().ID)
c.Assert(err, IsNil)
c.Assert(tbl.Meta(), DeepEquals, tblInfo2)
// Test for tryLoadSchemaDiffs when "isTooOldSchema" is false.
err = dd.CreateSchema(ctx, model.NewCIStr("bbb"), cs)
c.Assert(err, IsNil)
err = dom.Reload()
c.Assert(err, IsNil)
// for schemaValidator
schemaVer := dom.SchemaValidator.(*schemaValidator).LatestSchemaVersion()
ver, err := store.CurrentVersion()
c.Assert(err, IsNil)
ts := ver.Ver
_, succ := dom.SchemaValidator.Check(ts, schemaVer, nil)
c.Assert(succ, Equals, ResultSucc)
c.Assert(failpoint.Enable("github.com/pingcap/tidb/domain/ErrorMockReloadFailed", `return(true)`), IsNil)
err = dom.Reload()
c.Assert(err, NotNil)
_, succ = dom.SchemaValidator.Check(ts, schemaVer, nil)
c.Assert(succ, Equals, ResultSucc)
time.Sleep(ddlLease)
ver, err = store.CurrentVersion()
c.Assert(err, IsNil)
ts = ver.Ver
_, succ = dom.SchemaValidator.Check(ts, schemaVer, nil)
c.Assert(succ, Equals, ResultUnknown)
c.Assert(failpoint.Disable("github.com/pingcap/tidb/domain/ErrorMockReloadFailed"), IsNil)
err = dom.Reload()
c.Assert(err, IsNil)
_, succ = dom.SchemaValidator.Check(ts, schemaVer, nil)
c.Assert(succ, Equals, ResultSucc)
// For slow query.
dom.LogSlowQuery(&SlowQueryInfo{SQL: "aaa", Duration: time.Second, Internal: true})
dom.LogSlowQuery(&SlowQueryInfo{SQL: "bbb", Duration: 3 * time.Second})
dom.LogSlowQuery(&SlowQueryInfo{SQL: "ccc", Duration: 2 * time.Second})
// Collecting slow queries is asynchronous, wait a while to ensure it's done.
time.Sleep(5 * time.Millisecond)
res := dom.ShowSlowQuery(&ast.ShowSlow{Tp: ast.ShowSlowTop, Count: 2})
c.Assert(res, HasLen, 2)
c.Assert(res[0].SQL, Equals, "bbb")
c.Assert(res[0].Duration, Equals, 3*time.Second)
c.Assert(res[1].SQL, Equals, "ccc")
c.Assert(res[1].Duration, Equals, 2*time.Second)
res = dom.ShowSlowQuery(&ast.ShowSlow{Tp: ast.ShowSlowTop, Count: 2, Kind: ast.ShowSlowKindInternal})
c.Assert(res, HasLen, 1)
c.Assert(res[0].SQL, Equals, "aaa")
c.Assert(res[0].Duration, Equals, time.Second)
c.Assert(res[0].Internal, Equals, true)
res = dom.ShowSlowQuery(&ast.ShowSlow{Tp: ast.ShowSlowTop, Count: 4, Kind: ast.ShowSlowKindAll})
c.Assert(res, HasLen, 3)
c.Assert(res[0].SQL, Equals, "bbb")
c.Assert(res[0].Duration, Equals, 3*time.Second)
c.Assert(res[1].SQL, Equals, "ccc")
c.Assert(res[1].Duration, Equals, 2*time.Second)
c.Assert(res[2].SQL, Equals, "aaa")
c.Assert(res[2].Duration, Equals, time.Second)
c.Assert(res[2].Internal, Equals, true)
res = dom.ShowSlowQuery(&ast.ShowSlow{Tp: ast.ShowSlowRecent, Count: 2})
c.Assert(res, HasLen, 2)
c.Assert(res[0].SQL, Equals, "ccc")
c.Assert(res[0].Duration, Equals, 2*time.Second)
c.Assert(res[1].SQL, Equals, "bbb")
c.Assert(res[1].Duration, Equals, 3*time.Second)
metrics.PanicCounter.Reset()
// Since the stats lease is 0 now, so create a new ticker will panic.
// Test that they can recover from panic correctly.
dom.updateStatsWorker(ctx, nil)
dom.autoAnalyzeWorker(nil)
counter := metrics.PanicCounter.WithLabelValues(metrics.LabelDomain)
pb := &dto.Metric{}
counter.Write(pb)
c.Assert(pb.GetCounter().GetValue(), Equals, float64(2))
scope := dom.GetScope("status")
c.Assert(scope, Equals, variable.DefaultStatusVarScopeFlag)
// For schema check, it tests for getting the result of "ResultUnknown".
schemaChecker := NewSchemaChecker(dom, is.SchemaMetaVersion(), nil)
originalRetryTime := SchemaOutOfDateRetryTimes
originalRetryInterval := SchemaOutOfDateRetryInterval
// Make sure it will retry one time and doesn't take a long time.
SchemaOutOfDateRetryTimes = 1
SchemaOutOfDateRetryInterval = int64(time.Millisecond * 1)
defer func() {
SchemaOutOfDateRetryTimes = originalRetryTime
SchemaOutOfDateRetryInterval = originalRetryInterval
}()
dom.SchemaValidator.Stop()
_, err = schemaChecker.Check(uint64(123456))
c.Assert(err.Error(), Equals, ErrInfoSchemaExpired.Error())
dom.SchemaValidator.Reset()
// Test for reporting min start timestamp.
infoSyncer := dom.InfoSyncer()
sm := &mockSessionManager{
PS: make([]*util.ProcessInfo, 0),
}
infoSyncer.SetSessionManager(sm)
beforeTS := variable.GoTimeToTS(time.Now())
infoSyncer.ReportMinStartTS(dom.Store())
afterTS := variable.GoTimeToTS(time.Now())
c.Assert(infoSyncer.GetMinStartTS() > beforeTS && infoSyncer.GetMinStartTS() < afterTS, IsFalse)
lowerLimit := time.Now().Add(-time.Duration(kv.MaxTxnTimeUse) * time.Millisecond)
validTS := variable.GoTimeToTS(lowerLimit.Add(time.Minute))
sm.PS = []*util.ProcessInfo{
{CurTxnStartTS: 0},
{CurTxnStartTS: math.MaxUint64},
{CurTxnStartTS: variable.GoTimeToTS(lowerLimit)},
{CurTxnStartTS: validTS},
}
infoSyncer.SetSessionManager(sm)
infoSyncer.ReportMinStartTS(dom.Store())
c.Assert(infoSyncer.GetMinStartTS() == validTS, IsTrue)
err = store.Close()
c.Assert(err, IsNil)
isClose := dom.isClose()
c.Assert(isClose, IsFalse)
dom.Close()
isClose = dom.isClose()
c.Assert(isClose, IsTrue)
}
type testResource struct {
status int
}
func (tr *testResource) Close() { tr.status = 1 }
func (*testSuite) TestSessionPool(c *C) {
f := func() (pools.Resource, error) { return &testResource{}, nil }
pool := newSessionPool(1, f)
tr, err := pool.Get()
c.Assert(err, IsNil)
tr1, err := pool.Get()
c.Assert(err, IsNil)
pool.Put(tr)
// Capacity is 1, so tr1 is closed.
pool.Put(tr1)
c.Assert(tr1.(*testResource).status, Equals, 1)
pool.Close()
pool.Close()
pool.Put(tr1)
tr, err = pool.Get()
c.Assert(err.Error(), Equals, "session pool closed")
c.Assert(tr, IsNil)
}
func (*testSuite) TestErrorCode(c *C) {
c.Assert(int(terror.ToSQLError(ErrInfoSchemaExpired).Code), Equals, errno.ErrInfoSchemaExpired)
c.Assert(int(terror.ToSQLError(ErrInfoSchemaChanged).Code), Equals, errno.ErrInfoSchemaChanged)
}