📄 Leader Election in Go with a Postgres database using Kubernetes tooling

package main

import (
	"context"
	"fmt"
	"log"
	"path/filepath"
	"time"

	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/tools/leaderelection"
	"k8s.io/client-go/tools/leaderelection/resourcelock"
	"k8s.io/client-go/util/homedir"
)

func work(ctx context.Context, cancel context.CancelFunc) {
	log.Println("WORK: performing work")
}

func main() {
	config, err := clientcmd.BuildConfigFromFlags("", filepath.Join(homedir.HomeDir(), ".kube", "config"))
	if err != nil {
		panic(err.Error())
	}
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		panic(err.Error())
	}

	id := fmt.Sprintf("%v", time.Now().Unix())
	log.Println("id:", id)

	l, err := resourcelock.New(
		resourcelock.LeasesResourceLock,
		"default", "leader-election-example",
		clientset.CoreV1(),
		clientset.CoordinationV1(),
		resourcelock.ResourceLockConfig{
			Identity: id,
		},
      )
	if err != nil {
		log.Println("error: %w", err)
		return
	}

	leaderelector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
		Lock:            l,
		ReleaseOnCancel: true,
		LeaseDuration:   15 * time.Second,
		RenewDeadline:   10 * time.Second,
		RetryPeriod:     2 * time.Second,
		Callbacks: leaderelection.LeaderCallbacks{
			OnStartedLeading: func(ctx context.Context) {
				log.Printf("i (%v) am now the leader!\n", id)
			},
			OnStoppedLeading: func() {
				log.Println("no longer the leader, staying inactive.")
			},
			OnNewLeader: func(currentID string) {
				if currentID == id {
					log.Printf("i (%v) am still the leader!\n", id)
					return
				}
				log.Printf("new/current leader is %s\n", currentID)
			},
		},
	})
	if err != nil {
		log.Fatalf("leader election error: %w\n", err)
		return
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		for {
			if leaderelector.IsLeader() {
				work(ctx, cancel)
			}
			time.Sleep(10 * time.Second)
		}
	}()
	leaderelector.Run(ctx)
}

type lock struct {
	id   string
	name string
	db   *sql.DB

	lease *resourcelock.LeaderElectionRecord
}

func (l *lock) Get(ctx context.Context) (ler *resourcelock.LeaderElectionRecord, rb []byte, err error) {
	log.Printf("getting: %v\n", l.name)
	sqlStatement := `
	select
	    holderIdentity, leaseDurationSeconds, acquireTime, renewTime, leaderTransitions
	from leader_election where name = $1 limit 1`
	rows, err := l.db.Query(sqlStatement, l.name)
	if err != nil {
		return nil, nil, err
	}

	defer func() {
		if err := rows.Close(); err != nil {
			log.Printf("error: failed to close rows: %v\n", err)
		}
	}()

	for rows.Next() {
		ler = &resourcelock.LeaderElectionRecord{}
		var acquireTime, renewTime int
		if err = rows.Scan(&ler.HolderIdentity, &ler.LeaseDurationSeconds, &acquireTime, &renewTime, &ler.LeaderTransitions); err != nil {
			return nil, nil, err
		}
		ler.AcquireTime = metav1.NewTime(time.Unix(int64(acquireTime), 0))
		ler.RenewTime = metav1.NewTime(time.Unix(int64(renewTime), 0))
	}

	if ler == nil {
		l.lease = nil
		return &resourcelock.LeaderElectionRecord{}, nil, nil
	}

	l.lease = ler
	rb, err = json.Marshal(ler)
	if err != nil {
		return nil, nil, err
	}
	return ler, rb, nil
}

func (l *lock) Create(ctx context.Context, ler resourcelock.LeaderElectionRecord) error {
	log.Printf("creating: %+v\n", ler)
	sqlStatement := `
          insert into leader_election
              (name, holderIdentity, leaseDurationSeconds, acquireTime, renewTime, leaderTransitions)
          values ($1, $2, $3, $4, $5, $6)
              returning name, holderIdentity, leaseDurationSeconds, acquireTime, renewTime, leaderTransitions`
	rows, err := l.db.Query(
		sqlStatement,
		l.name,
		ler.HolderIdentity,
		ler.LeaseDurationSeconds,
		ler.AcquireTime.Unix(),
		ler.RenewTime.Unix(),
		ler.LeaderTransitions,
	)
	if err != nil {
		return err
	}

	defer func() {
		if err := rows.Close(); err != nil {
			log.Printf("error: failed to close rows: %v\n", err)
		}
	}()

	for rows.Next() {
		if l.lease == nil {
			l.lease = &resourcelock.LeaderElectionRecord{}
		}
		if err = rows.Scan(
			&l.lease.HolderIdentity,
			&l.lease.LeaseDurationSeconds,
			&l.lease.AcquireTime,
			&l.lease.RenewTime,
			&l.lease.LeaderTransitions,
		); err != nil {
			return err
		}
	}
	return nil
}

func (l *lock) Update(ctx context.Context, ler resourcelock.LeaderElectionRecord) error {
	if l.lease == nil {
		return l.Create(ctx, resourcelock.LeaderElectionRecord{})
	}
	log.Printf("updating: %+v\n", ler)
	sqlStatement := `update leader_election
                     set holderIdentity = $2, leaseDurationSeconds = $3, acquireTime = $4, renewTime = $5, leaderTransitions = $6
                     where name = $1`
	rows, err := l.db.Query(
		sqlStatement,
		l.name,
		ler.HolderIdentity,
		ler.LeaseDurationSeconds,
		ler.AcquireTime.Unix(),
		ler.RenewTime.Unix(),
		ler.LeaderTransitions,
	)
	if err != nil {
		return err
	}

	defer func() {
		if err := rows.Close(); err != nil {
			log.Printf("error: failed to close rows: %v\n", err)
		}
	}()
	return nil
}

func (l *lock) RecordEvent(s string) {
	log.Println("leader election event:", s)
}

func (l *lock) Describe() string {
	return l.name
}

func (l *lock) Identity() string {
	return l.id
}

func newLock() *lock {
	db, err := sql.Open("postgres",
		"postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable")
	if err != nil {
		return nil, err
	}
	id := fmt.Sprintf("%v", time.Now().Unix())
	return &lock{
		id:    id,
		name:  "default",
		db:    db,
		lease: nil,
	}
}

...
	lock := newLock()
	log.Println("id:", lock.Identity())
	leaderelector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
		Lock:            lock,
...