-
Notifications
You must be signed in to change notification settings - Fork 432
/
factory.go
117 lines (99 loc) · 3.51 KB
/
factory.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
package kube
import (
"context"
"os"
"time"
"github.com/solo-io/gloo/pkg/bootstrap/leaderelector"
"github.com/solo-io/go-utils/contextutils"
"k8s.io/client-go/rest"
k8sleaderelection "k8s.io/client-go/tools/leaderelection"
"sigs.k8s.io/controller-runtime/pkg/leaderelection"
)
var _ leaderelector.ElectionFactory = new(kubeElectionFactory)
const (
// Define the following values according to the defaults:
// https://github.com/kubernetes/client-go/blob/master/tools/leaderelection/leaderelection.go
defaultLeaseDuration = 15 * time.Second
defaultRetryPeriod = 2 * time.Second
defaultRenewPeriod = 10 * time.Second
leaseDurationEnvName = "LEADER_ELECTION_LEASE_DURATION"
retryPeriodEnvName = "LEADER_ELECTION_RETRY_PERIOD"
renewPeriodEnvName = "LEADER_ELECTION_RENEW_PERIOD"
)
// kubeElectionFactory is the implementation for coordinating leader election using
// the k8s leader election tool: https://github.com/kubernetes/client-go/tree/master/tools/leaderelection
type kubeElectionFactory struct {
restCfg *rest.Config
}
func NewElectionFactory(config *rest.Config) *kubeElectionFactory {
return &kubeElectionFactory{
restCfg: config,
}
}
func (f *kubeElectionFactory) StartElection(ctx context.Context, config *leaderelector.ElectionConfig) (leaderelector.Identity, error) {
elected := make(chan struct{})
identity := leaderelector.NewIdentity(elected)
leOpts := leaderelection.Options{
LeaderElection: true,
LeaderElectionID: config.Id,
LeaderElectionNamespace: config.Namespace,
}
// Create the resource Lock interface necessary for leader election.
// Controller runtime requires an event handler provider, but that package is
// internal so for right now we pass a noop handler.
resourceLock, err := leaderelection.NewResourceLock(f.restCfg, NewNoopProvider(), leOpts)
if err != nil {
return identity, err
}
l, err := k8sleaderelection.NewLeaderElector(
k8sleaderelection.LeaderElectionConfig{
Lock: resourceLock,
LeaseDuration: getLeaseDuration(),
RenewDeadline: getRenewPeriod(),
RetryPeriod: getRetryPeriod(),
Callbacks: k8sleaderelection.LeaderCallbacks{
OnStartedLeading: func(callbackCtx context.Context) {
contextutils.LoggerFrom(callbackCtx).Debug("Started Leading")
close(elected)
config.OnStartedLeading(callbackCtx)
},
OnStoppedLeading: func() {
contextutils.LoggerFrom(ctx).Error("Stopped Leading")
config.OnStoppedLeading()
},
OnNewLeader: func(identity string) {
contextutils.LoggerFrom(ctx).Debugf("New Leader Elected with Identity: %s", identity)
config.OnNewLeader(identity)
},
},
Name: config.Id,
ReleaseOnCancel: true,
},
)
if err != nil {
return identity, err
}
// Start the leader elector process in a goroutine
contextutils.LoggerFrom(ctx).Debug("Starting Kube Leader Election")
go l.Run(ctx)
return identity, nil
}
func getLeaseDuration() time.Duration {
return getDurationFromEnvOrDefault(leaseDurationEnvName, defaultLeaseDuration)
}
func getRenewPeriod() time.Duration {
return getDurationFromEnvOrDefault(renewPeriodEnvName, defaultRenewPeriod)
}
func getRetryPeriod() time.Duration {
return getDurationFromEnvOrDefault(retryPeriodEnvName, defaultRetryPeriod)
}
func getDurationFromEnvOrDefault(envName string, defaultDuration time.Duration) time.Duration {
duration := defaultDuration
durationStr := os.Getenv(envName)
if durationStr != "" {
if dur, err := time.ParseDuration(durationStr); err == nil {
duration = dur
}
}
return duration
}