-
Notifications
You must be signed in to change notification settings - Fork 2.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
VReplication: throttling info for both source and target; Online DDL …
…propagates said info (#10601) * vreplication throttling information Signed-off-by: Shlomi Noach <[email protected]> * using RateLimiter Signed-off-by: Shlomi Noach <[email protected]> * rowstreamer reports throttling status, captured by vcopier and logged to vreplication table Signed-off-by: Shlomi Noach <[email protected]> * remove magic hint, add 'throttled' field Signed-off-by: Shlomi Noach <[email protected]> * propagate throttling information into schema_migrations table Signed-off-by: Shlomi Noach <[email protected]> * onlineddl/vrepl endtoend test: verify propagation of throttling info Signed-off-by: Shlomi Noach <[email protected]> * test vstreamer throttling Signed-off-by: Shlomi Noach <[email protected]> * simplify: reuse RateLimiter Signed-off-by: Shlomi Noach <[email protected]> * vtadmin_web_proto_types Signed-off-by: Shlomi Noach <[email protected]> * fix EOF return path Signed-off-by: Shlomi Noach <[email protected]> * expose the two new column in 'vtctl Workflow ... Show' Signed-off-by: Shlomi Noach <[email protected]> * adapt test queries to change in wrangler query Signed-off-by: Shlomi Noach <[email protected]> * fix test Signed-off-by: Shlomi Noach <[email protected]> * proto: add Heartbeat indicator in VStreamRowsResponse Signed-off-by: Shlomi Noach <[email protected]> * fix unit tests Signed-off-by: Shlomi Noach <[email protected]> * RateLimiter.Stop() Signed-off-by: Shlomi Noach <[email protected]> * rowstreamer now sends heartbeats, received by vcopier, and updated in _vt.vreplication Signed-off-by: Shlomi Noach <[email protected]> * whoops. Removed debug Sleep() Signed-off-by: Shlomi Noach <[email protected]> * make vtadmin_web_proto_types Signed-off-by: Shlomi Noach <[email protected]> * skip throttled events Signed-off-by: Shlomi Noach <[email protected]> * f allowed to be nil Signed-off-by: Shlomi Noach <[email protected]> * fix 'now' calculation Signed-off-by: Shlomi Noach <[email protected]> * remove numeric precision Signed-off-by: Shlomi Noach <[email protected]> * ix typo Signed-off-by: Shlomi Noach <[email protected]> * indicate Unix timestamp in field name Signed-off-by: Shlomi Noach <[email protected]> * grammar Signed-off-by: Shlomi Noach <[email protected]> * refactor WaitForThrottledTimestamp Signed-off-by: Shlomi Noach <[email protected]> * formalize ComponentName Signed-off-by: Shlomi Noach <[email protected]> * throttled app is vstreamer Signed-off-by: Shlomi Noach <[email protected]>
- Loading branch information
1 parent
01e5005
commit 3012060
Showing
28 changed files
with
873 additions
and
224 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| /* | ||
| Copyright 2022 The Vitess Authors. | ||
| 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, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package timer | ||
|
|
||
| import ( | ||
| "context" | ||
| "math" | ||
| "sync" | ||
| "sync/atomic" | ||
| "time" | ||
| ) | ||
|
|
||
| // RateLimiter runs given tasks, at no more than one per defined duration. | ||
| // For example, we can create a RateLimiter of 1second. Then, we can ask it, over time, to run many | ||
| // tasks. It will only ever run a single task in any 1 second time frame. The rest are ignored. | ||
| type RateLimiter struct { | ||
| tickerValue int64 | ||
| lastDoValue int64 | ||
|
|
||
| mu sync.Mutex | ||
| cancel context.CancelFunc | ||
| } | ||
|
|
||
| // NewRateLimiter creates a new limiter with given duration. It is immediately ready to run tasks. | ||
| func NewRateLimiter(d time.Duration) *RateLimiter { | ||
| r := &RateLimiter{tickerValue: 1} | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| r.cancel = cancel | ||
| go func() { | ||
| ticker := time.NewTicker(d) | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| ticker.Stop() | ||
| case <-ticker.C: | ||
| atomic.StoreInt64(&r.tickerValue, r.tickerValue+1) | ||
| } | ||
| } | ||
| }() | ||
| return r | ||
| } | ||
|
|
||
| // Do runs a given func assuming rate limiting allows. This function is thread safe. | ||
| // f may be nil, in which case it is not invoked. | ||
| func (r *RateLimiter) Do(f func() error) (err error) { | ||
| r.mu.Lock() | ||
| defer r.mu.Unlock() | ||
|
|
||
| if r.lastDoValue >= atomic.LoadInt64(&r.tickerValue) { | ||
| return nil // rate limited. Skipped. | ||
| } | ||
| if f != nil { | ||
| err = f() | ||
| } | ||
| r.lastDoValue = atomic.LoadInt64(&r.tickerValue) | ||
| return err | ||
| } | ||
|
|
||
| // Stop terminates rate limiter's operation and will not allow any more Do() executions. | ||
| func (r *RateLimiter) Stop() { | ||
| r.cancel() | ||
|
|
||
| r.mu.Lock() | ||
| defer r.mu.Unlock() | ||
|
|
||
| r.lastDoValue = math.MaxInt64 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| /* | ||
| Copyright 2020 The Vitess Authors. | ||
| 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, | ||
| WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| See the License for the specific language governing permissions and | ||
| limitations under the License. | ||
| */ | ||
|
|
||
| package timer | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestRateLimiterLong(t *testing.T) { | ||
| r := NewRateLimiter(time.Hour) | ||
| require.NotNil(t, r) | ||
| val := 0 | ||
| incr := func() error { val++; return nil } | ||
| for i := 0; i < 10; i++ { | ||
| err := r.Do(incr) | ||
| assert.NoError(t, err) | ||
| } | ||
| assert.Equal(t, 1, val) | ||
| } | ||
|
|
||
| func TestRateLimiterShort(t *testing.T) { | ||
| r := NewRateLimiter(time.Millisecond * 250) | ||
| require.NotNil(t, r) | ||
| val := 0 | ||
| incr := func() error { val++; return nil } | ||
| for i := 0; i < 10; i++ { | ||
| time.Sleep(time.Millisecond * 100) | ||
| err := r.Do(incr) | ||
| assert.NoError(t, err) | ||
| } | ||
| // we expect some 3-5 entries; this depends on the CI server performance. | ||
| assert.Greater(t, val, 2) | ||
| assert.Less(t, val, 10) | ||
| } | ||
|
|
||
| func TestRateLimiterStop(t *testing.T) { | ||
| r := NewRateLimiter(time.Millisecond * 10) | ||
| require.NotNil(t, r) | ||
| val := 0 | ||
| incr := func() error { val++; return nil } | ||
| for i := 0; i < 5; i++ { | ||
| time.Sleep(time.Millisecond * 10) | ||
| err := r.Do(incr) | ||
| assert.NoError(t, err) | ||
| } | ||
| // we expect some 3-5 entries; this depends on the CI server performance. | ||
| assert.Greater(t, val, 2) | ||
| valSnapshot := val | ||
| r.Stop() | ||
| for i := 0; i < 5; i++ { | ||
| time.Sleep(time.Millisecond * 10) | ||
| err := r.Do(incr) | ||
| assert.NoError(t, err) | ||
| } | ||
| assert.Equal(t, valSnapshot, val) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.