Leader Election Pattern

Overview

The Leader Election Pattern ensures that exactly one instance in a distributed system is the active leader at any time. Other instances remain as followers and take over only if the leader fails. This pattern prevents split-brain situations, duplicate work, and conflicting writes when several nodes could perform the same task.

It is commonly used in distributed schedulers, cluster managers, and stateful services where a single coordinator simplifies coordination.

When to Use

Use this pattern when:

• Multiple nodes could perform the same operation but only one should do it
• You need to coordinate shared resources such as locks, queues, or writes
• A service must have a single source of truth for configuration or scheduling
• You want to avoid split-brain or race conditions in a cluster
• You can tolerate a short failover delay while a new leader is elected

Solution

# Simplified leader election using a lease in a shared database
import time
import uuid
from datetime import datetime, timedelta

class LeaderElection:
    def __init__(self, db, lease_seconds=10):
        self.db = db
        self.node_id = uuid.uuid4().hex
        self.lease_seconds = lease_seconds

    def is_leader(self):
        leader = self.db.get('leader')
        if leader and leader['node_id'] == self.node_id and leader['expires'] > datetime.utcnow():
            return True
        return False

    def try_acquire(self):
        now = datetime.utcnow()
        leader = self.db.get('leader')
        if not leader or leader['expires'] < now:
            self.db.set('leader', {'node_id': self.node_id, 'expires': now + timedelta(seconds=self.lease_seconds)})
            return True
        return False

    def renew(self):
        if self.is_leader():
            self.db.set('leader', {'node_id': self.node_id, 'expires': datetime.utcnow() + timedelta(seconds=self.lease_seconds)})

# Kubernetes leader election with a Lease resource
kubectl create lease app-leader --holder=node-1 --lease-duration=15s

Explanation

Leader election works by having all candidates compete for a shared lock or lease. The node that successfully acquires the lease becomes the leader and must renew it periodically. If the leader stops renewing, the lease expires and other candidates can claim it. The system uses a fencing token or unique node identifier to ensure that an old leader cannot act after it has lost leadership.

A typical leader election mechanism includes:

• Lease acquisition: a node writes a unique identifier into a shared store with a timeout
• Heartbeats: the leader renews the lease before it expires
• Failure detection: followers watch the lease and detect expiration
• Failover: a follower acquires the lease and assumes leadership

Variants

Variant	Coordination Store	Trade-off
Database Lease	PostgreSQL, MySQL	Simple but depends on DB availability
Distributed Lock	Redis Redlock	Fast but vulnerable to clock drift
Consensus Algorithm	etcd, ZooKeeper	Strong consistency but more complex
Kubernetes Lease	API server Lease object	Native for K8s workloads, easy integration

What Works

• Use a short lease with automatic renewal to detect failures quickly
• Generate a fencing token per leadership term to prevent stale leaders from acting
• Ensure the leader gracefully steps down on shutdown
• Watch the lease from followers rather than polling aggressively
• Keep leader responsibilities idempotent where possible
• Log leadership changes clearly for operational visibility

Common Mistakes

• Allowing a former leader to perform writes after losing the lease
• Using a lease duration that is too long, delaying failover
• Not handling network partitions correctly, causing split-brain
• Implementing leader election without a strong fencing token
• Forgetting to release the lease on graceful shutdown

Advanced Solutions

Leader election with etcd using Raft consensus

Implement leader election using etcd’s built-in consensus:

import etcd3

class EtcdLeaderElection:
    def __init__(self, etcd_client, election_key='leader', lease_ttl=10):
        self.etcd = etcd_client
        self.election_key = election_key
        self.lease_ttl = lease_ttl
        self.lease = None
        self.election = None

    async def campaign(self, node_id):
        # Create a lease with TTL
        self.lease = await self.etcd.lease(self.lease_ttl)
        
        # Campaign for leadership
        self.election = self.etcd.election(self.election_key)
        await self.election.campaign(node_id, lease=self.lease)
        
        # Keep lease alive
        while True:
            await self.lease.refresh()
            await asyncio.sleep(self.lease_ttl / 2)

    async def observe(self):
        # Observe current leader
        election = self.etcd.election(self.election_key)
        async for leader in election.observe():
            print(f"Current leader: {leader}")

    async def resign(self):
        if self.election:
            await self.election.resign()
        if self.lease:
            await self.lease.revoke()

Kubernetes leader election with Lease resource

Use Kubernetes Lease API for native leader election:

package main

import (
	"context"
	"fmt"
	"time"

	coordinationv1 "k8s.io/api/coordination/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
)

type LeaseLeaderElection struct {
	clientset     *kubernetes.Clientset
	leaseName     string
	leaseNamespace string
	holderIdentity string
	leaseDuration  time.Duration
}

func NewLeaseLeaderElection(leaseName, namespace, holderIdentity string, duration time.Duration) (*LeaseLeaderElection, error) {
	config, err := rest.InClusterConfig()
	if err != nil {
		return nil, err
	}

	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		return nil, err
	}

	return &LeaseLeaderElection{
		clientset:      clientset,
		leaseName:      leaseName,
		leaseNamespace: namespace,
		holderIdentity: holderIdentity,
		leaseDuration:  duration,
	}, nil
}

func (l *LeaseLeaderElection) Acquire(ctx context.Context) error {
	lease := &coordinationv1.Lease{
		ObjectMeta: metav1.ObjectMeta{
			Name:      l.leaseName,
			Namespace: l.leaseNamespace,
		},
		Spec: coordinationv1.LeaseSpec{
			HolderIdentity:       &l.holderIdentity,
			LeaseDurationSeconds: pointerToInt32(int32(l.leaseDuration.Seconds())),
			AcquireTime:          &metav1.Time{Time: time.Now()},
			RenewTime:            &metav1.Time{Time: time.Now()},
		},
	}

	for {
		current, err := l.clientset.CoordinationV1().Leases(l.leaseNamespace).Get(ctx, l.leaseName, metav1.GetOptions{})
		if err != nil {
			// Lease doesn't exist, try to create
			_, err = l.clientset.CoordinationV1().Leases(l.leaseNamespace).Create(ctx, lease, metav1.CreateOptions{})
			if err == nil {
				return nil // Acquired leadership
			}
			time.Sleep(time.Second)
			continue
		}

		// Check if lease is expired or we are the holder
		if current.Spec.HolderIdentity == nil || *current.Spec.HolderIdentity == l.holderIdentity {
			// Renew lease
			current.Spec.HolderIdentity = &l.holderIdentity
			current.Spec.RenewTime = &metav1.Time{Time: time.Now()}
			_, err = l.clientset.CoordinationV1().Leases(l.leaseNamespace).Update(ctx, current, metav1.UpdateOptions{})
			if err == nil {
				return nil // Leadership maintained
			}
		}

		time.Sleep(time.Second)
	}
}

func (l *LeaseLeaderElection) Renew(ctx context.Context) error {
	ticker := time.NewTicker(l.leaseDuration / 2)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			lease, err := l.clientset.CoordinationV1().Leases(l.leaseNamespace).Get(ctx, l.leaseName, metav1.GetOptions{})
			if err != nil {
				return fmt.Errorf("failed to get lease: %w", err)
			}

			if lease.Spec.HolderIdentity == nil || *lease.Spec.HolderIdentity != l.holderIdentity {
				return fmt.Errorf("lost leadership")
			}

			lease.Spec.RenewTime = &metav1.Time{Time: time.Now()}
			_, err = l.clientset.CoordinationV1().Leases(l.leaseNamespace).Update(ctx, lease, metav1.UpdateOptions{})
			if err != nil {
				return fmt.Errorf("failed to renew lease: %w", err)
			}

		case <-ctx.Done():
			return nil
		}
	}
}

func pointerToInt32(i int32) *int32 {
	return &i
}

Redis-based leader election with Redlock

Implement leader election using Redis Redlock algorithm:

import redis
import time
import uuid

class RedisLeaderElection:
    def __init__(self, redis_client, lock_key='leader_lock', ttl=10000):
        self.redis = redis_client
        self.lock_key = lock_key
        self.ttl = ttl  # milliseconds
        self.node_id = str(uuid.uuid4())
        self.lock_value = None

    def acquire(self):
        """
        Try to acquire the lock using SET NX EX
        Returns True if acquired, False otherwise
        """
        self.lock_value = f"{self.node_id}:{int(time.time() * 1000)}"
        
        # SET key value NX EX ttl
        result = self.redis.set(
            self.lock_key,
            self.lock_value,
            nx=True,
            ex=self.ttl / 1000
        )
        
        return result is not None

    def renew(self):
        """
        Renew the lock if we still hold it
        """
        if not self.is_leader():
            return False
        
        # Check if current value matches our lock value
        current_value = self.redis.get(self.lock_key)
        if current_value and current_value.decode() == self.lock_value:
            # Extend the TTL
            self.redis.expire(self.lock_key, self.ttl / 1000)
            return True
        
        return False

    def is_leader(self):
        """
        Check if we are the current leader
        """
        current_value = self.redis.get(self.lock_key)
        if not current_value:
            return False
        
        return current_value.decode() == self.lock_value

    def release(self):
        """
        Release the lock if we hold it
        """
        current_value = self.redis.get(self.lock_key)
        if current_value and current_value.decode() == self.lock_value:
            self.redis.delete(self.lock_key)