Bring semantic versioning to provide for rolling upgrades (#5495)

This PR brings semver capabilities in our RPC layer to
ensure that we can upgrade the servers in rolling fashion
while keeping I/O in progress. This is only a framework change
the functionality remains the same as such and we do not
have any special API changes for now. But in future when
we bring in API changes we will be able to upgrade servers
without a downtime.

Additional change in this PR is to not abort when serverVersions
mismatch in a distributed cluster, instead wait for the quorum
treat the situation as if the server is down. This allows
for administrator to properly upgrade all the servers in the cluster.

Fixes #5393
master
Harshavardhana 7 years ago committed by kannappanr
parent 48218272cc
commit 1164fc60f3
  1. 18
      cmd/admin-rpc-server_test.go
  2. 22
      cmd/auth-rpc-client.go
  3. 10
      cmd/auth-rpc-server_test.go
  4. 3
      cmd/globals.go
  5. 12
      cmd/lock-rpc-server_test.go
  6. 14
      cmd/prepare-storage.go
  7. 26
      cmd/prepare-storage_test.go
  8. 68
      cmd/rpc-common.go
  9. 56
      cmd/rpc-common_test.go
  10. 4
      cmd/storage-rpc-client.go
  11. 4
      cmd/storage-rpc-client_test.go
  12. 10
      cmd/storage-rpc-server_test.go
  13. 4
      cmd/typed-errors.go
  14. 22
      cmd/web-handlers_test.go

@ -42,7 +42,7 @@ func testAdminCmd(cmd cmdType, t *testing.T) {
adminServer := adminCmd{} adminServer := adminCmd{}
args := LoginRPCArgs{ args := LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
err = adminServer.Login(&args, &LoginRPCReply{}) err = adminServer.Login(&args, &LoginRPCReply{})
@ -56,9 +56,10 @@ func testAdminCmd(cmd cmdType, t *testing.T) {
}() }()
sa := SignalServiceArgs{ sa := SignalServiceArgs{
AuthRPCArgs: AuthRPCArgs{AuthToken: token}, AuthRPCArgs: AuthRPCArgs{AuthToken: token, Version: globalRPCAPIVersion},
Sig: cmd.toServiceSignal(), Sig: cmd.toServiceSignal(),
} }
genReply := AuthRPCReply{} genReply := AuthRPCReply{}
switch cmd { switch cmd {
case restartCmd, stopCmd: case restartCmd, stopCmd:
@ -123,7 +124,7 @@ func TestReInitDisks(t *testing.T) {
args := LoginRPCArgs{ args := LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
err = adminServer.Login(&args, &LoginRPCReply{}) err = adminServer.Login(&args, &LoginRPCReply{})
@ -133,6 +134,7 @@ func TestReInitDisks(t *testing.T) {
authArgs := AuthRPCArgs{ authArgs := AuthRPCArgs{
AuthToken: token, AuthToken: token,
Version: globalRPCAPIVersion,
} }
authReply := AuthRPCReply{} authReply := AuthRPCReply{}
@ -150,7 +152,7 @@ func TestReInitDisks(t *testing.T) {
fsAdminServer := adminCmd{} fsAdminServer := adminCmd{}
fsArgs := LoginRPCArgs{ fsArgs := LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
fsReply := LoginRPCReply{} fsReply := LoginRPCReply{}
@ -161,6 +163,7 @@ func TestReInitDisks(t *testing.T) {
authArgs = AuthRPCArgs{ authArgs = AuthRPCArgs{
AuthToken: token, AuthToken: token,
Version: globalRPCAPIVersion,
} }
authReply = AuthRPCReply{} authReply = AuthRPCReply{}
// Attempt ReInitDisks service on a FS backend. // Attempt ReInitDisks service on a FS backend.
@ -192,7 +195,7 @@ func TestGetConfig(t *testing.T) {
args := LoginRPCArgs{ args := LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
reply := LoginRPCReply{} reply := LoginRPCReply{}
@ -203,6 +206,7 @@ func TestGetConfig(t *testing.T) {
authArgs := AuthRPCArgs{ authArgs := AuthRPCArgs{
AuthToken: token, AuthToken: token,
Version: globalRPCAPIVersion,
} }
configReply := ConfigReply{} configReply := ConfigReply{}
@ -239,7 +243,7 @@ func TestWriteAndCommitConfig(t *testing.T) {
} }
args := LoginRPCArgs{ args := LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
reply := LoginRPCReply{} reply := LoginRPCReply{}
@ -254,6 +258,7 @@ func TestWriteAndCommitConfig(t *testing.T) {
wArgs := WriteConfigArgs{ wArgs := WriteConfigArgs{
AuthRPCArgs: AuthRPCArgs{ AuthRPCArgs: AuthRPCArgs{
AuthToken: token, AuthToken: token,
Version: globalRPCAPIVersion,
}, },
TmpFileName: tmpFileName, TmpFileName: tmpFileName,
Buf: buf, Buf: buf,
@ -271,6 +276,7 @@ func TestWriteAndCommitConfig(t *testing.T) {
cArgs := CommitConfigArgs{ cArgs := CommitConfigArgs{
AuthRPCArgs: AuthRPCArgs{ AuthRPCArgs: AuthRPCArgs{
AuthToken: token, AuthToken: token,
Version: globalRPCAPIVersion,
}, },
FileName: tmpFileName, FileName: tmpFileName,
} }

@ -26,6 +26,7 @@ import (
"net" "net"
"net/http" "net/http"
"net/rpc" "net/rpc"
"strings"
"sync" "sync"
"time" "time"
) )
@ -61,9 +62,10 @@ type authConfig struct {
// AuthRPCClient is a authenticated RPC client which does authentication before doing Call(). // AuthRPCClient is a authenticated RPC client which does authentication before doing Call().
type AuthRPCClient struct { type AuthRPCClient struct {
sync.RWMutex // Mutex to lock this object. sync.RWMutex // Mutex to lock this object.
rpcClient *rpc.Client // RPC Client to make any RPC call. rpcClient *rpc.Client // RPC client to make any RPC call.
config authConfig // Authentication configuration information. config authConfig // Authentication configuration information.
authToken string // Authentication token. authToken string // Authentication token.
version semVersion // RPC version.
} }
// newAuthRPCClient - returns a JWT based authenticated (go) rpc client, which does automatic reconnect. // newAuthRPCClient - returns a JWT based authenticated (go) rpc client, which does automatic reconnect.
@ -81,7 +83,8 @@ func newAuthRPCClient(config authConfig) *AuthRPCClient {
} }
return &AuthRPCClient{ return &AuthRPCClient{
config: config, config: config,
version: globalRPCAPIVersion,
} }
} }
@ -117,7 +120,7 @@ func (authClient *AuthRPCClient) Login() (err error) {
loginMethod = authClient.config.serviceName + loginMethodName loginMethod = authClient.config.serviceName + loginMethodName
loginArgs = LoginRPCArgs{ loginArgs = LoginRPCArgs{
AuthToken: authToken, AuthToken: authToken,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
) )
@ -130,6 +133,11 @@ func (authClient *AuthRPCClient) Login() (err error) {
} }
if err = rpcClient.Call(loginMethod, &loginArgs, &LoginRPCReply{}); err != nil { if err = rpcClient.Call(loginMethod, &loginArgs, &LoginRPCReply{}); err != nil {
// gob doesn't provide any typed errors for us to reflect
// upon, this is the only way to return proper error.
if strings.Contains(err.Error(), "gob: wrong type") {
return errRPCAPIVersionUnsupported
}
return err return err
} }
@ -144,6 +152,7 @@ func (authClient *AuthRPCClient) Login() (err error) {
// call makes a RPC call after logs into the server. // call makes a RPC call after logs into the server.
func (authClient *AuthRPCClient) call(serviceMethod string, args interface { func (authClient *AuthRPCClient) call(serviceMethod string, args interface {
SetAuthToken(authToken string) SetAuthToken(authToken string)
SetRPCAPIVersion(version semVersion)
}, reply interface{}) (err error) { }, reply interface{}) (err error) {
if err = authClient.Login(); err != nil { if err = authClient.Login(); err != nil {
return err return err
@ -153,6 +162,7 @@ func (authClient *AuthRPCClient) call(serviceMethod string, args interface {
authClient.RLock() authClient.RLock()
defer authClient.RUnlock() defer authClient.RUnlock()
args.SetAuthToken(authClient.authToken) args.SetAuthToken(authClient.authToken)
args.SetRPCAPIVersion(authClient.version)
// Do an RPC call. // Do an RPC call.
return authClient.rpcClient.Call(serviceMethod, args, reply) return authClient.rpcClient.Call(serviceMethod, args, reply)
@ -161,6 +171,7 @@ func (authClient *AuthRPCClient) call(serviceMethod string, args interface {
// Call executes RPC call till success or globalAuthRPCRetryThreshold on ErrShutdown. // Call executes RPC call till success or globalAuthRPCRetryThreshold on ErrShutdown.
func (authClient *AuthRPCClient) Call(serviceMethod string, args interface { func (authClient *AuthRPCClient) Call(serviceMethod string, args interface {
SetAuthToken(authToken string) SetAuthToken(authToken string)
SetRPCAPIVersion(version semVersion)
}, reply interface{}) (err error) { }, reply interface{}) (err error) {
// Done channel is used to close any lingering retry routine, as soon // Done channel is used to close any lingering retry routine, as soon
@ -181,6 +192,11 @@ func (authClient *AuthRPCClient) Call(serviceMethod string, args interface {
} }
} }
} }
// gob doesn't provide any typed errors for us to reflect
// upon, this is the only way to return proper error.
if err != nil && strings.Contains(err.Error(), "gob: wrong type") {
err = errRPCAPIVersionUnsupported
}
break break
} }
return err return err

@ -43,7 +43,7 @@ func TestLogin(t *testing.T) {
{ {
args: LoginRPCArgs{ args: LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
}, },
skewTime: 0, skewTime: 0,
expectedErr: nil, expectedErr: nil,
@ -52,16 +52,16 @@ func TestLogin(t *testing.T) {
{ {
args: LoginRPCArgs{ args: LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: "INVALID-" + Version, Version: semVersion{2, 0, 0},
}, },
skewTime: 0, skewTime: 0,
expectedErr: errServerVersionMismatch, expectedErr: errRPCAPIVersionUnsupported,
}, },
// Valid username, password and version, not request time // Valid username, password and version, not request time
{ {
args: LoginRPCArgs{ args: LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
}, },
skewTime: 20 * time.Minute, skewTime: 20 * time.Minute,
expectedErr: errServerTimeMismatch, expectedErr: errServerTimeMismatch,
@ -70,7 +70,7 @@ func TestLogin(t *testing.T) {
{ {
args: LoginRPCArgs{ args: LoginRPCArgs{
AuthToken: "", AuthToken: "",
Version: Version, Version: globalRPCAPIVersion,
}, },
skewTime: 0, skewTime: 0,
expectedErr: errAuthentication, expectedErr: errAuthentication,

@ -165,6 +165,9 @@ var (
// Set to store standard storage class // Set to store standard storage class
globalStandardStorageClass storageClass globalStandardStorageClass storageClass
// RPC version.
globalRPCAPIVersion = semVersion{1, 0, 0}
// Add new variable global values here. // Add new variable global values here.
) )

@ -64,7 +64,7 @@ func createLockTestServer(t *testing.T) (string, *lockServer, string) {
} }
loginArgs := LoginRPCArgs{ loginArgs := LoginRPCArgs{
AuthToken: token, AuthToken: token,
Version: Version, Version: globalRPCAPIVersion,
RequestTime: UTCNow(), RequestTime: UTCNow(),
} }
loginReply := LoginRPCReply{} loginReply := LoginRPCReply{}
@ -87,6 +87,7 @@ func TestLockRpcServerLock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la.SetAuthToken(token) la.SetAuthToken(token)
la.SetRPCAPIVersion(globalRPCAPIVersion)
// Claim a lock // Claim a lock
var result bool var result bool
@ -120,6 +121,7 @@ func TestLockRpcServerLock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la2.SetAuthToken(token) la2.SetAuthToken(token)
la2.SetRPCAPIVersion(globalRPCAPIVersion)
err = locker.Lock(&la2, &result) err = locker.Lock(&la2, &result)
if err != nil { if err != nil {
@ -143,6 +145,7 @@ func TestLockRpcServerUnlock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la.SetAuthToken(token) la.SetAuthToken(token)
la.SetRPCAPIVersion(globalRPCAPIVersion)
// First test return of error when attempting to unlock a lock that does not exist // First test return of error when attempting to unlock a lock that does not exist
var result bool var result bool
@ -188,6 +191,7 @@ func TestLockRpcServerRLock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la.SetAuthToken(token) la.SetAuthToken(token)
la.SetRPCAPIVersion(globalRPCAPIVersion)
// Claim a lock // Claim a lock
var result bool var result bool
@ -221,6 +225,7 @@ func TestLockRpcServerRLock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la2.SetAuthToken(token) la2.SetAuthToken(token)
la2.SetRPCAPIVersion(globalRPCAPIVersion)
err = locker.RLock(&la2, &result) err = locker.RLock(&la2, &result)
if err != nil { if err != nil {
@ -244,6 +249,7 @@ func TestLockRpcServerRUnlock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la.SetAuthToken(token) la.SetAuthToken(token)
la.SetRPCAPIVersion(globalRPCAPIVersion)
// First test return of error when attempting to unlock a read-lock that does not exist // First test return of error when attempting to unlock a read-lock that does not exist
var result bool var result bool
@ -268,6 +274,7 @@ func TestLockRpcServerRUnlock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la2.SetAuthToken(token) la2.SetAuthToken(token)
la2.SetRPCAPIVersion(globalRPCAPIVersion)
// ... and create a second lock on same resource // ... and create a second lock on same resource
err = locker.RLock(&la2, &result) err = locker.RLock(&la2, &result)
@ -330,6 +337,7 @@ func TestLockRpcServerForceUnlock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
laForce.SetAuthToken(token) laForce.SetAuthToken(token)
laForce.SetRPCAPIVersion(globalRPCAPIVersion)
// First test that UID should be empty // First test that UID should be empty
var result bool var result bool
@ -352,6 +360,7 @@ func TestLockRpcServerForceUnlock(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la.SetAuthToken(token) la.SetAuthToken(token)
la.SetRPCAPIVersion(globalRPCAPIVersion)
// Create lock ... (so that we can force unlock) // Create lock ... (so that we can force unlock)
err = locker.Lock(&la, &result) err = locker.Lock(&la, &result)
@ -394,6 +403,7 @@ func TestLockRpcServerExpired(t *testing.T) {
ServiceEndpoint: "rpc-path", ServiceEndpoint: "rpc-path",
}) })
la.SetAuthToken(token) la.SetAuthToken(token)
la.SetRPCAPIVersion(globalRPCAPIVersion)
// Unknown lock at server will return expired = true // Unknown lock at server will return expired = true
var expired bool var expired bool

@ -101,13 +101,13 @@ const (
var configErrs = []error{ var configErrs = []error{
errInvalidAccessKeyID, errInvalidAccessKeyID,
errAuthentication, errAuthentication,
errServerVersionMismatch, errRPCAPIVersionUnsupported,
errServerTimeMismatch, errServerTimeMismatch,
} }
// Quick error to actions converts looking for specific errors // Config errs to actions converts looking for specific config errors
// which need to be returned quickly and server should wait instead. // which need to be returned quickly and server should wait instead.
func quickErrToActions(errMap map[error]int) InitActions { func configErrsToActions(errMap map[error]int) InitActions {
var action InitActions var action InitActions
for _, configErr := range configErrs { for _, configErr := range configErrs {
if errMap[configErr] > 0 { if errMap[configErr] > 0 {
@ -168,7 +168,7 @@ func prepForInitXL(firstDisk bool, sErrs []error, diskCount int) InitActions {
} }
// Validates and converts specific config errors into WaitForConfig. // Validates and converts specific config errors into WaitForConfig.
if quickErrToActions(errMap) == WaitForConfig { if configErrsToActions(errMap) == WaitForConfig {
return WaitForConfig return WaitForConfig
} }
@ -311,13 +311,15 @@ func retryFormattingXLDisks(firstDisk bool, endpoints EndpointList, storageDisks
) )
case WaitForConfig: case WaitForConfig:
// Print configuration errors. // Print configuration errors.
return reduceInitXLErrs(storageDisks, sErrs) log.Printf(
"Initializing data volume. Waiting for configuration issues to be fixed (%s). (elapsed %s)\n",
reduceInitXLErrs(storageDisks, sErrs), getElapsedTime())
case WaitForAll: case WaitForAll:
log.Printf("Initializing data volume for first time. Waiting for other servers to come online (elapsed %s)\n", getElapsedTime()) log.Printf("Initializing data volume for first time. Waiting for other servers to come online (elapsed %s)\n", getElapsedTime())
case WaitForFormatting: case WaitForFormatting:
log.Printf("Initializing data volume for first time. Waiting for first server to come online (elapsed %s)\n", getElapsedTime()) log.Printf("Initializing data volume for first time. Waiting for first server to come online (elapsed %s)\n", getElapsedTime())
} }
case <-globalServiceDoneCh: case <-globalOSSignalCh:
return fmt.Errorf("Initializing data volumes gracefully stopped") return fmt.Errorf("Initializing data volumes gracefully stopped")
} }
} }

@ -67,7 +67,7 @@ func TestReduceInitXLErrs(t *testing.T) {
{[]error{nil, nil, nil, nil}, ""}, {[]error{nil, nil, nil, nil}, ""},
{[]error{errUnformattedDisk, nil, nil, nil}, "\n[01/04] " + storageDisks[0].String() + " : unformatted disk found"}, {[]error{errUnformattedDisk, nil, nil, nil}, "\n[01/04] " + storageDisks[0].String() + " : unformatted disk found"},
{[]error{errUnformattedDisk, errUnformattedDisk, nil, nil}, "\n[01/04] " + storageDisks[0].String() + " : unformatted disk found" + "\n[02/04] " + storageDisks[1].String() + " : unformatted disk found"}, {[]error{errUnformattedDisk, errUnformattedDisk, nil, nil}, "\n[01/04] " + storageDisks[0].String() + " : unformatted disk found" + "\n[02/04] " + storageDisks[1].String() + " : unformatted disk found"},
{[]error{errUnformattedDisk, errUnformattedDisk, errServerVersionMismatch, nil}, storageDisks[2].String() + ": Server versions do not match"}, {[]error{errUnformattedDisk, errUnformattedDisk, errRPCAPIVersionUnsupported, nil}, storageDisks[2].String() + ": Unsupported rpc API version"},
} }
for i, test := range testCases { for i, test := range testCases {
actual := reduceInitXLErrs(storageDisks, test.sErrs) actual := reduceInitXLErrs(storageDisks, test.sErrs)
@ -123,24 +123,29 @@ func TestPrepForInitXL(t *testing.T) {
} }
// Invalid access key id. // Invalid access key id.
accessKeyIDErr := []error{ accessKeyIDErr := []error{
errInvalidAccessKeyID, errInvalidAccessKeyID, errInvalidAccessKeyID, errInvalidAccessKeyID,
errInvalidAccessKeyID, nil, nil, nil, errInvalidAccessKeyID, nil, nil, nil,
nil, nil, nil, nil,
} }
// Authentication error. // Authentication error.
authenticationErr := []error{ authenticationErr := []error{
nil, nil, nil, nil, nil, nil, nil, errAuthentication,
errAuthentication, nil, nil, nil, errAuthentication, errAuthentication, errAuthentication, errAuthentication,
} }
// Server version mismatch. // Unsupported rpc API version.
serverVersionMismatch := []error{ rpcUnsupportedVersion := []error{
errServerVersionMismatch, nil, nil, nil, errRPCAPIVersionUnsupported, errRPCAPIVersionUnsupported, errRPCAPIVersionUnsupported, errRPCAPIVersionUnsupported,
errServerVersionMismatch, nil, nil, nil, errRPCAPIVersionUnsupported, nil, nil, nil,
} }
// Server time mismatch. // Server time mismatch.
serverTimeMismatch := []error{ serverTimeMismatch := []error{
nil, nil, nil, nil, errServerTimeMismatch, errServerTimeMismatch, errServerTimeMismatch, errServerTimeMismatch,
errServerTimeMismatch, nil, nil, nil, errServerTimeMismatch, nil, nil, nil,
} }
// Collection of config errs.
configErrs := []error{
errServerTimeMismatch, errServerTimeMismatch, errRPCAPIVersionUnsupported, errAuthentication,
errInvalidAccessKeyID, nil, nil, nil,
}
// Suggest to heal under formatted disks in quorum. // Suggest to heal under formatted disks in quorum.
formattedDisksInQuorum := []error{ formattedDisksInQuorum := []error{
nil, nil, nil, nil, nil, nil, nil, nil,
@ -187,8 +192,9 @@ func TestPrepForInitXL(t *testing.T) {
// Config mistakes. // Config mistakes.
{true, accessKeyIDErr, 8, WaitForConfig}, {true, accessKeyIDErr, 8, WaitForConfig},
{true, authenticationErr, 8, WaitForConfig}, {true, authenticationErr, 8, WaitForConfig},
{true, serverVersionMismatch, 8, WaitForConfig}, {true, rpcUnsupportedVersion, 8, WaitForConfig},
{true, serverTimeMismatch, 8, WaitForConfig}, {true, serverTimeMismatch, 8, WaitForConfig},
{true, configErrs, 8, WaitForConfig},
} }
for i, test := range testCases { for i, test := range testCases {
actual := prepForInitXL(test.firstDisk, test.errs, test.diskCount) actual := prepForInitXL(test.firstDisk, test.errs, test.diskCount)

@ -1,5 +1,5 @@
/* /*
* Minio Cloud Storage, (C) 2016, 2017 Minio, Inc. * Minio Cloud Storage, (C) 2016, 2017, 2018 Minio, Inc.
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@ -17,6 +17,7 @@
package cmd package cmd
import ( import (
"strconv"
"time" "time"
"github.com/minio/dsync" "github.com/minio/dsync"
@ -33,10 +34,51 @@ func isRequestTimeAllowed(requestTime time.Time) bool {
utcNow.Sub(requestTime) > rpcSkewTimeAllowed) utcNow.Sub(requestTime) > rpcSkewTimeAllowed)
} }
// semVersion - RPC semantic versioning.
type semVersion struct {
Major uint64
Minor uint64
Patch uint64
}
// semver comparator implementation based on the semver 2.0.0 https://semver.org/.
func (v semVersion) Compare(o semVersion) int {
if v.Major != o.Major {
if v.Major > o.Major {
return 1
}
return -1
}
if v.Minor != o.Minor {
if v.Minor > o.Minor {
return 1
}
return -1
}
if v.Patch != o.Patch {
if v.Patch > o.Patch {
return 1
}
return -1
}
return 0
}
func (v semVersion) String() string {
b := make([]byte, 0, 5)
b = strconv.AppendUint(b, v.Major, 10)
b = append(b, '.')
b = strconv.AppendUint(b, v.Minor, 10)
b = append(b, '.')
b = strconv.AppendUint(b, v.Patch, 10)
return string(b)
}
// AuthRPCArgs represents minimum required arguments to make any authenticated RPC call. // AuthRPCArgs represents minimum required arguments to make any authenticated RPC call.
type AuthRPCArgs struct { type AuthRPCArgs struct {
// Authentication token to be verified by the server for every RPC call. // Authentication token to be verified by the server for every RPC call.
AuthToken string AuthToken string
Version semVersion
} }
// SetAuthToken - sets the token to the supplied value. // SetAuthToken - sets the token to the supplied value.
@ -44,8 +86,21 @@ func (args *AuthRPCArgs) SetAuthToken(authToken string) {
args.AuthToken = authToken args.AuthToken = authToken
} }
// SetRPCAPIVersion - sets the rpc version to the supplied value.
func (args *AuthRPCArgs) SetRPCAPIVersion(version semVersion) {
args.Version = version
}
// IsAuthenticated - validated whether this auth RPC args are already authenticated or not. // IsAuthenticated - validated whether this auth RPC args are already authenticated or not.
func (args AuthRPCArgs) IsAuthenticated() error { func (args AuthRPCArgs) IsAuthenticated() error {
// checks if rpc Version is not equal to current server rpc version.
// this is fine for now, but in future when we add backward compatible
// APIs we need to make sure to allow lesser versioned clients to
// talk over RPC, until then we are fine with this check.
if args.Version.Compare(globalRPCAPIVersion) != 0 {
return errRPCAPIVersionUnsupported
}
// Check whether the token is valid // Check whether the token is valid
if !isAuthTokenValid(args.AuthToken) { if !isAuthTokenValid(args.AuthToken) {
return errInvalidToken return errInvalidToken
@ -61,15 +116,18 @@ type AuthRPCReply struct{}
// LoginRPCArgs - login username and password for RPC. // LoginRPCArgs - login username and password for RPC.
type LoginRPCArgs struct { type LoginRPCArgs struct {
AuthToken string AuthToken string
Version string Version semVersion
RequestTime time.Time RequestTime time.Time
} }
// IsValid - validates whether this LoginRPCArgs are valid for authentication. // IsValid - validates whether this LoginRPCArgs are valid for authentication.
func (args LoginRPCArgs) IsValid() error { func (args LoginRPCArgs) IsValid() error {
// Check if version matches. // checks if rpc Version is not equal to current server rpc version.
if args.Version != Version { // this is fine for now, but in future when we add backward compatible
return errServerVersionMismatch // APIs we need to make sure to allow lesser versioned clients to
// talk over RPC, until then we are fine with this check.
if args.Version.Compare(globalRPCAPIVersion) != 0 {
return errRPCAPIVersionUnsupported
} }
if !isRequestTimeAllowed(args.RequestTime) { if !isRequestTimeAllowed(args.RequestTime) {

@ -0,0 +1,56 @@
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, semVersion 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"
// Tests version comparator.
func TestCompare(t *testing.T) {
type compareTest struct {
v1 semVersion
v2 semVersion
result int
}
var compareTests = []compareTest{
{semVersion{1, 0, 0}, semVersion{1, 0, 0}, 0},
{semVersion{2, 0, 0}, semVersion{1, 0, 0}, 1},
{semVersion{0, 1, 0}, semVersion{0, 1, 0}, 0},
{semVersion{0, 2, 0}, semVersion{0, 1, 0}, 1},
{semVersion{0, 0, 1}, semVersion{0, 0, 1}, 0},
{semVersion{0, 0, 2}, semVersion{0, 0, 1}, 1},
{semVersion{1, 2, 3}, semVersion{1, 2, 3}, 0},
{semVersion{2, 2, 4}, semVersion{1, 2, 4}, 1},
{semVersion{1, 3, 3}, semVersion{1, 2, 3}, 1},
{semVersion{1, 2, 4}, semVersion{1, 2, 3}, 1},
// Spec Examples #11
{semVersion{1, 0, 0}, semVersion{2, 0, 0}, -1},
{semVersion{2, 0, 0}, semVersion{2, 1, 0}, -1},
{semVersion{2, 1, 0}, semVersion{2, 1, 1}, -1},
}
for _, test := range compareTests {
if res := test.v1.Compare(test.v2); res != test.result {
t.Errorf("Comparing %q : %q, expected %d but got %d", test.v1, test.v2, test.result, res)
}
// Test if reverse is true as well.
if res := test.v2.Compare(test.v1); res != -test.result {
t.Errorf("Comparing %q : %q, expected %d but got %d", test.v2, test.v1, -test.result, res)
}
}
}

@ -85,8 +85,8 @@ func toStorageErr(err error) error {
return errInvalidAccessKeyID return errInvalidAccessKeyID
case errAuthentication.Error(): case errAuthentication.Error():
return errAuthentication return errAuthentication
case errServerVersionMismatch.Error(): case errRPCAPIVersionUnsupported.Error():
return errServerVersionMismatch return errRPCAPIVersionUnsupported
case errServerTimeMismatch.Error(): case errServerTimeMismatch.Error():
return errServerTimeMismatch return errServerTimeMismatch
} }

@ -190,8 +190,8 @@ func TestStorageErr(t *testing.T) {
err: fmt.Errorf("%s", errAuthentication.Error()), err: fmt.Errorf("%s", errAuthentication.Error()),
}, },
{ {
expectedErr: errServerVersionMismatch, expectedErr: errRPCAPIVersionUnsupported,
err: fmt.Errorf("%s", errServerVersionMismatch.Error()), err: fmt.Errorf("%s", errRPCAPIVersionUnsupported.Error()),
}, },
{ {
expectedErr: errServerTimeMismatch, expectedErr: errServerTimeMismatch,

@ -55,9 +55,8 @@ func createTestStorageServer(t *testing.T) *testStorageRPCServer {
t.Fatalf("unable to initialize storage disks, %s", err) t.Fatalf("unable to initialize storage disks, %s", err)
} }
stServer := &storageServer{ stServer := &storageServer{
storage: storageDisks[0], storage: storageDisks[0],
path: "/disk1", path: "/disk1",
timestamp: UTCNow(),
} }
return &testStorageRPCServer{ return &testStorageRPCServer{
token: token, token: token,
@ -85,7 +84,10 @@ func TestStorageRPCInvalidToken(t *testing.T) {
// Following test cases are meant to exercise the invalid // Following test cases are meant to exercise the invalid
// token code path of the storage RPC methods. // token code path of the storage RPC methods.
var err error var err error
badAuthRPCArgs := AuthRPCArgs{AuthToken: "invalidToken"} badAuthRPCArgs := AuthRPCArgs{
Version: globalRPCAPIVersion,
AuthToken: "invalidToken",
}
badGenericVolArgs := GenericVolArgs{ badGenericVolArgs := GenericVolArgs{
AuthRPCArgs: badAuthRPCArgs, AuthRPCArgs: badAuthRPCArgs,
Vol: "myvol", Vol: "myvol",

@ -42,8 +42,8 @@ var errDataTooSmall = errors.New("Object size smaller than expected")
// errServerNotInitialized - server not initialized. // errServerNotInitialized - server not initialized.
var errServerNotInitialized = errors.New("Server not initialized, please try again") var errServerNotInitialized = errors.New("Server not initialized, please try again")
// errServerVersionMismatch - server versions do not match. // errRPCAPIVersionUnsupported - unsupported rpc API version.
var errServerVersionMismatch = errors.New("Server versions do not match") var errRPCAPIVersionUnsupported = errors.New("Unsupported rpc API version")
// errServerTimeMismatch - server times are too far apart. // errServerTimeMismatch - server times are too far apart.
var errServerTimeMismatch = errors.New("Server times are too far apart") var errServerTimeMismatch = errors.New("Server times are too far apart")

@ -192,7 +192,9 @@ func testStorageInfoWebHandler(obj ObjectLayer, instanceType string, t TestErrHa
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
storageInfoRequest := AuthRPCArgs{} storageInfoRequest := AuthRPCArgs{
Version: globalRPCAPIVersion,
}
storageInfoReply := &StorageInfoRep{} storageInfoReply := &StorageInfoRep{}
req, err := newTestWebRPCRequest("Web.StorageInfo", authorization, storageInfoRequest) req, err := newTestWebRPCRequest("Web.StorageInfo", authorization, storageInfoRequest)
if err != nil { if err != nil {
@ -229,7 +231,9 @@ func testServerInfoWebHandler(obj ObjectLayer, instanceType string, t TestErrHan
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
serverInfoRequest := AuthRPCArgs{} serverInfoRequest := AuthRPCArgs{
Version: globalRPCAPIVersion,
}
serverInfoReply := &ServerInfoRep{} serverInfoReply := &ServerInfoRep{}
req, err := newTestWebRPCRequest("Web.ServerInfo", authorization, serverInfoRequest) req, err := newTestWebRPCRequest("Web.ServerInfo", authorization, serverInfoRequest)
if err != nil { if err != nil {
@ -1493,7 +1497,9 @@ func TestWebCheckAuthorization(t *testing.T) {
"PresignedGet", "PresignedGet",
} }
for _, rpcCall := range webRPCs { for _, rpcCall := range webRPCs {
args := &AuthRPCArgs{} args := &AuthRPCArgs{
Version: globalRPCAPIVersion,
}
reply := &WebGenericRep{} reply := &WebGenericRep{}
req, nerr := newTestWebRPCRequest("Web."+rpcCall, "Bearer fooauthorization", args) req, nerr := newTestWebRPCRequest("Web."+rpcCall, "Bearer fooauthorization", args)
if nerr != nil { if nerr != nil {
@ -1577,7 +1583,9 @@ func TestWebObjectLayerNotReady(t *testing.T) {
webRPCs := []string{"StorageInfo", "MakeBucket", "ListBuckets", "ListObjects", "RemoveObject", webRPCs := []string{"StorageInfo", "MakeBucket", "ListBuckets", "ListObjects", "RemoveObject",
"GetBucketPolicy", "SetBucketPolicy", "ListAllBucketPolicies"} "GetBucketPolicy", "SetBucketPolicy", "ListAllBucketPolicies"}
for _, rpcCall := range webRPCs { for _, rpcCall := range webRPCs {
args := &AuthRPCArgs{} args := &AuthRPCArgs{
Version: globalRPCAPIVersion,
}
reply := &WebGenericRep{} reply := &WebGenericRep{}
req, nerr := newTestWebRPCRequest("Web."+rpcCall, authorization, args) req, nerr := newTestWebRPCRequest("Web."+rpcCall, authorization, args)
if nerr != nil { if nerr != nil {
@ -1690,7 +1698,7 @@ func TestWebObjectLayerFaultyDisks(t *testing.T) {
RepArgs interface{} RepArgs interface{}
}{ }{
{"MakeBucket", MakeBucketArgs{BucketName: bucketName}, WebGenericRep{}}, {"MakeBucket", MakeBucketArgs{BucketName: bucketName}, WebGenericRep{}},
{"ListBuckets", AuthRPCArgs{}, ListBucketsRep{}}, {"ListBuckets", AuthRPCArgs{Version: globalRPCAPIVersion}, ListBucketsRep{}},
{"ListObjects", ListObjectsArgs{BucketName: bucketName, Prefix: ""}, ListObjectsRep{}}, {"ListObjects", ListObjectsArgs{BucketName: bucketName, Prefix: ""}, ListObjectsRep{}},
{"GetBucketPolicy", GetBucketPolicyArgs{BucketName: bucketName, Prefix: ""}, GetBucketPolicyRep{}}, {"GetBucketPolicy", GetBucketPolicyArgs{BucketName: bucketName, Prefix: ""}, GetBucketPolicyRep{}},
{"SetBucketPolicy", SetBucketPolicyArgs{BucketName: bucketName, Prefix: "", Policy: "none"}, WebGenericRep{}}, {"SetBucketPolicy", SetBucketPolicyArgs{BucketName: bucketName, Prefix: "", Policy: "none"}, WebGenericRep{}},
@ -1714,7 +1722,9 @@ func TestWebObjectLayerFaultyDisks(t *testing.T) {
} }
// Test Web.StorageInfo // Test Web.StorageInfo
storageInfoRequest := AuthRPCArgs{} storageInfoRequest := AuthRPCArgs{
Version: globalRPCAPIVersion,
}
storageInfoReply := &StorageInfoRep{} storageInfoReply := &StorageInfoRep{}
req, err := newTestWebRPCRequest("Web.StorageInfo", authorization, storageInfoRequest) req, err := newTestWebRPCRequest("Web.StorageInfo", authorization, storageInfoRequest)
if err != nil { if err != nil {

Loading…
Cancel
Save