/* * Minio Cloud Storage, (C) 2019 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 ( "context" "encoding/gob" "errors" "math/rand" "net/http" "time" "github.com/gorilla/mux" "github.com/minio/dsync" "github.com/minio/minio/cmd/logger" xnet "github.com/minio/minio/pkg/net" ) const ( // Lock rpc server endpoint. lockServiceSubPath = "/lock" // Lock maintenance interval. lockMaintenanceInterval = 1 * time.Minute // Lock validity check interval. lockValidityCheckInterval = 2 * time.Minute ) // To abstract a node over network. type lockRESTServer struct { ll localLocker } func (l *lockRESTServer) writeErrorResponse(w http.ResponseWriter, err error) { w.WriteHeader(http.StatusForbidden) w.Write([]byte(err.Error())) } // IsValid - To authenticate and verify the time difference. func (l *lockRESTServer) IsValid(w http.ResponseWriter, r *http.Request) bool { if err := storageServerRequestValidate(r); err != nil { l.writeErrorResponse(w, err) return false } return true } // LockHandler - Acquires a lock. func (l *lockRESTServer) LockHandler(w http.ResponseWriter, r *http.Request) { if !l.IsValid(w, r) { l.writeErrorResponse(w, errors.New("Invalid request")) return } ctx := newContext(r, w, "Lock") var lockArgs dsync.LockArgs if r.ContentLength < 0 { l.writeErrorResponse(w, errInvalidArgument) return } err := gob.NewDecoder(r.Body).Decode(&lockArgs) if err != nil { l.writeErrorResponse(w, err) return } success, err := l.ll.Lock(lockArgs) if err != nil { l.writeErrorResponse(w, err) return } resp := lockResponse{Success: success} logger.LogIf(ctx, gob.NewEncoder(w).Encode(resp)) w.(http.Flusher).Flush() } // UnlockHandler - releases the acquired lock. func (l *lockRESTServer) UnlockHandler(w http.ResponseWriter, r *http.Request) { if !l.IsValid(w, r) { l.writeErrorResponse(w, errors.New("Invalid request")) return } ctx := newContext(r, w, "Unlock") var lockArgs dsync.LockArgs if r.ContentLength < 0 { l.writeErrorResponse(w, errInvalidArgument) return } err := gob.NewDecoder(r.Body).Decode(&lockArgs) if err != nil { l.writeErrorResponse(w, err) return } success, err := l.ll.Unlock(lockArgs) if err != nil { l.writeErrorResponse(w, err) return } resp := lockResponse{Success: success} logger.LogIf(ctx, gob.NewEncoder(w).Encode(resp)) w.(http.Flusher).Flush() } // LockHandler - Acquires an RLock. func (l *lockRESTServer) RLockHandler(w http.ResponseWriter, r *http.Request) { if !l.IsValid(w, r) { l.writeErrorResponse(w, errors.New("Invalid request")) return } ctx := newContext(r, w, "RLock") var lockArgs dsync.LockArgs if r.ContentLength < 0 { l.writeErrorResponse(w, errInvalidArgument) return } err := gob.NewDecoder(r.Body).Decode(&lockArgs) if err != nil { l.writeErrorResponse(w, err) return } success, err := l.ll.RLock(lockArgs) if err != nil { l.writeErrorResponse(w, err) return } resp := lockResponse{Success: success} logger.LogIf(ctx, gob.NewEncoder(w).Encode(resp)) w.(http.Flusher).Flush() } // RUnlockHandler - releases the acquired read lock. func (l *lockRESTServer) RUnlockHandler(w http.ResponseWriter, r *http.Request) { if !l.IsValid(w, r) { l.writeErrorResponse(w, errors.New("Invalid request")) return } ctx := newContext(r, w, "RUnlock") var lockArgs dsync.LockArgs if r.ContentLength < 0 { l.writeErrorResponse(w, errInvalidArgument) return } err := gob.NewDecoder(r.Body).Decode(&lockArgs) if err != nil { l.writeErrorResponse(w, err) return } success, err := l.ll.RUnlock(lockArgs) if err != nil { l.writeErrorResponse(w, err) return } resp := lockResponse{Success: success} logger.LogIf(ctx, gob.NewEncoder(w).Encode(resp)) w.(http.Flusher).Flush() } // ForceUnlockHandler - force releases the acquired lock. func (l *lockRESTServer) ForceUnlockHandler(w http.ResponseWriter, r *http.Request) { if !l.IsValid(w, r) { l.writeErrorResponse(w, errors.New("Invalid request")) return } ctx := newContext(r, w, "ForceUnlock") var lockArgs dsync.LockArgs if r.ContentLength < 0 { l.writeErrorResponse(w, errInvalidArgument) return } err := gob.NewDecoder(r.Body).Decode(&lockArgs) if err != nil { l.writeErrorResponse(w, err) return } success, err := l.ll.ForceUnlock(lockArgs) if err != nil { l.writeErrorResponse(w, err) return } resp := lockResponse{Success: success} logger.LogIf(ctx, gob.NewEncoder(w).Encode(resp)) w.(http.Flusher).Flush() } // ExpiredHandler - query expired lock status. func (l *lockRESTServer) ExpiredHandler(w http.ResponseWriter, r *http.Request) { if !l.IsValid(w, r) { l.writeErrorResponse(w, errors.New("Invalid request")) return } ctx := newContext(r, w, "Expired") var lockArgs dsync.LockArgs if r.ContentLength < 0 { l.writeErrorResponse(w, errInvalidArgument) return } err := gob.NewDecoder(r.Body).Decode(&lockArgs) if err != nil { l.writeErrorResponse(w, err) return } success := true l.ll.mutex.Lock() defer l.ll.mutex.Unlock() // Lock found, proceed to verify if belongs to given uid. if lri, ok := l.ll.lockMap[lockArgs.Resource]; ok { // Check whether uid is still active for _, entry := range lri { if entry.UID == lockArgs.UID { success = false // When uid found, lock is still active so return not expired. break } } } // When we get here lock is no longer active due to either dsync.LockArgs.Resource // being absent from map or uid not found for given dsync.LockArgs.Resource resp := lockResponse{Success: success} logger.LogIf(ctx, gob.NewEncoder(w).Encode(resp)) w.(http.Flusher).Flush() } // lockMaintenance loops over locks that have been active for some time and checks back // with the original server whether it is still alive or not // // Following logic inside ignores the errors generated for Dsync.Active operation. // - server at client down // - some network error (and server is up normally) // // We will ignore the error, and we will retry later to get a resolve on this lock func (l *lockRESTServer) lockMaintenance(interval time.Duration) { l.ll.mutex.Lock() // Get list of long lived locks to check for staleness. nlripLongLived := getLongLivedLocks(l.ll.lockMap, interval) l.ll.mutex.Unlock() // Validate if long lived locks are indeed clean. for _, nlrip := range nlripLongLived { // Initialize client based on the long live locks. host, err := xnet.ParseHost(nlrip.lri.Node) if err != nil { logger.LogIf(context.Background(), err) continue } c := newlockRESTClient(host) if !c.connected { continue } // Call back to original server verify whether the lock is still active (based on name & uid) expired, _ := c.Expired(dsync.LockArgs{ UID: nlrip.lri.UID, Resource: nlrip.name, }) // Close the connection regardless of the call response. c.Close() // For successful response, verify if lock is indeed active or stale. if expired { // The lock is no longer active at server that originated the lock // So remove the lock from the map. l.ll.mutex.Lock() l.ll.removeEntryIfExists(nlrip) // Purge the stale entry if it exists. l.ll.mutex.Unlock() } } } // Start lock maintenance from all lock servers. func startLockMaintenance(lkSrv *lockRESTServer) { // Initialize a new ticker with a minute between each ticks. ticker := time.NewTicker(lockMaintenanceInterval) // Stop the timer upon service closure and cleanup the go-routine. defer ticker.Stop() // 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 2 minutes. select { case <-GlobalServiceDoneCh: return case <-ticker.C: lkSrv.lockMaintenance(lockValidityCheckInterval) } } } // registerLockRESTHandlers - register lock rest router. func registerLockRESTHandlers(router *mux.Router) { subrouter := router.PathPrefix(lockRESTPath).Subrouter() subrouter.Methods(http.MethodPost).Path("/" + lockRESTMethodLock).HandlerFunc(httpTraceHdrs(globalLockServer.LockHandler)) subrouter.Methods(http.MethodPost).Path("/" + lockRESTMethodRLock).HandlerFunc(httpTraceHdrs(globalLockServer.RLockHandler)) subrouter.Methods(http.MethodPost).Path("/" + lockRESTMethodUnlock).HandlerFunc(httpTraceHdrs(globalLockServer.UnlockHandler)) subrouter.Methods(http.MethodPost).Path("/" + lockRESTMethodRUnlock).HandlerFunc(httpTraceHdrs(globalLockServer.RUnlockHandler)) subrouter.Methods(http.MethodPost).Path("/" + lockRESTMethodForceUnlock).HandlerFunc(httpTraceHdrs(globalLockServer.ForceUnlockHandler)) subrouter.Methods(http.MethodPost).Path("/" + lockRESTMethodExpired).HandlerFunc(httpTraceAll(globalLockServer.ExpiredHandler)) router.NotFoundHandler = http.HandlerFunc(httpTraceAll(notFoundHandler)) // Start lock maintenance from all lock servers. go startLockMaintenance(globalLockServer) }