Coastal & Marine Spatial Analysis Pipelines

Using Spot Instances for Bathymetric Jobs

Interruptible capacity is sixty to ninety per cent cheaper and can be reclaimed with about two minutes’ notice. A bathymetric tile pipeline is close to the ideal workload for it — hundreds of independent units, each a few minutes long, no inter-worker communication — and it is also a workload that corrupts its output on preemption if the write path was designed for stable nodes. This page makes preemption the normal case, within the Kubernetes scheduling workflow.

Root cause: preemption is a kill in the middle of whatever was happening

A preempted pod receives SIGTERM, then SIGKILL after the grace period, and the underlying node disappears shortly after. Whatever the container was doing stops. For a tile job that is one of three states, and only one of them is safe:

Between tiles — nothing in flight, the next attempt starts at the next index. Safe.

Mid-computation — CPU work is lost and the tile is re-run. Wasteful and safe.

Mid-write — a partially written object. Whether that corrupts the product depends entirely on the write path: a single-object PUT is atomic and the partial write never becomes visible, while an append to a Zarr store or a multi-part upload left incomplete does become visible or does leave garbage.

# Unsafe: the store is mutated incrementally, so a kill leaves it half-updated.
ds.to_zarr(store, region={"y": slice(y0, y1), "x": slice(x0, x1)})

# Safe: the tile is written to its own object and promoted atomically.
tmp = f"{store}/_staging/{run_id}/{tile_id}.tif"
write_cog(result, tmp)
storage.copy(tmp, f"{store}/tiles/{tile_id}.tif")   # atomic per object

The second form has the property that matters: re-running the tile after a preemption produces the same bytes at the same key, so the output is identical whether the tile ran once or four times.

Where the Kill Lands Decides What It Costs A timeline of one tile's processing divided into three phases: reading input, computing, and writing output. Three preemption arrows point at different phases. An arrow landing between tiles costs nothing, and the next attempt resumes at the next index. An arrow landing during computation loses the CPU time spent so far and the tile is simply re-run. An arrow landing during the write has two possible outcomes shown side by side: with an atomic per-object write, the partial upload never becomes visible and the re-run overwrites it cleanly; with an incremental region write into a shared store, the store now holds a half-updated region that no later run can distinguish from a complete one. read input compute write output between mid-compute mid-write preemption arrives costs nothing costs the CPU time so far atomic per object: invisible, re-run overwrites region write: a half-updated store, undetectable Only the last row differs between designs, and it is the one that decides whether spot capacity is usable at all. Make the write atomic and preemption becomes a cost question rather than a correctness one.

Step-by-step

1. Catch the signal and checkpoint 

import logging
import signal
import sys
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class Draining:
    """Set when the platform signals an imminent preemption."""
    flag: bool = False


DRAIN = Draining()


def install_drain_handler() -> None:
    def _handler(signum, frame):
        logger.warning("received signal %d — draining after the current tile", signum)
        DRAIN.flag = True
    signal.signal(signal.SIGTERM, _handler)


def process_tiles(indices, do_tile, mark_done) -> int:
    """Process tiles until done or drained. Returns the exit code."""
    install_drain_handler()
    for i in indices:
        if DRAIN.flag:
            logger.info("draining: stopping before tile %s", i)
            return 0                     # a clean exit; the Job will reschedule
        do_tile(i)
        mark_done(i)
    return 0

Exiting zero on a drain rather than propagating a failure matters: a non-zero exit consumes one of the Job’s backoffLimit retries, and on spot capacity a job can be preempted a dozen times legitimately. A drained pod has not failed; it has stopped early.

2. Track completion where the pod cannot lose it 

def completed_indices(store, run_id: str) -> set[int]:
    """Which tiles already have an output object. The store is the source of truth."""
    prefix = f"{store}/tiles/"
    done = set()
    for key in storage.list(prefix):
        name = key.rsplit("/", 1)[-1]
        if name.endswith(".tif"):
            done.add(int(name.split(".")[0]))
    logger.info("run %s: %d tile(s) already complete", run_id, len(done))
    return done

Listing the outputs rather than maintaining a separate ledger removes an entire class of bug: the ledger and the reality cannot disagree, because the reality is the ledger. On a store with a slow list operation, a small side index is faster and needs a reconciliation step; on most object stores the listing is fast enough that the simpler form wins.

3. Size the tile to the notice period

The termination notice is about two minutes on most platforms. A tile taking twenty minutes cannot finish inside it, so a preemption always wastes the work in flight. Sizing tiles to a few minutes bounds that waste:

def expected_waste_fraction(tile_minutes: float, preemption_rate_per_hour: float) -> float:
    """Fraction of compute lost to preemption, on average."""
    if tile_minutes <= 0 or preemption_rate_per_hour < 0:
        raise ValueError("inputs must be non-negative and the tile time positive")
    # On average a preemption lands halfway through the tile in flight.
    lost_minutes_per_hour = preemption_rate_per_hour * tile_minutes / 2.0
    return lost_minutes_per_hour / 60.0


for t in (2, 5, 20, 60):
    print(f"{t:>3} min tiles: {expected_waste_fraction(t, 0.5):.1%} of compute wasted")
#   2 min tiles: 0.8% of compute wasted
#   5 min tiles: 2.1% of compute wasted
#  20 min tiles: 8.3% of compute wasted
#  60 min tiles: 25.0% of compute wasted

At an hour per tile a quarter of the compute is thrown away, which erases most of the discount. The tile size is therefore an economic parameter as much as a memory one.

4. Keep the coordinator off spot

A Dask scheduler or a Prefect agent holds state that the workers do not. Losing a worker costs one tile; losing the coordinator costs the run. Put the coordinator on on-demand capacity and the workers on spot, and the cost profile barely changes because the coordinator is one small pod among hundreds.

The Discount Erodes With Tile Duration A chart of effective cost per completed tile against tile duration in minutes, at a preemption rate of one every two hours. The on-demand price is a flat horizontal reference line. The spot curve starts at about thirty per cent of the on-demand price for two-minute tiles, rises slowly through forty per cent at twenty minutes, and climbs steeply beyond an hour as re-run waste dominates, crossing the on-demand line at roughly ninety minutes. A shaded band up to about fifteen minutes marks the region where the discount is essentially the headline rate. A note states that the tile size is an economic parameter and that at an hour per tile a quarter of all compute is discarded. tile duration (minutes) cost per tile on-demand price full discount crosses at ≈ 90 min 2154590 Tile size is an economic parameter here, not only a memory one — which is a reason to keep tiles small.

Verification and acceptance test 

import pytest


def test_a_tile_is_idempotent(tile_index, store):
    """Running a tile twice must produce byte-identical output."""
    do_tile(tile_index)
    first = storage.read(f"{store}/tiles/{tile_index}.tif")
    do_tile(tile_index)
    second = storage.read(f"{store}/tiles/{tile_index}.tif")
    assert first == second, "re-running the tile changed its output"


def test_a_drain_exits_zero_and_leaves_no_partial_object(store, run_id):
    DRAIN.flag = True
    rc = process_tiles([0, 1, 2], do_tile, mark_done)
    assert rc == 0, "a drain must not consume a backoffLimit retry"
    partials = [k for k in storage.list(f"{store}/_staging/{run_id}/")]
    assert not partials, f"{len(partials)} staged object(s) left behind after a drain"

The idempotence test is the one to run in CI against real storage rather than a mock. Object stores have behaviours — conditional writes, versioning, eventual listing — that a mock does not reproduce, and idempotence is a property of the combination rather than of the code.

Making the completion set survive a whole-job restart

A preempted pod is one failure mode; a preempted Job is another. If the control plane loses the Job object, or an operator deletes and reapplies it, the new Job starts from index zero with no memory of what the previous one finished. On a four-hundred-tile survey that is potentially four hundred re-runs of work already done.

The completion set from the previous section solves it, provided one condition holds: the run identifier must be derived from the work, not from the Job. A Job-generated run ID changes on reapply and the new Job looks at an empty prefix.

import hashlib
import json


def deterministic_run_id(manifest_uri: str, params: dict, image_digest: str) -> str:
    """A run ID that is the same for the same work, across Job objects."""
    payload = json.dumps(
        {"manifest": manifest_uri, "params": params, "image": image_digest},
        sort_keys=True, separators=(",", ":"),
    ).encode()
    rid = hashlib.sha256(payload).hexdigest()[:16]
    logger.info("run id %s for manifest %s image %s", rid, manifest_uri, image_digest[:19])
    return rid

Hashing the manifest, the parameters and the image digest gives an identifier that is stable across restarts and changes when any input changes — which is the behaviour wanted at both ends. Reapplying the same Job resumes; changing a parameter starts a fresh run rather than mixing outputs from two configurations under one prefix.

That last property is worth stating plainly, because the alternative failure is quiet. A run ID derived only from the date would let a re-run with a corrected threshold write tiles alongside tiles produced with the old one, and the resulting mosaic would be internally inconsistent with nothing in the output to indicate it. Deriving the ID from the inputs makes that impossible: different inputs, different prefix, no mixing.

The cost is that a genuinely trivial change — a comment in the code, a new image build with identical libraries — produces a new run ID and a full re-run. Including the image digest is the deliberate choice there: it is the only way to guarantee that all tiles in a prefix were produced by the same bytes, and a re-run is cheaper than an unexplained seam.

Derive the Run ID From the Work, Not From the Job Two scenarios applied to two identifier schemes. Under a date-derived run identifier, deleting and reapplying the Job resumes into the same prefix, which is correct; but re-running with a changed threshold also writes into the same prefix, so tiles produced under two different configurations sit side by side in one mosaic with nothing to distinguish them. Under an identifier hashed from the manifest, the parameters and the image digest, the reapply case resolves to the same hash and resumes, while the changed-threshold case resolves to a different hash and starts a clean prefix. A note records that the second scheme also forces a full re-run when only the image changes, and that this is the deliberate price of guaranteeing every tile in a prefix came from the same bytes. run id = the date Job deleted and reapplied → same prefix, resumes ✓ threshold changed, re-run → same prefix, two configs mixed ✗ run id = hash(manifest, params, image) Job deleted and reapplied → same hash, resumes ✓ threshold changed, re-run → new hash, clean prefix ✓ The failure on the left is quiet: a mosaic assembled from two configurations, with nothing in the output saying so. The cost on the right is a full re-run whenever the image changes, even trivially. That is the deliberate price of being able to say every tile in a prefix came from the same bytes.

Edge cases and gotchas

  • Preemption rates are not uniform across instance types or zones. A job pinned to one instance type in one zone can see a rate ten times the fleet average during a capacity crunch. Spread across types and zones, and treat a sudden rate change as a signal to fall back to on-demand rather than as noise.
  • The grace period is a maximum, not a guarantee. Some platforms reclaim faster under pressure. A drain handler that assumes it has the full two minutes to finish a tile will occasionally be wrong; design it to abandon rather than to complete.
  • Staged objects accumulate. Every preemption mid-write leaves a staged object that no run will promote. A lifecycle rule on the staging prefix, as in setting S3 lifecycle rules, is what stops the discount being spent on abandoned uploads.

Up: Kubernetes Job Scheduling for Bathymetric Processing