mirror of
https://github.com/rancher/norman.git
synced 2025-09-24 12:38:49 +00:00
Revert "Make leader election have leader callback"
This reverts commit 807103dc58
.
This commit is contained in:
@@ -3,7 +3,6 @@ package leader
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"os"
|
"os"
|
||||||
"sync"
|
|
||||||
|
|
||||||
"github.com/sirupsen/logrus"
|
"github.com/sirupsen/logrus"
|
||||||
"k8s.io/api/core/v1"
|
"k8s.io/api/core/v1"
|
||||||
@@ -16,37 +15,17 @@ import (
|
|||||||
"k8s.io/kubernetes/pkg/client/leaderelectionconfig"
|
"k8s.io/kubernetes/pkg/client/leaderelectionconfig"
|
||||||
)
|
)
|
||||||
|
|
||||||
type State struct {
|
|
||||||
sync.Mutex
|
|
||||||
identity string
|
|
||||||
leader bool
|
|
||||||
}
|
|
||||||
|
|
||||||
func (l *State) Get() (string, bool) {
|
|
||||||
l.Lock()
|
|
||||||
defer l.Unlock()
|
|
||||||
return l.identity, l.leader
|
|
||||||
}
|
|
||||||
|
|
||||||
func (l *State) Status(identity string, leader bool) {
|
|
||||||
l.Lock()
|
|
||||||
l.identity = identity
|
|
||||||
l.leader = leader
|
|
||||||
l.Unlock()
|
|
||||||
}
|
|
||||||
|
|
||||||
type Callback func(cb context.Context)
|
type Callback func(cb context.Context)
|
||||||
type StatusCallback func(identity string, leader bool)
|
|
||||||
|
|
||||||
func RunOrDie(ctx context.Context, name string, client kubernetes.Interface, cb Callback, status StatusCallback) {
|
func RunOrDie(ctx context.Context, name string, client kubernetes.Interface, cb Callback) {
|
||||||
err := run(ctx, name, client, cb, status)
|
err := run(ctx, name, client, cb)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
logrus.Fatalf("Failed to start leader election for %s", name)
|
logrus.Fatalf("Failed to start leader election for %s", name)
|
||||||
}
|
}
|
||||||
panic("Failed to start leader election for " + name)
|
panic("Failed to start leader election for " + name)
|
||||||
}
|
}
|
||||||
|
|
||||||
func run(ctx context.Context, name string, client kubernetes.Interface, cb Callback, status StatusCallback) error {
|
func run(ctx context.Context, name string, client kubernetes.Interface, cb Callback) error {
|
||||||
id, err := os.Hostname()
|
id, err := os.Hostname()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -70,8 +49,6 @@ func run(ctx context.Context, name string, client kubernetes.Interface, cb Callb
|
|||||||
logrus.Fatalf("error creating leader lock for %s: %v", name, err)
|
logrus.Fatalf("error creating leader lock for %s: %v", name, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
status(id, false)
|
|
||||||
|
|
||||||
leaderelection.RunOrDie(leaderelection.LeaderElectionConfig{
|
leaderelection.RunOrDie(leaderelection.LeaderElectionConfig{
|
||||||
Lock: rl,
|
Lock: rl,
|
||||||
LeaseDuration: le.LeaseDuration.Duration,
|
LeaseDuration: le.LeaseDuration.Duration,
|
||||||
@@ -81,16 +58,12 @@ func run(ctx context.Context, name string, client kubernetes.Interface, cb Callb
|
|||||||
OnStartedLeading: func(stop <-chan struct{}) {
|
OnStartedLeading: func(stop <-chan struct{}) {
|
||||||
subCtx, cancel := context.WithCancel(ctx)
|
subCtx, cancel := context.WithCancel(ctx)
|
||||||
go cb(subCtx)
|
go cb(subCtx)
|
||||||
status(id, true)
|
|
||||||
<-stop
|
<-stop
|
||||||
cancel()
|
cancel()
|
||||||
},
|
},
|
||||||
OnStoppedLeading: func() {
|
OnStoppedLeading: func() {
|
||||||
logrus.Fatalf("leaderelection lost for %s", name)
|
logrus.Fatalf("leaderelection lost for %s", name)
|
||||||
},
|
},
|
||||||
OnNewLeader: func(identity string) {
|
|
||||||
status(identity, identity == id)
|
|
||||||
},
|
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
panic("unreachable")
|
panic("unreachable")
|
||||||
|
Reference in New Issue
Block a user