You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 
minio/cmd/rpc.go

281 lines
6.4 KiB

/*
* Minio Cloud Storage, (C) 2018 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 (
"crypto/tls"
"fmt"
"net"
"net/url"
"sync"
"time"
xrpc "github.com/minio/minio/cmd/rpc"
xnet "github.com/minio/minio/pkg/net"
)
// DefaultSkewTime - skew time is 15 minutes between minio peers.
const DefaultSkewTime = 15 * time.Minute
// defaultRPCTimeout - default RPC timeout is one minute.
const defaultRPCTimeout = 5 * time.Minute
// defaultRPCRetryTime - default RPC time to wait before retry after a network error
const defaultRPCRetryTime = 1 * time.Minute
var errRPCRetry = fmt.Errorf("rpc: retry error")
func isNetError(err error) bool {
if err == nil {
return false
}
if uerr, isURLError := err.(*url.Error); isURLError {
if uerr.Timeout() {
return true
}
err = uerr.Err
}
_, isNetOpError := err.(*net.OpError)
return isNetOpError
}
// RPCVersion - RPC semantic version based on semver 2.0.0 https://semver.org/.
type RPCVersion struct {
Major uint64
Minor uint64
Patch uint64
}
// Compare - compares given version with this version.
func (v RPCVersion) Compare(o RPCVersion) int {
compare := func(v1, v2 uint64) int {
if v1 == v2 {
return 0
}
if v1 > v2 {
return 1
}
return -1
}
if r := compare(v.Major, o.Major); r != 0 {
return r
}
if r := compare(v.Minor, o.Minor); r != 0 {
return r
}
return compare(v.Patch, o.Patch)
}
func (v RPCVersion) String() string {
return fmt.Sprintf("%v.%v.%v", v.Major, v.Minor, v.Patch)
}
// AuthArgs - base argument for any RPC call for authentication.
type AuthArgs struct {
Token string
RPCVersion RPCVersion
RequestTime time.Time
}
// Authenticate - checks if given arguments are valid to allow RPC call.
// This is xrpc.Authenticator and is called in RPC server.
func (args AuthArgs) Authenticate() error {
// Check whether request time is within acceptable skew time.
utcNow := time.Now().UTC()
if args.RequestTime.Sub(utcNow) > DefaultSkewTime || utcNow.Sub(args.RequestTime) > DefaultSkewTime {
return fmt.Errorf("client time %v is too apart with server time %v", args.RequestTime, utcNow)
}
if globalRPCAPIVersion.Compare(args.RPCVersion) != 0 {
return fmt.Errorf("version mismatch. expected: %v, received: %v", globalRPCAPIVersion, args.RPCVersion)
}
if !isAuthTokenValid(args.Token) {
return errAuthentication
}
return nil
}
// SetAuthArgs - sets given authentication arguments to this args. This is called in RPC client.
func (args *AuthArgs) SetAuthArgs(authArgs AuthArgs) {
*args = authArgs
}
// VoidReply - void (empty) RPC reply.
type VoidReply struct{}
// RPCClientArgs - RPC client arguments.
type RPCClientArgs struct {
NewAuthTokenFunc func() string
RPCVersion RPCVersion
ServiceName string
ServiceURL *xnet.URL
TLSConfig *tls.Config
}
// validate - checks whether given args are valid or not.
func (args RPCClientArgs) validate() error {
if args.NewAuthTokenFunc == nil {
return fmt.Errorf("NewAuthTokenFunc must not be empty")
}
if args.ServiceName == "" {
return fmt.Errorf("ServiceName must not be empty")
}
if args.ServiceURL.Scheme != "http" && args.ServiceURL.Scheme != "https" {
return fmt.Errorf("unknown RPC URL %v", args.ServiceURL)
}
if args.ServiceURL.User != nil || args.ServiceURL.ForceQuery || args.ServiceURL.RawQuery != "" || args.ServiceURL.Fragment != "" {
return fmt.Errorf("unknown RPC URL %v", args.ServiceURL)
}
if args.ServiceURL.Scheme == "https" && args.TLSConfig == nil {
return fmt.Errorf("tls configuration must not be empty for https url %v", args.ServiceURL)
}
return nil
}
// RPCClient - base RPC client.
type RPCClient struct {
sync.RWMutex
args RPCClientArgs
authToken string
rpcClient *xrpc.Client
retryTicker *time.Ticker
}
func (client *RPCClient) setRetryTicker(ticker *time.Ticker) {
if ticker == nil {
client.RLock()
isNil := client.retryTicker == nil
client.RUnlock()
if isNil {
return
}
}
client.Lock()
defer client.Unlock()
if client.retryTicker != nil {
client.retryTicker.Stop()
}
client.retryTicker = ticker
}
// Call - calls servicemethod on remote server.
func (client *RPCClient) Call(serviceMethod string, args interface {
SetAuthArgs(args AuthArgs)
}, reply interface{}) (err error) {
lockedCall := func() error {
client.RLock()
retryTicker := client.retryTicker
client.RUnlock()
if retryTicker != nil {
select {
case <-retryTicker.C:
default:
return errRPCRetry
}
}
client.RLock()
authToken := client.authToken
client.RUnlock()
// Make RPC call.
args.SetAuthArgs(AuthArgs{authToken, client.args.RPCVersion, time.Now().UTC()})
return client.rpcClient.Call(serviceMethod, args, reply)
}
call := func() error {
err = lockedCall()
if err == errRPCRetry {
return err
}
if isNetError(err) {
client.setRetryTicker(time.NewTicker(defaultRPCRetryTime))
} else {
client.setRetryTicker(nil)
}
return err
}
// If authentication error is received, retry the same call only once
// with new authentication token.
if err = call(); err == nil {
return nil
}
if err.Error() != errAuthentication.Error() {
return err
}
client.Lock()
client.authToken = client.args.NewAuthTokenFunc()
client.Unlock()
return call()
}
// Close - closes underneath RPC client.
func (client *RPCClient) Close() error {
client.Lock()
defer client.Unlock()
client.authToken = ""
return client.rpcClient.Close()
}
// ServiceURL - returns service URL used for RPC call.
func (client *RPCClient) ServiceURL() *xnet.URL {
// Take copy of ServiceURL
u := *(client.args.ServiceURL)
return &u
}
// NewRPCClient - returns new RPC client.
func NewRPCClient(args RPCClientArgs) (*RPCClient, error) {
if err := args.validate(); err != nil {
return nil, err
}
rpcClient, err := xrpc.NewClient(args.ServiceURL, args.TLSConfig, defaultRPCTimeout)
if err != nil {
return nil, err
}
return &RPCClient{
args: args,
authToken: args.NewAuthTokenFunc(),
rpcClient: rpcClient,
}, nil
}