Skip to content

Task Workers on Kubernetes

Lakekeeper runs background work in task-queue workers, for example purging the files of dropped tables and, in Lakekeeper Plus, expiring snapshots, removing orphan files and optimizing tables. By default every Lakekeeper pod runs these workers next to the HTTP API. Optimize workers run only where you turn them on, see Optimize Workers. On Kubernetes you can move this work to worker pods that KEDA starts only while tasks wait. The API pods then stay free of heavy work, and the cluster can remove nodes when there is nothing to do.

How It Works

  1. The API pods run no workers of the queue. On their metrics port they report the task queue demand: how many tasks of the queue wait or run.
  2. A KEDA ScaledJob reads the demand and starts worker Jobs.
  3. Each Job runs Lakekeeper without the HTTP API (LAKEKEEPER__SERVE_HTTP_API=false). It takes tasks until it had no task for LAKEKEEPER__TASK_IDLE_EXIT_AFTER, and then exits with code 0, so the Job completes. See Stopping Task Workers.

We recommend one ScaledJob per queue. Set the queue's LAKEKEEPER__TASK_*_WORKERS to 0 on the API pods, and on the workers set the worker counts of all other queues to 0.

The demand comes from the API pods. While none of them is available, KEDA starts no workers. We recommend at least two API pods and a PodDisruptionBudget, so that a node drain or an upgrade does not stop the workers from starting.

Example

A ScaledJob for the tabular_purge queue:

apiVersion: keda.sh/v1alpha1
kind: ScaledJob
metadata:
  name: lakekeeper-tabular-purge
spec:
  maxReplicaCount: 5
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 10
  jobTargetRef:
    backoffLimit: 0
    template:
      spec:
        restartPolicy: Never
        initContainers:
          - name: check-db
            image: quay.io/lakekeeper/catalog:<version>
            args: ["wait-for-db", "-dm", "-r", "100", "-b", "2"]
            env:
              # The database settings of the worker.
        containers:
          - name: worker
            image: quay.io/lakekeeper/catalog:<version>
            args: ["serve"]
            env:
              # Also add the database, secret, storage and authorization settings of the API pods.
              - name: LAKEKEEPER__SERVE_HTTP_API
                value: "false"
              - name: LAKEKEEPER__TASK_TABULAR_PURGE_WORKERS
                value: "2"
              - name: LAKEKEEPER__TASK_SOFT_DELETION_WORKERS
                value: "0"
              - name: LAKEKEEPER__TASK_LOG_CLEANUP_WORKERS
                value: "0"
              - name: LAKEKEEPER__TASK_EXPIRE_SNAPSHOTS_WORKERS
                value: "0"
              - name: LAKEKEEPER__TASK_REMOVE_ORPHAN_FILES_WORKERS
                value: "0"
              - name: LAKEKEEPER__TASK_IDLE_EXIT_AFTER
                value: "180s"
  triggers:
    - type: metrics-api
      metadata:
        url: "http://<service>.<namespace>.svc:9000/task-queue/tabular_purge/demand"
        valueLocation: "demand"
        targetValue: "2"

targetValue is the number of tasks one worker pod runs at the same time, here LAKEKEEPER__TASK_TABULAR_PURGE_WORKERS. With restartPolicy: Never and backoffLimit: 0, Kubernetes does not restart a worker that failed. The task queue tries the task again, and KEDA starts a new Job while tasks wait. Task Queue Demand lists the queue names and further recommended settings.

serve refuses to start while the database is not migrated for its version. The check-db init container waits up to 200 seconds for that migration, like the Helm chart does for the API pods, so a worker of the new version that starts before the migration finished waits for it.

On workers that run more than one task at a time, as in this example, we recommend leaving LAKEKEEPER__TASK_MAX_LIFETIME unset. A worker that reached it takes no new tasks while it finishes a long task, but KEDA still counts it, so other due tasks can wait for that task.

Size Classes

The demand endpoint also accepts ?class=<name>. It then counts only the tasks of one size class of the queue, so that each class can have its own ScaledJob with pods of a matching size. For a class that the queue does not have, the endpoint answers 404.

In Lakekeeper Plus, the optimize queue has size classes. The Lakekeeper Helm chart creates one ScaledJob per class for you, see Optimize Workers with the Helm Chart.

Database Connections

Each Lakekeeper pod opens at most LAKEKEEPER__PG_READ_POOL_CONNECTIONS plus LAKEKEEPER__PG_WRITE_POOL_CONNECTIONS connections to the database (default 10 plus 5). Worker pods count too. Plan the connection limit of your database for all API pods plus the maxReplicaCount of every ScaledJob, each pod with its own pools. A worker pod runs only a few tasks at the same time, so we recommend smaller pools on worker pods than on API pods.

Leftover Files

A worker that is stopped in the middle of a task, for example on a spot node, can leave files in storage that no table references. We recommend:

  • enabling Remove Orphan Files for these warehouses;
  • on S3, a bucket lifecycle rule that aborts incomplete multipart uploads, for example after 7 days. Remove orphan files cannot see these uploads, because they are not objects yet.

Alerting

A task waits as long as no worker takes it, for example when KEDA does not run, the ScaledJob reached maxReplicaCount, or worker pods cannot be scheduled. We recommend an alert on the age of the oldest due task of each queue, for example when it is older than one hour. A running task counts from its last heartbeat, so the alert also covers a task whose worker stopped. This query returns the age in seconds; you can run it as a custom query of postgres_exporter or CloudNativePG (see Database Monitoring):

SELECT queue_name,
    EXTRACT(EPOCH FROM now() - MIN(
        CASE WHEN status = 'scheduled' THEN scheduled_for
             ELSE last_heartbeat_at END
    )) AS oldest_due_seconds
FROM task
WHERE (status = 'scheduled' AND scheduled_for <= now())
    OR status IN ('running', 'should-stop')
GROUP BY queue_name;

If you set LAKEKEEPER__PG_SCHEMA, write the schema in double quotes before the table name, for example "lakekeeper".task.

A worker that is killed during a task, for example because it ran out of memory or used more disk than its limit, fails its Job, because the ScaledJob above does not restart it. We recommend alerts on failed Jobs with kube-state-metrics. If you use kube-prometheus, it already includes the first rule as KubeJobFailed.

- alert: LakekeeperWorkerJobFailed
  expr: kube_job_failed{namespace="<namespace>"} > 0
- alert: LakekeeperWorkerOutOfMemory
  expr: kube_pod_container_status_terminated_reason{namespace="<namespace>", container="worker", reason="OOMKilled"} > 0

The first rule fires for every failed Job in the namespace, the second only for workers that ran out of memory. container is the container name of the ScaledJob above. The optimize workers of the Helm chart use the container name optimize-worker. Both alerts stay until the failed Job is deleted. The ScaledJob keeps the last failedJobsHistoryLimit failed Jobs, so delete the failed Job after the investigation. To find the task of the killed worker, look for the log line "Task attempt timed out" in the logs of the other workers of the queue (see Task Queue Metrics). The killed worker cannot write it: the next worker of the queue writes it after the heartbeat timeout.