Skip to content
This repository has been archived by the owner on Dec 8, 2021. It is now read-only.

restore: add importing progress and optimize the accuracy of restore progress #506

Merged
merged 9 commits into from
Jan 27, 2021
Merged

restore: add importing progress and optimize the accuracy of restore progress #506

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b4c2e65
add importing progress and more accurate restore progress
glorv Dec 3, 2020
a937eb7
fix
glorv Dec 3, 2020
c2f60f6
add total progress
glorv Dec 7, 2020
6403c98
Merge branch 'master' into import-progress
glorv Dec 7, 2020
65e0ac2
resolve comments
glorv Jan 6, 2021
ab68717
Merge branch 'import-progress' of https://github.com/pingcap/tidb-lig…
glorv Jan 6, 2021
078d4ea
Merge branch 'master' of https://github.com/pingcap/tidb-lightning in…
glorv Jan 7, 2021
aa6e750
replace TODO with Background context
glorv Jan 27, 2021
82b5b3c
Merge branch 'master' into import-progress
glorv Jan 27, 2021
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
44 changes: 28 additions & 16 deletions lightning/backend/local.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ import (
"github.com/pingcap/tidb-lightning/lightning/common"
"github.com/pingcap/tidb-lightning/lightning/log"
"github.com/pingcap/tidb-lightning/lightning/manual"
"github.com/pingcap/tidb-lightning/lightning/metric"
"github.com/pingcap/tidb-lightning/lightning/worker"
)

Expand Down Expand Up @@ -486,6 +487,11 @@ func (local *local) getImportClient(ctx context.Context, peer *metapb.Peer) (sst
return sst.NewImportSSTClient(conn), nil
}

type rangeStats struct {
count int64
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

seems this member is not used

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, currently only uses the totalBytes

totalBytes int64
}

// WriteToTiKV writer engine key-value pairs to tikv and return the sst meta generated by tikv.
// we don't need to do cleanup for the pairs written to tikv if encounters an error,
// tikv will takes the responsibility to do so.
Expand All @@ -494,18 +500,19 @@ func (local *local) WriteToTiKV(
engineFile *LocalFile,
region *split.RegionInfo,
start, end []byte,
) ([]*sst.SSTMeta, *Range, error) {
) ([]*sst.SSTMeta, *Range, rangeStats, error) {
begin := time.Now()
regionRange := intersectRange(region.Region, Range{start: start, end: end})
opt := &pebble.IterOptions{LowerBound: regionRange.start, UpperBound: regionRange.end}
iter := engineFile.db.NewIter(opt)
defer iter.Close()

stats := rangeStats{}
if !iter.First() {
log.L().Info("keys within region is empty, skip ingest", zap.Binary("start", start),
zap.Binary("regionStart", region.Region.StartKey), zap.Binary("end", end),
zap.Binary("regionEnd", region.Region.EndKey))
return nil, nil, nil
return nil, nil, stats, nil
}

firstKey := codec.EncodeBytes([]byte{}, iter.Key())
Expand All @@ -529,12 +536,12 @@ func (local *local) WriteToTiKV(
for _, peer := range region.Region.GetPeers() {
cli, err := local.getImportClient(ctx, peer)
if err != nil {
return nil, nil, err
return nil, nil, stats, err
}

wstream, err := cli.Write(ctx)
if err != nil {
return nil, nil, errors.Trace(err)
return nil, nil, stats, errors.Trace(err)
}

// Bind uuid for this write request
Expand All @@ -544,7 +551,7 @@ func (local *local) WriteToTiKV(
},
}
if err = wstream.Send(req); err != nil {
return nil, nil, errors.Trace(err)
return nil, nil, stats, errors.Trace(err)
}
req.Chunk = &sst.WriteRequest_Batch{
Batch: &sst.WriteBatch{
Expand All @@ -560,7 +567,7 @@ func (local *local) WriteToTiKV(
pairs := make([]*sst.Pair, 0, local.batchWriteKVPairs)
count := 0
size := int64(0)
totalCount := 0
totalCount := int64(0)
firstLoop := true
regionMaxSize := local.regionSplitSize * 4 / 3

Expand All @@ -584,7 +591,7 @@ func (local *local) WriteToTiKV(
for i := range clients {
requests[i].Chunk.(*sst.WriteRequest_Batch).Batch.Pairs = pairs[:count]
if err := clients[i].Send(requests[i]); err != nil {
return nil, nil, err
return nil, nil, stats, err
}
}
count = 0
Expand All @@ -600,19 +607,19 @@ func (local *local) WriteToTiKV(
for i := range clients {
requests[i].Chunk.(*sst.WriteRequest_Batch).Batch.Pairs = pairs[:count]
if err := clients[i].Send(requests[i]); err != nil {
return nil, nil, err
return nil, nil, stats, err
}
}
}

if iter.Error() != nil {
return nil, nil, errors.Trace(iter.Error())
return nil, nil, stats, errors.Trace(iter.Error())
}

var leaderPeerMetas []*sst.SSTMeta
for i, wStream := range clients {
if resp, closeErr := wStream.CloseAndRecv(); closeErr != nil {
return nil, nil, closeErr
return nil, nil, stats, closeErr
} else {
if leaderID == region.Region.Peers[i].GetId() {
leaderPeerMetas = resp.Metas
Expand All @@ -625,28 +632,30 @@ func (local *local) WriteToTiKV(
if leaderPeerMetas == nil {
log.L().Warn("write to tikv no leader", zap.Reflect("region", region),
zap.Uint64("leader_id", leaderID), zap.Reflect("meta", meta),
zap.Int("kv_pairs", totalCount), zap.Int64("total_bytes", size))
return nil, nil, errors.Errorf("write to tikv with no leader returned, region '%d', leader: %d",
zap.Int64("kv_pairs", totalCount), zap.Int64("total_bytes", size))
return nil, nil, stats, errors.Errorf("write to tikv with no leader returned, region '%d', leader: %d",
region.Region.Id, leaderID)
}

log.L().Debug("write to kv", zap.Reflect("region", region), zap.Uint64("leader", leaderID),
zap.Reflect("meta", meta), zap.Reflect("return metas", leaderPeerMetas),
zap.Int("kv_pairs", totalCount), zap.Int64("total_bytes", size),
zap.Int64("kv_pairs", totalCount), zap.Int64("total_bytes", size),
zap.Int64("buf_size", bytesBuf.totalSize()),
zap.Stringer("takeTime", time.Since(begin)))

var remainRange *Range
if iter.Valid() && iter.Next() {
firstKey := append([]byte{}, iter.Key()...)
remainRange = &Range{start: firstKey, end: regionRange.end}
log.L().Info("write to tikv partial finish", zap.Int("count", totalCount),
log.L().Info("write to tikv partial finish", zap.Int64("count", totalCount),
zap.Int64("size", size), zap.Binary("startKey", regionRange.start), zap.Binary("endKey", regionRange.end),
zap.Binary("remainStart", remainRange.start), zap.Binary("remainEnd", remainRange.end),
zap.Reflect("region", region))
}
stats.count = totalCount
stats.totalBytes = size

return leaderPeerMetas, remainRange, nil
return leaderPeerMetas, remainRange, stats, nil
}

func (local *local) Ingest(ctx context.Context, meta *sst.SSTMeta, region *split.RegionInfo) (*sst.IngestResponse, error) {
Expand Down Expand Up @@ -935,10 +944,11 @@ func (local *local) writeAndIngestPairs(
) (*Range, error) {
var err error
var remainRange *Range
var rangeStats rangeStats
loopWrite:
for i := 0; i < maxRetryTimes; i++ {
var metas []*sst.SSTMeta
metas, remainRange, err = local.WriteToTiKV(ctx, engineFile, region, start, end)
metas, remainRange, rangeStats, err = local.WriteToTiKV(ctx, engineFile, region, start, end)
if err != nil {
log.L().Warn("write to tikv failed", log.ShortError(err))
return nil, err
Expand Down Expand Up @@ -992,6 +1002,8 @@ loopWrite:
if err != nil {
log.L().Warn("write and ingest region, will retry import full range", log.ShortError(err),
zap.Stringer("region", region.Region), zap.Binary("start", start), zap.Binary("end", end))
} else {
metric.BytesCounter.WithLabelValues(metric.TableStateImported).Add(float64(rangeStats.totalBytes))
}
return remainRange, errors.Trace(err)
}
Expand Down
7 changes: 7 additions & 0 deletions lightning/metric/metric.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,12 @@ var (
Name: "chunks",
Help: "count number of chunks processed",
}, []string{"state"})
BytesCounter = prometheus.NewCounterVec(
prometheus.CounterOpts{
Namespace: "lightning",
Name: "bytes",
Help: "count of total bytes",
}, []string{"state"})
// state can be one of:
// - estimated (an estimation derived from the file size)
// - pending
Expand Down Expand Up @@ -190,6 +196,7 @@ func init() {
prometheus.MustRegister(TableCounter)
prometheus.MustRegister(ProcessedEngineCounter)
prometheus.MustRegister(ChunkCounter)
prometheus.MustRegister(BytesCounter)
prometheus.MustRegister(ImportSecondsHistogram)
prometheus.MustRegister(RowReadSecondsHistogram)
prometheus.MustRegister(RowReadBytesHistogram)
Expand Down
Loading