Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

br-stream: add precheck before create task #33088

Merged
merged 3 commits into from
Mar 17, 2022
Merged

br-stream: add precheck before create task #33088

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
34db9bd
add switch check before stream starts
3pointer Mar 15, 2022
0b7999a
add unit tests
3pointer Mar 15, 2022
be18824
fix lint
3pointer Mar 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 8 additions & 17 deletions br/pkg/conn/conn_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,21 +10,12 @@ import (
"github.com/pingcap/failpoint"
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/pingcap/tidb/br/pkg/pdutil"
"github.com/pingcap/tidb/br/pkg/utils"
"github.com/stretchr/testify/require"
pd "github.com/tikv/pd/client"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

type fakePDClient struct {
pd.Client
stores []*metapb.Store
}

func (c fakePDClient) GetAllStores(context.Context, ...pd.GetStoreOption) ([]*metapb.Store, error) {
return append([]*metapb.Store{}, c.stores...), nil
}

func TestGetAllTiKVStoresWithRetryCancel(t *testing.T) {
_ = failpoint.Enable("github.com/pingcap/tidb/br/pkg/conn/hint-GetAllTiKVStores-cancel", "return(true)")
defer func() {
Expand Down Expand Up @@ -56,8 +47,8 @@ func TestGetAllTiKVStoresWithRetryCancel(t *testing.T) {
},
}

fpdc := fakePDClient{
stores: stores,
fpdc := utils.FakePDClient{
Stores: stores,
}

_, err := GetAllTiKVStoresWithRetry(ctx, fpdc, SkipTiFlash)
Expand Down Expand Up @@ -96,8 +87,8 @@ func TestGetAllTiKVStoresWithUnknown(t *testing.T) {
},
}

fpdc := fakePDClient{
stores: stores,
fpdc := utils.FakePDClient{
Stores: stores,
}

_, err := GetAllTiKVStoresWithRetry(ctx, fpdc, SkipTiFlash)
Expand Down Expand Up @@ -151,8 +142,8 @@ func TestCheckStoresAlive(t *testing.T) {
},
}

fpdc := fakePDClient{
stores: stores,
fpdc := utils.FakePDClient{
Stores: stores,
}

kvStores, err := GetAllTiKVStoresWithRetry(ctx, fpdc, SkipTiFlash)
Expand Down Expand Up @@ -240,7 +231,7 @@ func TestGetAllTiKVStores(t *testing.T) {
}

for _, testCase := range testCases {
pdClient := fakePDClient{stores: testCase.stores}
pdClient := utils.FakePDClient{Stores: testCase.stores}
stores, err := GetAllTiKVStores(context.Background(), pdClient, testCase.storeBehavior)
if len(testCase.expectedError) != 0 {
require.Error(t, err)
Expand Down
113 changes: 89 additions & 24 deletions br/pkg/task/stream.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
// Copyright 2015 PingCAP, Inc.
// Copyright 2022 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
Expand All @@ -17,17 +17,22 @@ package task
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"sort"
"strings"

"github.com/opentracing/opentracing-go"
"github.com/pingcap/errors"
backuppb "github.com/pingcap/kvproto/pkg/brpb"
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/pingcap/log"
"github.com/pingcap/tidb/br/pkg/backup"
"github.com/pingcap/tidb/br/pkg/conn"
berrors "github.com/pingcap/tidb/br/pkg/errors"
"github.com/pingcap/tidb/br/pkg/glue"
"github.com/pingcap/tidb/br/pkg/httputil"
"github.com/pingcap/tidb/br/pkg/logutil"
"github.com/pingcap/tidb/br/pkg/metautil"
"github.com/pingcap/tidb/br/pkg/restore"
Expand Down Expand Up @@ -88,10 +93,10 @@ type StreamConfig struct {
RestoreTS uint64 `json:"restore-ts" toml:"restore-ts"`
}

func (sc *StreamConfig) adjustRestoreConfig() {
sc.Config.adjust()
if sc.Concurrency == 0 {
sc.Concurrency = 32
func (cfg *StreamConfig) adjustRestoreConfig() {
cfg.Config.adjust()
if cfg.Concurrency == 0 {
cfg.Concurrency = 32
}
}

Expand Down Expand Up @@ -188,12 +193,13 @@ func (cfg *StreamConfig) ParseStreamCommonFromFlags(flags *pflag.FlagSet) error
}

type streamMgr struct {
Cfg *StreamConfig
mgr *conn.Mgr
bc *backup.Client
cfg *StreamConfig
mgr *conn.Mgr
bc *backup.Client
httpCli *http.Client
}

func NewStreamMgr(ctx context.Context, cfg *StreamConfig, g glue.Glue, needStorage bool,
func NewStreamMgr(ctx context.Context, cfg *StreamConfig, g glue.Glue, isStreamStart bool,
) (*streamMgr, error) {
mgr, err := NewMgr(ctx, g, cfg.PD, cfg.TLS, GetKeepalive(&cfg.Config),
cfg.CheckRequirements, true)
Expand All @@ -208,10 +214,10 @@ func NewStreamMgr(ctx context.Context, cfg *StreamConfig, g glue.Glue, needStora

// just stream start need Storage
s := &streamMgr{
Cfg: cfg,
cfg: cfg,
mgr: mgr,
}
if needStorage {
if isStreamStart {
backend, err := storage.ParseBackend(cfg.Storage, &cfg.BackendOptions)
if err != nil {
return nil, errors.Trace(err)
Expand All @@ -230,6 +236,9 @@ func NewStreamMgr(ctx context.Context, cfg *StreamConfig, g glue.Glue, needStora
return nil, errors.Trace(err)
}
s.bc = client

// create http client to do some requirements check.
s.httpCli = httputil.NewClient(mgr.GetTLSConfig())
}
return s, nil
}
Expand All @@ -250,19 +259,19 @@ func (s *streamMgr) adjustAndCheckStartTS(ctx context.Context) error {
return errors.Trace(err)
}
// set currentTS to startTS as a default value
if s.Cfg.StartTS == 0 {
s.Cfg.StartTS = currentTS
if s.cfg.StartTS == 0 {
s.cfg.StartTS = currentTS
}

if currentTS < s.Cfg.StartTS || s.Cfg.EndTS <= currentTS {
if currentTS < s.cfg.StartTS || s.cfg.EndTS <= currentTS {
return errors.Annotatef(berrors.ErrInvalidArgument,
"invalid timestamps, startTS %d should be smaller than currentTS %d",
s.Cfg.StartTS, currentTS)
s.cfg.StartTS, currentTS)
}
if s.Cfg.EndTS <= currentTS {
if s.cfg.EndTS <= currentTS {
return errors.Annotatef(berrors.ErrInvalidArgument,
"invalid timestamps, endTS %d should be larger than currentTS %d",
s.Cfg.EndTS, currentTS)
s.cfg.EndTS, currentTS)
}

return nil
Expand All @@ -275,15 +284,15 @@ func (s *streamMgr) setGCSafePoint(ctx context.Context) error {
return errors.Trace(err)
}

err := utils.CheckGCSafePoint(ctx, s.mgr.GetPDClient(), s.Cfg.StartTS)
err := utils.CheckGCSafePoint(ctx, s.mgr.GetPDClient(), s.cfg.StartTS)
if err != nil {
return errors.Trace(err)
}

sp := utils.BRServiceSafePoint{
ID: utils.MakeSafePointID(),
TTL: s.Cfg.SafePointTTL,
BackupTS: s.Cfg.StartTS,
TTL: s.cfg.SafePointTTL,
BackupTS: s.cfg.StartTS,
}
err = utils.UpdateServiceSafePoint(ctx, s.mgr.GetPDClient(), sp)
if err != nil {
Expand All @@ -306,8 +315,8 @@ func (s *streamMgr) getTS(ctx context.Context) (uint64, error) {
func (s *streamMgr) buildObserveRanges(ctx context.Context) ([]kv.KeyRange, error) {
dRanges, err := stream.BuildObserveDataRanges(
s.mgr.GetStorage(),
s.Cfg.TableFilter,
s.Cfg.StartTS,
s.cfg.TableFilter,
s.cfg.StartTS,
)
if err != nil {
return nil, errors.Trace(err)
Expand All @@ -322,6 +331,52 @@ func (s *streamMgr) buildObserveRanges(ctx context.Context) ([]kv.KeyRange, erro
return rs, nil
}

// checkRequirements will check some requirements before stream starts.
func (s *streamMgr) checkRequirements(ctx context.Context) (bool, error) {
allStores, err := conn.GetAllTiKVStoresWithRetry(ctx, s.mgr.GetPDClient(), conn.SkipTiFlash)
if err != nil {
return false, errors.Trace(err)
}

type backupStream struct {
EnableStreaming bool `json:"enable-streaming"`
}
type config struct {
BackupStream backupStream `json:"backup-stream"`
}

supportBackupStream := true
hasTiKV := false
for _, store := range allStores {
if store.State != metapb.StoreState_Up {
continue
}
hasTiKV = true
// we need make sure every available store support backup-stream otherwise we might lose data.
// so check every store's config
addr := fmt.Sprintf("%s/config", store.GetStatusAddress())
err = utils.WithRetry(ctx, func() error {
resp, e := s.httpCli.Get(addr)
if e != nil {
return e
}
c := &config{}
e = json.NewDecoder(resp.Body).Decode(c)
if e != nil {
return e
}
supportBackupStream = supportBackupStream && c.BackupStream.EnableStreaming
_ = resp.Body.Close()
return nil
}, utils.NewPDReqBackoffer())
if err != nil {
// if one store failed, break and return error
break
}
}
return hasTiKV && supportBackupStream, err
}

// RunStreamCommand run all kinds of `stream task``
func RunStreamCommand(
ctx context.Context,
Expand Down Expand Up @@ -368,10 +423,20 @@ func RunStreamStart(
}
defer streamMgr.close()

if err := streamMgr.setGCSafePoint(ctx); err != nil {
supportStream, err := streamMgr.checkRequirements(ctx)
if err != nil {
return errors.Trace(err)
}
if !supportStream {
return errors.New("Unable to create stream task. " +
"please set tikv config `backup-stream.enable-streaming` to true." +
"and restart tikv")
}

if err = streamMgr.setGCSafePoint(ctx); err != nil {
return errors.Trace(err)
}
if err := streamMgr.setLock(ctx); err != nil {
if err = streamMgr.setLock(ctx); err != nil {
return errors.Trace(err)
}

Expand Down
Loading