mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-01 20:26:27 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
22cb364770 | ||
|
|
2db45af511 |
@@ -42,6 +42,7 @@ type LiveLock struct {
|
||||
self string
|
||||
lc *LockClient
|
||||
owner string
|
||||
renewInterval time.Duration
|
||||
}
|
||||
|
||||
// NewShortLivedLock creates a lock with a 5-second duration
|
||||
@@ -60,7 +61,11 @@ func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLoc
|
||||
}
|
||||
|
||||
// StartLongLivedLock starts a goroutine to lock the key and returns immediately.
|
||||
func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string)) (lock *LiveLock) {
|
||||
func (lc *LockClient) StartLongLivedLock(key string, owner string, renewInterval time.Duration, onLockOwnerChange func(newLockOwner string)) (lock *LiveLock) {
|
||||
// Default to lock_manager.RenewInterval if zero duration is passed
|
||||
if renewInterval == 0 {
|
||||
renewInterval = lock_manager.RenewInterval
|
||||
}
|
||||
lock = &LiveLock{
|
||||
key: key,
|
||||
hostFiler: lc.seedFiler,
|
||||
@@ -69,6 +74,7 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh
|
||||
grpcDialOption: lc.grpcDialOption,
|
||||
self: owner,
|
||||
lc: lc,
|
||||
renewInterval: renewInterval,
|
||||
}
|
||||
go func() {
|
||||
isLocked := false
|
||||
@@ -104,9 +110,9 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh
|
||||
return
|
||||
default:
|
||||
if isLocked {
|
||||
time.Sleep(5 * lock_manager.RenewInterval)
|
||||
time.Sleep(5 * lock.renewInterval)
|
||||
} else {
|
||||
time.Sleep(lock_manager.RenewInterval)
|
||||
time.Sleep(lock.renewInterval)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster"
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/wdclient"
|
||||
"google.golang.org/grpc"
|
||||
@@ -139,7 +140,7 @@ func NewMessageBroker(option *MessageQueueBrokerOption, grpcDialOption grpc.Dial
|
||||
|
||||
newBrokerBalancerCh := make(chan string, 1)
|
||||
lockClient := cluster.NewLockClient(grpcDialOption, mqBroker.currentFiler)
|
||||
mqBroker.lockAsBalancer = lockClient.StartLongLivedLock(pub_balancer.LockBrokerBalancer, string(self), func(newLockOwner string) {
|
||||
mqBroker.lockAsBalancer = lockClient.StartLongLivedLock(pub_balancer.LockBrokerBalancer, string(self), lock_manager.RenewInterval, func(newLockOwner string) {
|
||||
glog.V(0).Infof("broker %s found balanacer %s", self, newLockOwner)
|
||||
newBrokerBalancerCh <- newLockOwner
|
||||
})
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster"
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer"
|
||||
"github.com/seaweedfs/seaweedfs/weed/filer_client"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
@@ -179,6 +180,7 @@ func (cr *CoordinatorRegistry) startLeaderElection() {
|
||||
cr.leaderLock = cr.lockClient.StartLongLivedLock(
|
||||
GatewayLeaderLockKey,
|
||||
cr.gatewayAddress,
|
||||
lock_manager.RenewInterval,
|
||||
cr.onLeadershipChange,
|
||||
)
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster"
|
||||
"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
@@ -59,7 +60,7 @@ func (s3a *S3ApiServer) startBucketSizeMetricsLoop(ctx context.Context) {
|
||||
owner := string(filer) + "-s3-metrics"
|
||||
|
||||
// Start long-lived lock - this S3 instance will only collect metrics when it holds the lock
|
||||
lock := lockClient.StartLongLivedLock(s3MetricsLockName, owner, func(newLockOwner string) {
|
||||
lock := lockClient.StartLongLivedLock(s3MetricsLockName, owner, lock_manager.RenewInterval, func(newLockOwner string) {
|
||||
glog.V(1).Infof("S3 bucket size metrics lock owner changed to: %s", newLockOwner)
|
||||
})
|
||||
defer lock.Stop()
|
||||
|
||||
Reference in New Issue
Block a user