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

Jump to bottom

add logs for cse region client #838

Merged
merged 3 commits into from
Jun 15, 2023
Merged

add logs for cse region client #838

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8417455
add logs for cse region client
iosmanthus Jun 15, 2023
3eb914f
fix store index
iosmanthus Jun 15, 2023
80fb9dd
fix lint
iosmanthus Jun 15, 2023
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
64 changes: 60 additions & 4 deletions internal/locate/cse/cse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"encoding/json"
"fmt"
"io"
"math/rand"
"net/http"
"sync"
"time"
Expand Down Expand Up @@ -38,6 +39,12 @@ var (
ErrRegionNotFound = errors.New("region not found")
)

type ctxKey string

var (
reqIDKey ctxKey = "cse-req-id"
)

type store struct {
*metapb.Store
breaker *asyncBreaker
Expand Down Expand Up @@ -221,8 +228,8 @@ type result struct {
index int
}

func mkOK(ok []*pdcore.RegionInfo) result {
return result{ok: ok}
func mkOK(idx int, ok []*pdcore.RegionInfo) result {
return result{ok: ok, index: idx}
}

func mkErr(idx int, err error) result {
Expand All @@ -233,6 +240,7 @@ func mkErr(idx int, err error) result {
// Then, collect the region results, and merge them into a BTreeMap: RegionsInfo provided by the PD.
// Finally, return the RegionsInfo.
func (c *Client) fanout(ctx context.Context, tag, method, endpoint string, req any) (*pdcore.RegionsInfo, error) {
reqID := getReqID(ctx)
start := time.Now()
defer func() {
metrics.TiKVSyncRegionDuration.
Expand All @@ -241,6 +249,11 @@ func (c *Client) fanout(ctx context.Context, tag, method, endpoint string, req a
}()

stores := c.getAliveTiKVStores()
ss := make([]string, 0, len(stores))
for _, s := range stores {
ss = append(ss, s.GetStatusAddress())
}
log.Info("sync region from", zap.Strings("stores", ss), zap.Uint64("req", reqID))
rchan := make(chan result, len(stores))
for i, s := range stores {
store := *s
Expand Down Expand Up @@ -309,20 +322,31 @@ func (c *Client) fanout(ctx context.Context, tag, method, endpoint string, req a
regionInfos = append(regionInfos, regionInfo)
}
}
rchan <- mkOK(regionInfos)
rchan <- mkOK(index, regionInfos)
})
}

regionsInfo := pdcore.NewRegionsInfo()
for i := 0; i < len(stores); i++ {
select {
case <-ctx.Done():
log.Warn("context canceled when syncing regions from cse",
zap.Error(ctx.Err()), zap.Uint64("req", reqID))
return nil, ctx.Err()
case r := <-rchan:
if err := r.err; err != nil {
log.Warn("failed to sync regions from cse", zap.Error(err), zap.String("store", stores[r.index].StatusAddress))
log.Warn("failed to sync regions from cse",
zap.Error(err),
zap.String("store", stores[r.index].StatusAddress),
zap.Uint64("req", reqID))
}
for _, regionInfo := range r.ok {
log.Info("sync region from cse",
zap.String("type", tag),
zap.String("region", regionInfo.GetMeta().String()),
zap.String("store", stores[r.index].StatusAddress),
zap.Uint64("req", reqID),
)
regionsInfo.CheckAndPutRegion(regionInfo)
}
}
Expand All @@ -347,7 +371,20 @@ func mkPDRegions(regions ...*pdcore.RegionInfo) []*pd.Region {
return rs
}

func withReqID(ctx context.Context) context.Context {
return context.WithValue(ctx, reqIDKey, rand.Uint64())
}

func getReqID(ctx context.Context) uint64 {
if id, ok := ctx.Value(reqIDKey).(uint64); ok {
return id
}
return 0
}

func (c *Client) GetRegion(ctx context.Context, key []byte, _ ...pd.GetRegionOption) (*pd.Region, error) {
ctx = withReqID(ctx)
reqID := getReqID(ctx)
_, start, err := codec.DecodeBytes(key, nil)
if err != nil {
return nil, err
Expand All @@ -362,13 +399,18 @@ func (c *Client) GetRegion(ctx context.Context, key []byte, _ ...pd.GetRegionOpt
}
region := regionsInfo.GetRegionByKey(key)
if region == nil {
log.Warn("region not found",
zap.String("key", hex.EncodeToString(key)),
zap.Uint64("req", reqID))
return nil, ErrRegionNotFound
}
resp := mkPDRegions(region)[0]
return resp, nil
}

func (c *Client) GetPrevRegion(ctx context.Context, key []byte, _ ...pd.GetRegionOption) (*pd.Region, error) {
ctx = withReqID(ctx)
reqID := getReqID(ctx)
_, start, err := codec.DecodeBytes(key, nil)
if err != nil {
return nil, err
Expand All @@ -386,12 +428,17 @@ func (c *Client) GetPrevRegion(ctx context.Context, key []byte, _ ...pd.GetRegio
}
region := regionsInfo.GetPrevRegionByKey(key)
if region == nil {
log.Warn("region not found",
zap.String("key", hex.EncodeToString(key)),
zap.Uint64("req", reqID))
return nil, ErrRegionNotFound
}
return mkPDRegions(region)[0], nil
}

func (c *Client) GetRegionByID(ctx context.Context, regionID uint64, _ ...pd.GetRegionOption) (*pd.Region, error) {
ctx = withReqID(ctx)
reqID := getReqID(ctx)
regionsInfo, err := c.fanout(ctx, "GetRegionByID", http.MethodGet, "sync_region_by_id", &SyncRegionByIDRequest{
RegionID: regionID,
})
Expand All @@ -400,12 +447,17 @@ func (c *Client) GetRegionByID(ctx context.Context, regionID uint64, _ ...pd.Get
}
region := regionsInfo.GetRegion(regionID)
if region == nil {
log.Warn("region not found",
zap.Uint64("id", regionID),
zap.Uint64("req", reqID))
return nil, ErrRegionNotFound
}
return mkPDRegions(region)[0], nil
}

func (c *Client) ScanRegions(ctx context.Context, startKey, endKey []byte, limit int) ([]*pd.Region, error) {
ctx = withReqID(ctx)
reqID := getReqID(ctx)
if limit <= 0 {
limit = 0
}
Expand All @@ -427,6 +479,10 @@ func (c *Client) ScanRegions(ctx context.Context, startKey, endKey []byte, limit
}
regions := regionsInfo.ScanRange(startKey, endKey, limit)
if len(regions) == 0 {
log.Warn("region not found",
zap.String("start", hex.EncodeToString(startKey)),
zap.String("end", hex.EncodeToString(endKey)),
zap.Uint64("req", reqID))
return nil, ErrRegionNotFound
}
return mkPDRegions(regions...), nil
Expand Down