Scheduled Jobs Across Replicas

apiVersion: batch/v1
kind: CronJob
metadata:
  name: invoice-reminders
  namespace: billing
spec:
  schedule: "30 2 * * *"
  timeZone: "Europe/London"
  concurrencyPolicy: Forbid
  startingDeadlineSeconds: 600
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 5
  jobTemplate:
    spec:
      backoffLimit: 2
      activeDeadlineSeconds: 900
      template:
        spec:
          restartPolicy: Never
          containers:
            - name: job
              image: registry.example.com/billing/jobs:1.14.0
              args: ["invoice-reminders"]
              env:
                - name: ConnectionStrings__Billing
                  valueFrom:
                    secretKeyRef:
                      name: billing-db
                      key: connection-string
              resources:
                requests: { cpu: "100m", memory: "128Mi" }
                limits: { memory: "256Mi" }

using Npgsql;

public sealed class SettlementLeader(IConfiguration config, ILogger<SettlementLeader> logger) : BackgroundService
{
    private const long LockKey = 72002;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Pooling off: closing this connection must really end the session, and with it the lock.
        // A pooled connection would go back to the pool with the lock still held.
        var lockConnectionString = new NpgsqlConnectionStringBuilder(config.GetConnectionString("Billing"))
        {
            Pooling = false,
            KeepAlive = 15, // client side; the server needs its own tcp_keepalives_* to spot a dead leader
        }.ConnectionString;

        while (!stoppingToken.IsCancellationRequested)
        {
            // A dedicated connection: the session lock lives exactly as long as this connection.
            await using var conn = new NpgsqlConnection(lockConnectionString);
            await conn.OpenAsync(stoppingToken);
            await using var tryLock = new NpgsqlCommand("SELECT pg_try_advisory_lock(@key)", conn);
            tryLock.Parameters.AddWithValue("key", LockKey);

            if (!(bool)(await tryLock.ExecuteScalarAsync(stoppingToken))!)
            {
                await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken); // follower: try again later
                continue;
            }

            logger.LogInformation("This replica is the settlement leader");
            try
            {
                using var timer = new PeriodicTimer(TimeSpan.FromMinutes(1));
                while (await timer.WaitForNextTickAsync(stoppingToken))
                {
                    await using var ping = new NpgsqlCommand("SELECT 1", conn); // a dead connection means the lock is gone
                    await ping.ExecuteScalarAsync(stoppingToken);
                    logger.LogInformation("Settling card payments captured in the last minute");
                }
            }
            catch (NpgsqlException ex)
            {
                logger.LogWarning(ex, "Lost the leader connection; re-electing");
            }
            // Disposing this non-pooled connection ends the session and releases the lock for the next leader.
        }
    }
}

# Run tonight's job now, for example after fixing a bug
kubectl create job --from=cronjob/invoice-reminders invoice-reminders-manual-1 -n billing

# Watch it and read its output
kubectl get jobs -n billing --watch
kubectl logs job/invoice-reminders-manual-1 -n billing

# Pause the schedule during an incident without deleting it
kubectl patch cronjob invoice-reminders -n billing -p '{"spec":{"suspend":true}}'

# See the last scheduled and successful times
kubectl get cronjob invoice-reminders -n billing -o jsonpath='{.status.lastScheduleTime}{" "}{.status.lastSuccessfulTime}{"\n"}'