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

// 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
}