add rate limit for local backend, parse limit args

[recall704]

What problem does this PR solve?

add rate limit for local backend

What is changed and how it works?

set rate limit for io.Writer

[ti-chi-bot]

[REVIEW NOTIFICATION]

This pull request has not been approved.

To complete the pull request process, please ask the reviewers in the list to review by filling /cc @reviewer in the comment. After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the list by filling /assign @committer in the comment to help you merge this pull request.

The full list of commands accepted by this bot can be found here.

Reviewer can indicate their review by writing /lgtm in a comment. Reviewer can cancel approval by writing /lgtm cancel in a comment.

[CLAassistant]

All committers have signed the CLA.

[ti-chi-bot]

Welcome @recall704!

It looks like this is your first PR to pingcap/br 🎉.

I'm the bot to help you request reviewers, add labels and more, See available commands.

We want to make sure your contribution gets all the attention it needs!



Thank you, and welcome to pingcap/br. :smiley:

[sre-bot]

No release note, Please follow https://github.com/pingcap/community/blob/master/contributors/release-note-checker.md

[sre-bot]

No release note, Please follow https://github.com/pingcap/community/blob/master/contributors/release-note-checker.md

[kennytm]

Instead of adding a rate_limit field to local storage only, I suggest adapting to the existing --ratelimit parameter.

• Not good: -o 'local:///data/output?rate_limit=100'
• Good: --ratelimit 100 -o 'local:///data/output'

The rate limiter should introduce a wrapper of ExternalStorage similar to the withCompression type.

[recall704]

Instead of adding a rate_limit field to local storage only, I suggest adapting to the existing --ratelimit parameter.

• Not good: -o 'local:///data/output?rate_limit=100'
• Good: --ratelimit 100 -o 'local:///data/output'

The rate limiter should introduce a wrapper of ExternalStorage similar to the withCompression type.

type withRatelimit struct {
	ExternalStorage
	Ratelimit uint64 // Byte/sec
}

// WithRatelimit
func WithRatelimit(inner ExternalStorage, ratelimit uint64) ExternalStorage {
	if ratelimit == 0 {
		return inner
	}
	return &withRatelimit{ExternalStorage: inner, Ratelimit: ratelimit}
}

func (r *withRatelimit) Create(ctx context.Context, name string) (ExternalFileWriter, error) {
	var (
		writer ExternalFileWriter
		err    error
	)
	if localStorage, ok := r.ExternalStorage.(*LocalStorage); ok {
		file, err := os.Create(filepath.Join(localStorage.base, name))
		if err != nil {
			return nil, errors.Trace(err)
		}
		buf := bufio.NewWriter(file)
		if r.Ratelimit > 0 {
			w := shapeio.NewWriterWithContext(buf, ctx)
			w.SetRateLimit(float64(r.Ratelimit))
			return newFlushStorageWriter(w, buf, file), nil
		}
		return newFlushStorageWriter(buf, buf, file), nil
	}
	writer, err = r.ExternalStorage.Create(ctx, name)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return writer, nil
}

I try to use withRatelimit to set rate limit, but I have to overwrite Create method, any idea?

thanks

[kennytm]

@recall704 sorry for late reply, there's a week-long holiday in China and Japan 🙃.

I try to use withRatelimit to set rate limit, but I have to overwrite Create method, any idea?

you could just return a wrapper type that implements ExternalFileWriter:

type withRateLimitWriter struct {
    ExternalFileWriter
    Ratelimit uint64
}

func (r *withRatelimit) Create(ctx context.Context, name string) (ExternalFileWriter, error) {
    inner, err := r.ExternalStorage.Create(ctx, name)
    if err != nil {
        return nil, err
    }
    return &withRateLimitWriter{
        ExternalFileWriter: inner,
        Ratelimit:          r.Ratelimit,
    }, nil
}

and then apply the Ratelimit in its Write method.

func (rw *withRateLimitWriter) Write(ctx context.Context, p []byte) (int, error) {
    // do rate limiting here.
}

There's no need to figure out how to fit this into shapeio's Reader/Writer API, since that package is just using golang.org/x/time/rate anyway. I suggest you use rate.Limiter directly instead of shapeio (allowing multiple writers from Create sharing the same limiter as a global rate limit).

[ti-chi-bot]

@recall704: PR needs rebase.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

[sre-bot]

No release note, Please follow https://github.com/pingcap/community/blob/master/contributors/release-note-checker.md