Adopt dsync interface changes and major cleanup on RPC server/client.

* Rename GenericArgs to AuthRPCArgs * Rename GenericReply to AuthRPCReply * Remove authConfig.loginMethod and add authConfig.ServiceName * Rename loginServer to AuthRPCServer * Rename RPCLoginArgs to LoginRPCArgs * Rename RPCLoginReply to LoginRPCReply * Version and RequestTime are added to LoginRPCArgs and verified by server side, not client side. * Fix data race in lockMaintainence loop.
2026-07-20 04:30:26 +03:00 · 2016-12-23 20:42:19 +05:30
parent cde6496172
commit 6d10f4c19a
39 changed files with 1083 additions and 949 deletions
@@ -25,32 +25,19 @@ import (
 	"time"

 	router "github.com/gorilla/mux"
+	"github.com/minio/dsync"
 )

-const lockRPCPath = "/minio/lock"
-const lockMaintenanceLoop = 1 * time.Minute
-const lockCheckValidityInterval = 2 * time.Minute
+const (
+	// Lock rpc server endpoint.
+	lockRPCPath = "/minio/lock"

-// LockArgs besides lock name, holds Token and Timestamp for session
-// authentication and validation server restart.
-type LockArgs struct {
-	Name      string
-	Token     string
-	Timestamp time.Time
-	Node      string
-	RPCPath   string
-	UID       string
-}
+	// Lock maintenance interval.
+	lockMaintenanceInterval = 1 * time.Minute // 1 minute.

-// SetToken - sets the token to the supplied value.
-func (l *LockArgs) SetToken(token string) {
-	l.Token = token
-}
-
-// SetTimestamp - sets the timestamp to the supplied value.
-func (l *LockArgs) SetTimestamp(tstamp time.Time) {
-	l.Timestamp = tstamp
-}
+	// Lock validity check interval.
+	lockValidityCheckInterval = 2 * time.Minute // 2 minutes.
+)

 // lockRequesterInfo stores various info from the client for each lock that is requested
 type lockRequesterInfo struct {
@@ -69,43 +56,61 @@ func isWriteLock(lri []lockRequesterInfo) bool {

 // lockServer is type for RPC handlers
 type lockServer struct {
-	loginServer
+	AuthRPCServer
 	rpcPath string
 	mutex   sync.Mutex
 	lockMap map[string][]lockRequesterInfo
 }

+// Start lock maintenance from all lock servers.
+func startLockMaintainence(lockServers []*lockServer) {
+	for _, locker := range lockServers {
+		// Start loop for stale lock maintenance
+		go func(lk *lockServer) {
+			// Initialize a new ticker with a minute between each ticks.
+			ticker := time.NewTicker(lockMaintenanceInterval)
+
+			// Start with random sleep time, so as to avoid "synchronous checks" between servers
+			time.Sleep(time.Duration(rand.Float64() * float64(lockMaintenanceInterval)))
+			for {
+				// Verifies every minute for locks held more than 2minutes.
+				select {
+				case <-ticker.C:
+					lk.lockMaintenance(lockValidityCheckInterval)
+				case <-globalServiceDoneCh:
+					// Stop the timer.
+					ticker.Stop()
+				}
+			}
+		}(locker)
+	}
+}
+
 // Register distributed NS lock handlers.
 func registerDistNSLockRouter(mux *router.Router, serverConfig serverCmdConfig) error {
+	// Initialize a new set of lock servers.
 	lockServers := newLockServers(serverConfig)
+
+	// Start lock maintenance from all lock servers.
+	startLockMaintainence(lockServers)
+
+	// Register initialized lock servers to their respective rpc endpoints.
 	return registerStorageLockers(mux, lockServers)
 }

 // Create one lock server for every local storage rpc server.
 func newLockServers(srvConfig serverCmdConfig) (lockServers []*lockServer) {
 	for _, ep := range srvConfig.endpoints {
-		// Not local storage move to the next node.
-		if !isLocalStorage(ep) {
-			continue
-		}
-
-		// Create handler for lock RPCs
-		locker := &lockServer{
-			rpcPath: getPath(ep),
-			mutex:   sync.Mutex{},
-			lockMap: make(map[string][]lockRequesterInfo),
-		}
-
-		// Start loop for stale lock maintenance
-		go func() {
-			// Start with random sleep time, so as to avoid "synchronous checks" between servers
-			time.Sleep(time.Duration(rand.Float64() * float64(lockMaintenanceLoop)))
-			for {
-				time.Sleep(lockMaintenanceLoop)
-				locker.lockMaintenance(lockCheckValidityInterval)
+		// Initialize new lock server for each local node.
+		if isLocalStorage(ep) {
+			// Create handler for lock RPCs
+			locker := &lockServer{
+				rpcPath: getPath(ep),
+				mutex:   sync.Mutex{},
+				lockMap: make(map[string][]lockRequesterInfo),
 			}
-		}()
-		lockServers = append(lockServers, locker)
+			lockServers = append(lockServers, locker)
+		}
 	}
 	return lockServers
 }
@@ -114,8 +119,7 @@ func newLockServers(srvConfig serverCmdConfig) (lockServers []*lockServer) {
 func registerStorageLockers(mux *router.Router, lockServers []*lockServer) error {
 	for _, lockServer := range lockServers {
 		lockRPCServer := rpc.NewServer()
-		err := lockRPCServer.RegisterName("Dsync", lockServer)
-		if err != nil {
+		if err := lockRPCServer.RegisterName("Dsync", lockServer); err != nil {
 			return traceError(err)
 		}
 		lockRouter := mux.PathPrefix(reservedBucket).Subrouter()
@@ -130,17 +134,17 @@ func registerStorageLockers(mux *router.Router, lockServers []*lockServer) error
 func (l *lockServer) Lock(args *LockArgs, reply *bool) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
-	if err := l.validateLockArgs(args); err != nil {
+	if err := args.IsAuthenticated(); err != nil {
 		return err
 	}
-	_, *reply = l.lockMap[args.Name]
+	_, *reply = l.lockMap[args.dsyncLockArgs.Resource]
 	if !*reply { // No locks held on the given name, so claim write lock
-		l.lockMap[args.Name] = []lockRequesterInfo{
+		l.lockMap[args.dsyncLockArgs.Resource] = []lockRequesterInfo{
 			{
 				writer:        true,
-				node:          args.Node,
-				rpcPath:       args.RPCPath,
-				uid:           args.UID,
+				node:          args.dsyncLockArgs.ServerAddr,
+				rpcPath:       args.dsyncLockArgs.ServiceEndpoint,
+				uid:           args.dsyncLockArgs.UID,
 				timestamp:     time.Now().UTC(),
 				timeLastCheck: time.Now().UTC(),
 			},
@@ -154,18 +158,18 @@ func (l *lockServer) Lock(args *LockArgs, reply *bool) error {
 func (l *lockServer) Unlock(args *LockArgs, reply *bool) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
-	if err := l.validateLockArgs(args); err != nil {
+	if err := args.IsAuthenticated(); err != nil {
 		return err
 	}
 	var lri []lockRequesterInfo
-	if lri, *reply = l.lockMap[args.Name]; !*reply { // No lock is held on the given name
-		return fmt.Errorf("Unlock attempted on an unlocked entity: %s", args.Name)
+	if lri, *reply = l.lockMap[args.dsyncLockArgs.Resource]; !*reply { // No lock is held on the given name
+		return fmt.Errorf("Unlock attempted on an unlocked entity: %s", args.dsyncLockArgs.Resource)
 	}
 	if *reply = isWriteLock(lri); !*reply { // Unless it is a write lock
-		return fmt.Errorf("Unlock attempted on a read locked entity: %s (%d read locks active)", args.Name, len(lri))
+		return fmt.Errorf("Unlock attempted on a read locked entity: %s (%d read locks active)", args.dsyncLockArgs.Resource, len(lri))
 	}
-	if !l.removeEntry(args.Name, args.UID, &lri) {
-		return fmt.Errorf("Unlock unable to find corresponding lock for uid: %s", args.UID)
+	if !l.removeEntry(args.dsyncLockArgs.Resource, args.dsyncLockArgs.UID, &lri) {
+		return fmt.Errorf("Unlock unable to find corresponding lock for uid: %s", args.dsyncLockArgs.UID)
 	}
 	return nil
 }
@@ -174,23 +178,23 @@ func (l *lockServer) Unlock(args *LockArgs, reply *bool) error {
 func (l *lockServer) RLock(args *LockArgs, reply *bool) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
-	if err := l.validateLockArgs(args); err != nil {
+	if err := args.IsAuthenticated(); err != nil {
 		return err
 	}
 	lrInfo := lockRequesterInfo{
 		writer:        false,
-		node:          args.Node,
-		rpcPath:       args.RPCPath,
-		uid:           args.UID,
+		node:          args.dsyncLockArgs.ServerAddr,
+		rpcPath:       args.dsyncLockArgs.ServiceEndpoint,
+		uid:           args.dsyncLockArgs.UID,
 		timestamp:     time.Now().UTC(),
 		timeLastCheck: time.Now().UTC(),
 	}
-	if lri, ok := l.lockMap[args.Name]; ok {
+	if lri, ok := l.lockMap[args.dsyncLockArgs.Resource]; ok {
 		if *reply = !isWriteLock(lri); *reply { // Unless there is a write lock
-			l.lockMap[args.Name] = append(l.lockMap[args.Name], lrInfo)
+			l.lockMap[args.dsyncLockArgs.Resource] = append(l.lockMap[args.dsyncLockArgs.Resource], lrInfo)
 		}
 	} else { // No locks held on the given name, so claim (first) read lock
-		l.lockMap[args.Name] = []lockRequesterInfo{lrInfo}
+		l.lockMap[args.dsyncLockArgs.Resource] = []lockRequesterInfo{lrInfo}
 		*reply = true
 	}
 	return nil
@@ -200,18 +204,18 @@ func (l *lockServer) RLock(args *LockArgs, reply *bool) error {
 func (l *lockServer) RUnlock(args *LockArgs, reply *bool) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
-	if err := l.validateLockArgs(args); err != nil {
+	if err := args.IsAuthenticated(); err != nil {
 		return err
 	}
 	var lri []lockRequesterInfo
-	if lri, *reply = l.lockMap[args.Name]; !*reply { // No lock is held on the given name
-		return fmt.Errorf("RUnlock attempted on an unlocked entity: %s", args.Name)
+	if lri, *reply = l.lockMap[args.dsyncLockArgs.Resource]; !*reply { // No lock is held on the given name
+		return fmt.Errorf("RUnlock attempted on an unlocked entity: %s", args.dsyncLockArgs.Resource)
 	}
 	if *reply = !isWriteLock(lri); !*reply { // A write-lock is held, cannot release a read lock
-		return fmt.Errorf("RUnlock attempted on a write locked entity: %s", args.Name)
+		return fmt.Errorf("RUnlock attempted on a write locked entity: %s", args.dsyncLockArgs.Resource)
 	}
-	if !l.removeEntry(args.Name, args.UID, &lri) {
-		return fmt.Errorf("RUnlock unable to find corresponding read lock for uid: %s", args.UID)
+	if !l.removeEntry(args.dsyncLockArgs.Resource, args.dsyncLockArgs.UID, &lri) {
+		return fmt.Errorf("RUnlock unable to find corresponding read lock for uid: %s", args.dsyncLockArgs.UID)
 	}
 	return nil
 }
@@ -220,14 +224,14 @@ func (l *lockServer) RUnlock(args *LockArgs, reply *bool) error {
 func (l *lockServer) ForceUnlock(args *LockArgs, reply *bool) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
-	if err := l.validateLockArgs(args); err != nil {
+	if err := args.IsAuthenticated(); err != nil {
 		return err
 	}
-	if len(args.UID) != 0 {
-		return fmt.Errorf("ForceUnlock called with non-empty UID: %s", args.UID)
+	if len(args.dsyncLockArgs.UID) != 0 {
+		return fmt.Errorf("ForceUnlock called with non-empty UID: %s", args.dsyncLockArgs.UID)
 	}
-	if _, ok := l.lockMap[args.Name]; ok { // Only clear lock when set
-		delete(l.lockMap, args.Name) // Remove the lock (irrespective of write or read lock)
+	if _, ok := l.lockMap[args.dsyncLockArgs.Resource]; ok { // Only clear lock when set
+		delete(l.lockMap, args.dsyncLockArgs.Resource) // Remove the lock (irrespective of write or read lock)
 	}
 	*reply = true
 	return nil
@@ -237,21 +241,21 @@ func (l *lockServer) ForceUnlock(args *LockArgs, reply *bool) error {
 func (l *lockServer) Expired(args *LockArgs, reply *bool) error {
 	l.mutex.Lock()
 	defer l.mutex.Unlock()
-	if err := l.validateLockArgs(args); err != nil {
+	if err := args.IsAuthenticated(); err != nil {
 		return err
 	}
 	// Lock found, proceed to verify if belongs to given uid.
-	if lri, ok := l.lockMap[args.Name]; ok {
+	if lri, ok := l.lockMap[args.dsyncLockArgs.Resource]; ok {
 		// Check whether uid is still active
 		for _, entry := range lri {
-			if entry.uid == args.UID {
+			if entry.uid == args.dsyncLockArgs.UID {
 				*reply = false // When uid found, lock is still active so return not expired.
 				return nil     // When uid found *reply is set to true.
 			}
 		}
 	}
-	// When we get here lock is no longer active due to either args.Name
-	// being absent from map or uid not found for given args.Name
+	// When we get here lock is no longer active due to either args.dsyncLockArgs.Resource
+	// being absent from map or uid not found for given args.dsyncLockArgs.Resource
 	*reply = true
 	return nil
 }
@@ -276,19 +280,24 @@ func (l *lockServer) lockMaintenance(interval time.Duration) {
 	nlripLongLived := getLongLivedLocks(l.lockMap, interval)
 	l.mutex.Unlock()

+	serverCred := serverConfig.GetCredential()
 	// Validate if long lived locks are indeed clean.
 	for _, nlrip := range nlripLongLived {
 		// Initialize client based on the long live locks.
-		c := newRPCClient(nlrip.lri.node, nlrip.lri.rpcPath, isSSL())
-
-		var expired bool
+		c := newLockRPCClient(authConfig{
+			accessKey:       serverCred.AccessKey,
+			secretKey:       serverCred.SecretKey,
+			serverAddr:      nlrip.lri.node,
+			serviceEndpoint: nlrip.lri.rpcPath,
+			secureConn:      isSSL(),
+			serviceName:     "Dsync",
+		})

 		// Call back to original server verify whether the lock is still active (based on name & uid)
-		c.Call("Dsync.Expired", &LockArgs{
-			Name: nlrip.name,
-			UID:  nlrip.lri.uid,
-		}, &expired)
-		c.Close() // Close the connection regardless of the call response.
+		expired, _ := c.Expired(dsync.LockArgs{UID: nlrip.lri.uid, Resource: nlrip.name})
+
+		// Close the connection regardless of the call response.
+		c.rpcClient.Close()

 		// For successful response, verify if lock is indeed active or stale.
 		if expired {