mirror of https://github.com/minio/minio.git
auth/rpc: Make auth rpc client retry configurable. (#3695)
Currently the auth rpc client defaults to to a maximum cap of 30seconds timeout. Make this to be configurable by the caller of authRPCClient during initialization, if no such config is provided then default to 30 seconds.
This commit is contained in:
parent
a170e44689
commit
310bf5bd36
|
@ -35,6 +35,19 @@ type authConfig struct {
|
|||
secureConn bool // Make TLS connection to RPC server or not.
|
||||
serviceName string // Service name of auth server.
|
||||
disableReconnect bool // Disable reconnect on failure or not.
|
||||
|
||||
/// Retry configurable values.
|
||||
|
||||
// Each retry unit multiplicative, measured in time.Duration.
|
||||
// This is the basic unit used for calculating backoffs.
|
||||
retryUnit time.Duration
|
||||
// Maximum retry duration i.e A caller would wait no more than this
|
||||
// duration to continue their loop.
|
||||
retryCap time.Duration
|
||||
|
||||
// Maximum retries an call authRPC client would do for a failed
|
||||
// RPC call.
|
||||
retryAttemptThreshold int
|
||||
}
|
||||
|
||||
// AuthRPCClient is a authenticated RPC client which does authentication before doing Call().
|
||||
|
@ -47,6 +60,18 @@ type AuthRPCClient struct {
|
|||
|
||||
// newAuthRPCClient - returns a JWT based authenticated (go) rpc client, which does automatic reconnect.
|
||||
func newAuthRPCClient(config authConfig) *AuthRPCClient {
|
||||
// Check if retry params are set properly if not default them.
|
||||
emptyDuration := time.Duration(int64(0))
|
||||
if config.retryUnit == emptyDuration {
|
||||
config.retryUnit = defaultRetryUnit
|
||||
}
|
||||
if config.retryCap == emptyDuration {
|
||||
config.retryCap = defaultRetryCap
|
||||
}
|
||||
if config.retryAttemptThreshold == 0 {
|
||||
config.retryAttemptThreshold = globalAuthRPCRetryThreshold
|
||||
}
|
||||
|
||||
return &AuthRPCClient{
|
||||
rpcClient: newRPCClient(config.serverAddr, config.serviceEndpoint, config.secureConn),
|
||||
config: config,
|
||||
|
@ -105,9 +130,13 @@ func (authClient *AuthRPCClient) Call(serviceMethod string, args interface {
|
|||
SetAuthToken(authToken string)
|
||||
SetRequestTime(requestTime time.Time)
|
||||
}, reply interface{}) (err error) {
|
||||
|
||||
// Done channel is used to close any lingering retry routine, as soon
|
||||
// as this function returns.
|
||||
doneCh := make(chan struct{})
|
||||
defer close(doneCh)
|
||||
for i := range newRetryTimer(time.Second, 30*time.Second, MaxJitter, doneCh) {
|
||||
|
||||
for i := range newRetryTimer(authClient.config.retryUnit, authClient.config.retryCap, doneCh) {
|
||||
if err = authClient.call(serviceMethod, args, reply); err == rpc.ErrShutdown {
|
||||
// As connection at server side is closed, close the rpc client.
|
||||
authClient.Close()
|
||||
|
@ -115,7 +144,7 @@ func (authClient *AuthRPCClient) Call(serviceMethod string, args interface {
|
|||
// Retry if reconnect is not disabled.
|
||||
if !authClient.config.disableReconnect {
|
||||
// Retry until threshold reaches.
|
||||
if i < globalAuthRPCRetryThreshold {
|
||||
if i < authClient.config.retryAttemptThreshold {
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
|
|
@ -198,10 +198,11 @@ func retryFormattingXLDisks(firstDisk bool, endpoints []*url.URL, storageDisks [
|
|||
return errInvalidArgument
|
||||
}
|
||||
|
||||
// Create a done channel to control 'ListObjects' go routine.
|
||||
doneCh := make(chan struct{}, 1)
|
||||
// Done channel is used to close any lingering retry routine, as soon
|
||||
// as this function returns.
|
||||
doneCh := make(chan struct{})
|
||||
|
||||
// Indicate to our routine to exit cleanly upon return.
|
||||
// Indicate to our retry routine to exit cleanly, upon this function return.
|
||||
defer close(doneCh)
|
||||
|
||||
// prepare getElapsedTime() to calculate elapsed time since we started trying formatting disks.
|
||||
|
@ -212,7 +213,7 @@ func retryFormattingXLDisks(firstDisk bool, endpoints []*url.URL, storageDisks [
|
|||
}
|
||||
|
||||
// Wait on the jitter retry loop.
|
||||
retryTimerCh := newRetryTimer(time.Second, time.Second*30, MaxJitter, doneCh)
|
||||
retryTimerCh := newRetryTimerSimple(doneCh)
|
||||
for {
|
||||
select {
|
||||
case retryCount := <-retryTimerCh:
|
||||
|
|
|
@ -220,9 +220,12 @@ func (f retryStorage) reInit() (err error) {
|
|||
// Close the underlying connection.
|
||||
f.remoteStorage.Close() // Error here is purposefully ignored.
|
||||
|
||||
// Done channel is used to close any lingering retry routine, as soon
|
||||
// as this function returns.
|
||||
doneCh := make(chan struct{})
|
||||
defer close(doneCh)
|
||||
for i := range newRetryTimer(f.retryUnit, f.retryCap, MaxJitter, doneCh) {
|
||||
|
||||
for i := range newRetryTimer(f.retryUnit, f.retryCap, doneCh) {
|
||||
// Initialize and make a new login attempt.
|
||||
err = f.remoteStorage.Init()
|
||||
if err != nil {
|
||||
|
|
33
cmd/retry.go
33
cmd/retry.go
|
@ -48,7 +48,8 @@ func (r *lockedRandSource) Seed(seed int64) {
|
|||
// MaxJitter will randomize over the full exponential backoff time
|
||||
const MaxJitter = 1.0
|
||||
|
||||
// NoJitter disables the use of jitter for randomizing the exponential backoff time
|
||||
// NoJitter disables the use of jitter for randomizing the
|
||||
// exponential backoff time
|
||||
const NoJitter = 0.0
|
||||
|
||||
// Global random source for fetching random values.
|
||||
|
@ -56,9 +57,11 @@ var globalRandomSource = rand.New(&lockedRandSource{
|
|||
src: rand.NewSource(time.Now().UTC().UnixNano()),
|
||||
})
|
||||
|
||||
// newRetryTimer creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached.
|
||||
func newRetryTimer(unit time.Duration, cap time.Duration, jitter float64, doneCh chan struct{}) <-chan int {
|
||||
// newRetryTimerJitter creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached. - this function is a fully
|
||||
// configurable version, meant for only advanced use cases. For the most part
|
||||
// one should use newRetryTimerSimple and newRetryTimer.
|
||||
func newRetryTimerWithJitter(unit time.Duration, cap time.Duration, jitter float64, doneCh chan struct{}) <-chan int {
|
||||
attemptCh := make(chan int)
|
||||
|
||||
// normalize jitter to the range [0, 1.0]
|
||||
|
@ -113,5 +116,27 @@ func newRetryTimer(unit time.Duration, cap time.Duration, jitter float64, doneCh
|
|||
|
||||
}
|
||||
}()
|
||||
|
||||
// Start reading..
|
||||
return attemptCh
|
||||
}
|
||||
|
||||
// Default retry constants.
|
||||
var (
|
||||
defaultRetryUnit = time.Second // 1 second.
|
||||
defaultRetryCap = 30 * time.Second // 30 seconds.
|
||||
)
|
||||
|
||||
// newRetryTimer creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached. - this function provides
|
||||
// resulting retry values to be of maximum jitter.
|
||||
func newRetryTimer(unit time.Duration, cap time.Duration, doneCh chan struct{}) <-chan int {
|
||||
return newRetryTimerWithJitter(unit, cap, MaxJitter, doneCh)
|
||||
}
|
||||
|
||||
// newRetryTimerSimple creates a timer with exponentially increasing delays
|
||||
// until the maximum retry attempts are reached. - this function is a
|
||||
// simpler version with all default values.
|
||||
func newRetryTimerSimple(doneCh chan struct{}) <-chan int {
|
||||
return newRetryTimerWithJitter(defaultRetryUnit, defaultRetryCap, MaxJitter, doneCh)
|
||||
}
|
||||
|
|
|
@ -0,0 +1,82 @@
|
|||
/*
|
||||
* Minio Cloud Storage, (C) 2016 Minio, Inc.
|
||||
*
|
||||
* 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 cmd
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Tests for retry timer.
|
||||
func TestRetryTimerSimple(t *testing.T) {
|
||||
doneCh := make(chan struct{})
|
||||
attemptCh := newRetryTimerSimple(doneCh)
|
||||
i := <-attemptCh
|
||||
if i != 0 {
|
||||
close(doneCh)
|
||||
t.Fatalf("Invalid attempt counter returned should be 0, found %d instead", i)
|
||||
}
|
||||
i = <-attemptCh
|
||||
if i <= 0 {
|
||||
close(doneCh)
|
||||
t.Fatalf("Invalid attempt counter returned should be greater than 0, found %d instead", i)
|
||||
}
|
||||
close(doneCh)
|
||||
_, ok := <-attemptCh
|
||||
if ok {
|
||||
t.Fatal("Attempt counter should be closed")
|
||||
}
|
||||
}
|
||||
|
||||
// Test retry time with no jitter.
|
||||
func TestRetryTimerWithNoJitter(t *testing.T) {
|
||||
doneCh := make(chan struct{})
|
||||
// No jitter
|
||||
attemptCh := newRetryTimerWithJitter(time.Millisecond, 5*time.Millisecond, NoJitter, doneCh)
|
||||
i := <-attemptCh
|
||||
if i != 0 {
|
||||
close(doneCh)
|
||||
t.Fatalf("Invalid attempt counter returned should be 0, found %d instead", i)
|
||||
}
|
||||
// Loop through the maximum possible attempt.
|
||||
for i = range attemptCh {
|
||||
if i == 30 {
|
||||
close(doneCh)
|
||||
}
|
||||
}
|
||||
_, ok := <-attemptCh
|
||||
if ok {
|
||||
t.Fatal("Attempt counter should be closed")
|
||||
}
|
||||
}
|
||||
|
||||
// Test retry time with Jitter greater than MaxJitter.
|
||||
func TestRetryTimerWithJitter(t *testing.T) {
|
||||
doneCh := make(chan struct{})
|
||||
// Jitter will be set back to 1.0
|
||||
attemptCh := newRetryTimerWithJitter(defaultRetryUnit, defaultRetryCap, 2.0, doneCh)
|
||||
i := <-attemptCh
|
||||
if i != 0 {
|
||||
close(doneCh)
|
||||
t.Fatalf("Invalid attempt counter returned should be 0, found %d instead", i)
|
||||
}
|
||||
close(doneCh)
|
||||
_, ok := <-attemptCh
|
||||
if ok {
|
||||
t.Fatal("Attempt counter should be closed")
|
||||
}
|
||||
}
|
Loading…
Reference in New Issue