Method Run

pkg/infra/leaderelection/leader_election.go:139–193

Run participates in leader election and calls fn when leadership is acquired. fn receives a context that is cancelled when leadership is lost. Run blocks until ctx is cancelled.

(ctx context.Context, fn func(ctx context.Context), opts ...RunOption)

// fn receives a context that is cancelled when leadership is lost.
// Run blocks until ctx is cancelled.
func (k *KubernetesElector) Run(ctx context.Context, fn func(ctx context.Context), opts ...RunOption) error {
	o := ResolveRunOptions([]RunOption{
		WithReleaseOnCancel(true),
		WithOnStartedLeading(func(ctx context.Context) {
			k.logger.Info("Acquired leader lease, starting leader work",
				"identity", k.identity,
				"lease", k.leaseName,
				"namespace", k.namespace,
			)
		}),
		WithOnStoppedLeading(func() {
			k.logger.Info("Lost leader lease, stopping leader work",
				"identity", k.identity,
			)
		}),
		WithOnNewLeader(func(identity string) {
			if identity != k.identity {
				k.logger.Info("New leader elected", "leader", identity)
			}
		}),
	}, opts)

	lock := &resourcelock.LeaseLock{
		LeaseMeta: metav1.ObjectMeta{
			Name:      k.leaseName,
			Namespace: k.namespace,
		},
		Client: k.kubeClient.CoordinationV1(),
		LockConfig: resourcelock.ResourceLockConfig{
			Identity: k.identity,
		},
	}

	le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
		Lock:            lock,
		LeaseDuration:   k.leaseDuration,
		RenewDeadline:   k.renewDeadline,
		RetryPeriod:     k.retryPeriod,
		ReleaseOnCancel: o.ReleaseOnCancel,
		Callbacks: leaderelection.LeaderCallbacks{
			OnStartedLeading: func(ctx context.Context) {
				o.OnStartedLeading(ctx)
				fn(ctx)
			},
			OnStoppedLeading: o.OnStoppedLeading,
			OnNewLeader:      o.OnNewLeader,
		},
	})
	if err != nil {
		return fmt.Errorf("failed to create leader elector: %w", err)
	}

	le.Run(ctx)
	return ctx.Err()
}