mirror of
https://github.com/kubernetes/client-go.git
synced 2025-07-04 10:46:16 +00:00
feedback: leasecandidate clients
Kubernetes-commit: fac758164029e278e9bda924090ed078bb6514c8
This commit is contained in:
parent
1f27757b2f
commit
18dd587a4b
@ -161,6 +161,7 @@ type LeaderElectionConfig struct {
|
||||
Name string
|
||||
|
||||
// Coordinated will use the Coordinated Leader Election feature
|
||||
// WARNING: Coordinated leader election is ALPHA.
|
||||
Coordinated bool
|
||||
}
|
||||
|
||||
@ -293,6 +294,7 @@ func (le *LeaderElector) renew(ctx context.Context) {
|
||||
return
|
||||
}
|
||||
le.metrics.leaderOff(le.config.Name)
|
||||
klog.Infof("failed to renew lease %v: %v", desc, err)
|
||||
cancel()
|
||||
}, le.config.RetryPeriod, ctx.Done())
|
||||
|
||||
@ -354,15 +356,15 @@ func (le *LeaderElector) tryCoordinatedRenew(ctx context.Context) bool {
|
||||
|
||||
le.observedRawRecord = oldLeaderElectionRawRecord
|
||||
}
|
||||
hasExpired := le.observedTime.Add(time.Second * time.Duration(oldLeaderElectionRecord.LeaseDurationSeconds)).Before(now.Time)
|
||||
|
||||
hasExpired := le.observedTime.Add(time.Second * time.Duration(oldLeaderElectionRecord.LeaseDurationSeconds)).Before(now.Time)
|
||||
if hasExpired {
|
||||
klog.Infof("lock has expired: %v", le.config.Lock.Describe())
|
||||
return false
|
||||
}
|
||||
|
||||
if !le.IsLeader() {
|
||||
klog.V(4).Infof("lock is held by %v and has not yet expired: %v", oldLeaderElectionRecord.HolderIdentity, le.config.Lock.Describe())
|
||||
klog.V(6).Infof("lock is held by %v and has not yet expired: %v", oldLeaderElectionRecord.HolderIdentity, le.config.Lock.Describe())
|
||||
return false
|
||||
}
|
||||
|
||||
@ -370,7 +372,7 @@ func (le *LeaderElector) tryCoordinatedRenew(ctx context.Context) bool {
|
||||
if le.IsLeader() && oldLeaderElectionRecord.PreferredHolder != "" {
|
||||
klog.V(4).Infof("lock is marked as 'end of term': %v", le.config.Lock.Describe())
|
||||
// TODO: Instead of letting lease expire, the holder may deleted it directly
|
||||
// This will not be compatible with all controllers, so it needs to be opt-in behavior..
|
||||
// This will not be compatible with all controllers, so it needs to be opt-in behavior.
|
||||
// We must ensure all code guarded by this lease has successfully completed
|
||||
// prior to releasing or there may be two processes
|
||||
// simultaneously acting on the critical path.
|
||||
|
@ -63,9 +63,14 @@ type LeaseCandidate struct {
|
||||
preferredStrategies []v1.CoordinatedLeaseStrategy
|
||||
}
|
||||
|
||||
// NewCandidate creates new LeaseCandidate controller that creates a
|
||||
// LeaseCandidate object if it does not exist and watches changes
|
||||
// to the corresponding object and renews if PingTime is set.
|
||||
// WARNING: This is an ALPHA feature. Ensure that the CoordinatedLeaderElection
|
||||
// feature gate is on.
|
||||
func NewCandidate(clientset kubernetes.Interface,
|
||||
candidateName string,
|
||||
candidateNamespace string,
|
||||
candidateName string,
|
||||
targetLease string,
|
||||
binaryVersion, emulationVersion string,
|
||||
preferredStrategies []v1.CoordinatedLeaseStrategy,
|
||||
@ -144,7 +149,6 @@ func (c *LeaseCandidate) processNextWorkItem(ctx context.Context) bool {
|
||||
}
|
||||
|
||||
utilruntime.HandleError(err)
|
||||
klog.Infof("processNextWorkItem.AddRateLimited: %v", key)
|
||||
c.queue.AddRateLimited(key)
|
||||
|
||||
return true
|
||||
@ -161,9 +165,8 @@ func (c *LeaseCandidate) ensureLease(ctx context.Context) error {
|
||||
if apierrors.IsNotFound(err) {
|
||||
klog.V(2).Infof("Creating lease candidate")
|
||||
// lease does not exist, create it.
|
||||
leaseToCreate := c.newLease()
|
||||
_, err := c.leaseClient.Create(ctx, leaseToCreate, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
leaseToCreate := c.newLeaseCandidate()
|
||||
if _, err := c.leaseClient.Create(ctx, leaseToCreate, metav1.CreateOptions{}); err != nil {
|
||||
return err
|
||||
}
|
||||
klog.V(2).Infof("Created lease candidate")
|
||||
@ -171,7 +174,7 @@ func (c *LeaseCandidate) ensureLease(ctx context.Context) error {
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
klog.V(2).Infof("lease candidate exists.. renewing")
|
||||
klog.V(2).Infof("lease candidate exists. Renewing.")
|
||||
clone := lease.DeepCopy()
|
||||
clone.Spec.RenewTime = &metav1.MicroTime{Time: c.clock.Now()}
|
||||
clone.Spec.PingTime = nil
|
||||
@ -182,8 +185,8 @@ func (c *LeaseCandidate) ensureLease(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *LeaseCandidate) newLease() *v1alpha1.LeaseCandidate {
|
||||
lease := &v1alpha1.LeaseCandidate{
|
||||
func (c *LeaseCandidate) newLeaseCandidate() *v1alpha1.LeaseCandidate {
|
||||
lc := &v1alpha1.LeaseCandidate{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: c.name,
|
||||
Namespace: c.namespace,
|
||||
@ -195,6 +198,6 @@ func (c *LeaseCandidate) newLease() *v1alpha1.LeaseCandidate {
|
||||
PreferredStrategies: c.preferredStrategies,
|
||||
},
|
||||
}
|
||||
lease.Spec.RenewTime = &metav1.MicroTime{Time: c.clock.Now()}
|
||||
return lease
|
||||
lc.Spec.RenewTime = &metav1.MicroTime{Time: c.clock.Now()}
|
||||
return lc
|
||||
}
|
||||
|
@ -48,8 +48,8 @@ func TestLeaseCandidateCreation(t *testing.T) {
|
||||
client := fake.NewSimpleClientset()
|
||||
candidate, _, err := NewCandidate(
|
||||
client,
|
||||
tc.candidateName,
|
||||
tc.candidateNamespace,
|
||||
tc.candidateName,
|
||||
tc.leaseName,
|
||||
tc.binaryVersion,
|
||||
tc.emulationVersion,
|
||||
|
Loading…
Reference in New Issue
Block a user