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.
180 lines
4.7 KiB
180 lines
4.7 KiB
// Copyright 2017 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 tikv
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/pingcap/errors"
|
|
"github.com/pingcap/tidb/util/logutil"
|
|
"go.etcd.io/etcd/clientv3"
|
|
"go.etcd.io/etcd/mvcc/mvccpb"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
// Safe point constants.
|
|
const (
|
|
// This is almost the same as 'tikv_gc_safe_point' in the table 'mysql.tidb',
|
|
// save this to pd instead of tikv, because we can't use interface of table
|
|
// if the safepoint on tidb is expired.
|
|
GcSavedSafePoint = "/tidb/store/gcworker/saved_safe_point"
|
|
|
|
GcSafePointCacheInterval = time.Second * 100
|
|
gcCPUTimeInaccuracyBound = time.Second
|
|
gcSafePointUpdateInterval = time.Second * 10
|
|
gcSafePointQuickRepeatInterval = time.Second
|
|
)
|
|
|
|
// SafePointKV is used for a seamingless integration for mockTest and runtime.
|
|
type SafePointKV interface {
|
|
Put(k string, v string) error
|
|
Get(k string) (string, error)
|
|
GetWithPrefix(k string) ([]*mvccpb.KeyValue, error)
|
|
Close() error
|
|
}
|
|
|
|
// MockSafePointKV implements SafePointKV at mock test
|
|
type MockSafePointKV struct {
|
|
store map[string]string
|
|
mockLock sync.RWMutex
|
|
}
|
|
|
|
// NewMockSafePointKV creates an instance of MockSafePointKV
|
|
func NewMockSafePointKV() *MockSafePointKV {
|
|
return &MockSafePointKV{
|
|
store: make(map[string]string),
|
|
}
|
|
}
|
|
|
|
// Put implements the Put method for SafePointKV
|
|
func (w *MockSafePointKV) Put(k string, v string) error {
|
|
w.mockLock.Lock()
|
|
defer w.mockLock.Unlock()
|
|
w.store[k] = v
|
|
return nil
|
|
}
|
|
|
|
// Get implements the Get method for SafePointKV
|
|
func (w *MockSafePointKV) Get(k string) (string, error) {
|
|
w.mockLock.RLock()
|
|
defer w.mockLock.RUnlock()
|
|
elem := w.store[k]
|
|
return elem, nil
|
|
}
|
|
|
|
// GetWithPrefix implements the Get method for SafePointKV
|
|
func (w *MockSafePointKV) GetWithPrefix(prefix string) ([]*mvccpb.KeyValue, error) {
|
|
w.mockLock.RLock()
|
|
defer w.mockLock.RUnlock()
|
|
kvs := make([]*mvccpb.KeyValue, 0, len(w.store))
|
|
for k, v := range w.store {
|
|
if strings.HasPrefix(k, prefix) {
|
|
kvs = append(kvs, &mvccpb.KeyValue{Key: []byte(k), Value: []byte(v)})
|
|
}
|
|
}
|
|
return kvs, nil
|
|
}
|
|
|
|
// Close implements the Close method for SafePointKV
|
|
func (w *MockSafePointKV) Close() error {
|
|
return nil
|
|
}
|
|
|
|
// EtcdSafePointKV implements SafePointKV at runtime
|
|
type EtcdSafePointKV struct {
|
|
cli *clientv3.Client
|
|
}
|
|
|
|
// NewEtcdSafePointKV creates an instance of EtcdSafePointKV
|
|
func NewEtcdSafePointKV(addrs []string, tlsConfig *tls.Config) (*EtcdSafePointKV, error) {
|
|
etcdCli, err := createEtcdKV(addrs, tlsConfig)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
return &EtcdSafePointKV{cli: etcdCli}, nil
|
|
}
|
|
|
|
// Put implements the Put method for SafePointKV
|
|
func (w *EtcdSafePointKV) Put(k string, v string) error {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
|
|
_, err := w.cli.Put(ctx, k, v)
|
|
cancel()
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Get implements the Get method for SafePointKV
|
|
func (w *EtcdSafePointKV) Get(k string) (string, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
|
|
resp, err := w.cli.Get(ctx, k)
|
|
cancel()
|
|
if err != nil {
|
|
return "", errors.Trace(err)
|
|
}
|
|
if len(resp.Kvs) > 0 {
|
|
return string(resp.Kvs[0].Value), nil
|
|
}
|
|
return "", nil
|
|
}
|
|
|
|
// GetWithPrefix implements the GetWithPrefix for SafePointKV
|
|
func (w *EtcdSafePointKV) GetWithPrefix(k string) ([]*mvccpb.KeyValue, error) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second*15)
|
|
resp, err := w.cli.Get(ctx, k, clientv3.WithPrefix())
|
|
cancel()
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
return resp.Kvs, nil
|
|
}
|
|
|
|
// Close implements the Close for SafePointKV
|
|
func (w *EtcdSafePointKV) Close() error {
|
|
return errors.Trace(w.cli.Close())
|
|
}
|
|
|
|
func saveSafePoint(kv SafePointKV, t uint64) error {
|
|
s := strconv.FormatUint(t, 10)
|
|
err := kv.Put(GcSavedSafePoint, s)
|
|
if err != nil {
|
|
logutil.BgLogger().Error("save safepoint failed", zap.Error(err))
|
|
return errors.Trace(err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func loadSafePoint(kv SafePointKV) (uint64, error) {
|
|
str, err := kv.Get(GcSavedSafePoint)
|
|
|
|
if err != nil {
|
|
return 0, errors.Trace(err)
|
|
}
|
|
|
|
if str == "" {
|
|
return 0, nil
|
|
}
|
|
|
|
t, err := strconv.ParseUint(str, 10, 64)
|
|
if err != nil {
|
|
return 0, errors.Trace(err)
|
|
}
|
|
return t, nil
|
|
}
|
|
|