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.
 
 

114 lines
3.3 KiB

// Copyright 2020 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 execdetails
import (
"context"
"sync/atomic"
"time"
"github.com/pingcap/kvproto/pkg/metapb"
pd "github.com/tikv/pd/client"
)
var (
_ pd.Client = &InterceptedPDClient{}
_ pd.TSFuture = &interceptedTsFuture{}
)
func recordPDWaitTime(ctx context.Context, start time.Time) {
stmtExec := ctx.Value(StmtExecDetailKey)
if stmtExec != nil {
detail := stmtExec.(*StmtExecDetails)
atomic.AddInt64(&detail.WaitPDRespDuration, int64(time.Since(start)))
}
}
// InterceptedPDClient is a PD's wrapper client to record stmt detail.
type InterceptedPDClient struct {
pd.Client
}
// interceptedTsFuture is a PD's wrapper future to record stmt detail.
type interceptedTsFuture struct {
pd.TSFuture
ctx context.Context
}
// Wait implements pd.Client#Wait.
func (m interceptedTsFuture) Wait() (int64, int64, error) {
start := time.Now()
physical, logical, err := m.TSFuture.Wait()
recordPDWaitTime(m.ctx, start)
return physical, logical, err
}
// GetTS implements pd.Client#GetTS.
func (m InterceptedPDClient) GetTS(ctx context.Context) (int64, int64, error) {
start := time.Now()
physical, logical, err := m.Client.GetTS(ctx)
recordPDWaitTime(ctx, start)
return physical, logical, err
}
// GetTSAsync implements pd.Client#GetTSAsync.
func (m InterceptedPDClient) GetTSAsync(ctx context.Context) pd.TSFuture {
start := time.Now()
f := m.Client.GetTSAsync(ctx)
recordPDWaitTime(ctx, start)
return interceptedTsFuture{
ctx: ctx,
TSFuture: f,
}
}
// GetRegion implements pd.Client#GetRegion.
func (m InterceptedPDClient) GetRegion(ctx context.Context, key []byte) (*pd.Region, error) {
start := time.Now()
r, err := m.Client.GetRegion(ctx, key)
recordPDWaitTime(ctx, start)
return r, err
}
// GetPrevRegion implements pd.Client#GetPrevRegion.
func (m InterceptedPDClient) GetPrevRegion(ctx context.Context, key []byte) (*pd.Region, error) {
start := time.Now()
r, err := m.Client.GetRegion(ctx, key)
recordPDWaitTime(ctx, start)
return r, err
}
// GetRegionByID implements pd.Client#GetRegionByID.
func (m InterceptedPDClient) GetRegionByID(ctx context.Context, regionID uint64) (*pd.Region, error) {
start := time.Now()
r, err := m.Client.GetRegionByID(ctx, regionID)
recordPDWaitTime(ctx, start)
return r, err
}
// ScanRegions implements pd.Client#ScanRegions.
func (m InterceptedPDClient) ScanRegions(ctx context.Context, key, endKey []byte, limit int) ([]*metapb.Region, []*metapb.Peer, error) {
start := time.Now()
r, p, err := m.Client.ScanRegions(ctx, key, endKey, limit)
recordPDWaitTime(ctx, start)
return r, p, err
}
// GetStore implements pd.Client#GetStore.
func (m InterceptedPDClient) GetStore(ctx context.Context, storeID uint64) (*metapb.Store, error) {
start := time.Now()
s, err := m.Client.GetStore(ctx, storeID)
recordPDWaitTime(ctx, start)
return s, err
}