master_grpc_server_admin.go 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195
  1. package weed_server
  2. import (
  3. "context"
  4. "fmt"
  5. "github.com/seaweedfs/seaweedfs/weed/stats"
  6. "math/rand/v2"
  7. "sync"
  8. "time"
  9. "github.com/seaweedfs/raft"
  10. "github.com/seaweedfs/seaweedfs/weed/cluster"
  11. "github.com/seaweedfs/seaweedfs/weed/glog"
  12. "github.com/seaweedfs/seaweedfs/weed/pb"
  13. "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
  14. "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
  15. "github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
  16. )
  17. /*
  18. How exclusive lock works?
  19. -----------
  20. Shell
  21. ------
  22. When shell lock,
  23. * lease an admin token (lockTime, token)
  24. * start a goroutine to renew the admin token periodically
  25. When shell unlock
  26. * stop the renewal goroutine
  27. * sends a release lock request
  28. Master
  29. ------
  30. Master maintains:
  31. * randomNumber
  32. * lastLockTime
  33. When master receives the lease/renew request from shell
  34. If lastLockTime still fresh {
  35. if is a renew and token is valid {
  36. // for renew
  37. generate the randomNumber => token
  38. return
  39. }
  40. refuse
  41. return
  42. } else {
  43. // for fresh lease request
  44. generate the randomNumber => token
  45. return
  46. }
  47. When master receives the release lock request from shell
  48. set the lastLockTime to zero
  49. The volume server does not need to verify.
  50. This makes the lock/unlock optional, similar to what golang code usually does.
  51. */
  52. const (
  53. LockDuration = 10 * time.Second
  54. )
  55. type AdminLock struct {
  56. accessSecret int64
  57. accessLockTime time.Time
  58. lastClient string
  59. lastMessage string
  60. }
  61. type AdminLocks struct {
  62. locks map[string]*AdminLock
  63. sync.RWMutex
  64. }
  65. func NewAdminLocks() *AdminLocks {
  66. return &AdminLocks{
  67. locks: make(map[string]*AdminLock),
  68. }
  69. }
  70. func (locks *AdminLocks) isLocked(lockName string) (clientName string, message string, isLocked bool) {
  71. locks.RLock()
  72. defer locks.RUnlock()
  73. adminLock, found := locks.locks[lockName]
  74. if !found {
  75. return "", "", false
  76. }
  77. glog.V(4).Infof("isLocked %v: %v", adminLock.lastClient, adminLock.lastMessage)
  78. return adminLock.lastClient, adminLock.lastMessage, adminLock.accessLockTime.Add(LockDuration).After(time.Now())
  79. }
  80. func (locks *AdminLocks) isValidToken(lockName string, ts time.Time, token int64) bool {
  81. locks.RLock()
  82. defer locks.RUnlock()
  83. adminLock, found := locks.locks[lockName]
  84. if !found {
  85. return false
  86. }
  87. return adminLock.accessLockTime.Equal(ts) && adminLock.accessSecret == token
  88. }
  89. func (locks *AdminLocks) generateToken(lockName string, clientName string) (ts time.Time, token int64) {
  90. locks.Lock()
  91. defer locks.Unlock()
  92. lock := &AdminLock{
  93. accessSecret: rand.Int64(),
  94. accessLockTime: time.Now(),
  95. lastClient: clientName,
  96. }
  97. locks.locks[lockName] = lock
  98. stats.MasterAdminLock.WithLabelValues(clientName).Set(1)
  99. return lock.accessLockTime, lock.accessSecret
  100. }
  101. func (locks *AdminLocks) deleteLock(lockName string) {
  102. locks.Lock()
  103. stats.MasterAdminLock.WithLabelValues(locks.locks[lockName].lastClient).Set(0)
  104. defer locks.Unlock()
  105. delete(locks.locks, lockName)
  106. }
  107. func (ms *MasterServer) LeaseAdminToken(ctx context.Context, req *master_pb.LeaseAdminTokenRequest) (*master_pb.LeaseAdminTokenResponse, error) {
  108. resp := &master_pb.LeaseAdminTokenResponse{}
  109. if !ms.Topo.IsLeader() {
  110. return resp, raft.NotLeaderError
  111. }
  112. if lastClient, lastMessage, isLocked := ms.adminLocks.isLocked(req.LockName); isLocked {
  113. glog.V(4).Infof("LeaseAdminToken %v", lastClient)
  114. if req.PreviousToken != 0 && ms.adminLocks.isValidToken(req.LockName, time.Unix(0, req.PreviousLockTime), req.PreviousToken) {
  115. // for renew
  116. ts, token := ms.adminLocks.generateToken(req.LockName, req.ClientName)
  117. resp.Token, resp.LockTsNs = token, ts.UnixNano()
  118. return resp, nil
  119. }
  120. // refuse since still locked
  121. return resp, fmt.Errorf("already locked by %v: %v", lastClient, lastMessage)
  122. }
  123. // for fresh lease request
  124. ts, token := ms.adminLocks.generateToken(req.LockName, req.ClientName)
  125. resp.Token, resp.LockTsNs = token, ts.UnixNano()
  126. return resp, nil
  127. }
  128. func (ms *MasterServer) ReleaseAdminToken(ctx context.Context, req *master_pb.ReleaseAdminTokenRequest) (*master_pb.ReleaseAdminTokenResponse, error) {
  129. resp := &master_pb.ReleaseAdminTokenResponse{}
  130. if ms.adminLocks.isValidToken(req.LockName, time.Unix(0, req.PreviousLockTime), req.PreviousToken) {
  131. ms.adminLocks.deleteLock(req.LockName)
  132. }
  133. return resp, nil
  134. }
  135. func (ms *MasterServer) Ping(ctx context.Context, req *master_pb.PingRequest) (resp *master_pb.PingResponse, pingErr error) {
  136. resp = &master_pb.PingResponse{
  137. StartTimeNs: time.Now().UnixNano(),
  138. }
  139. if req.TargetType == cluster.FilerType {
  140. pingErr = pb.WithFilerClient(false, 0, pb.ServerAddress(req.Target), ms.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  141. pingResp, err := client.Ping(ctx, &filer_pb.PingRequest{})
  142. if pingResp != nil {
  143. resp.RemoteTimeNs = pingResp.StartTimeNs
  144. }
  145. return err
  146. })
  147. }
  148. if req.TargetType == cluster.VolumeServerType {
  149. pingErr = pb.WithVolumeServerClient(false, pb.ServerAddress(req.Target), ms.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
  150. pingResp, err := client.Ping(ctx, &volume_server_pb.PingRequest{})
  151. if pingResp != nil {
  152. resp.RemoteTimeNs = pingResp.StartTimeNs
  153. }
  154. return err
  155. })
  156. }
  157. if req.TargetType == cluster.MasterType {
  158. pingErr = pb.WithMasterClient(false, pb.ServerAddress(req.Target), ms.grpcDialOption, false, func(client master_pb.SeaweedClient) error {
  159. pingResp, err := client.Ping(ctx, &master_pb.PingRequest{})
  160. if pingResp != nil {
  161. resp.RemoteTimeNs = pingResp.StartTimeNs
  162. }
  163. return err
  164. })
  165. }
  166. if pingErr != nil {
  167. pingErr = fmt.Errorf("ping %s %s: %v", req.TargetType, req.Target, pingErr)
  168. }
  169. resp.StopTimeNs = time.Now().UnixNano()
  170. return
  171. }