Run a job on a kind cluster¶

By the end of this page you'll have a local Kubernetes cluster with everything the adapter needs, and a small Go program that starts a two-node job through the adapter, lets it run, holds and releases it, and stops it. You'll see the job's gate, its execution identity changing across a re-admission, and the admission controller doing its part.

The cluster is the one the adapter's own test suite uses, so what you build here is the tested configuration. Allow about an hour, most of it waiting for images and controllers.

Before you start¶

You'll need:

• Docker, kind and kubectl;
• Go 1.27.2 or later (the version go.mod names), and gunzip;
• a machine with room for three node containers. The setup script caps each at 2500 MB of memory, so allow about 7.5 GB for the cluster on top of what the rest of the machine is doing.

The setup script does not pin a Kubernetes version: the cluster runs the default node image of your kind binary. The tested run used kind v0.33.0, whose default node image is Kubernetes 1.37.0. Tested versions lists the rest.

Stand up the cluster¶

Clone the repository. Its internal/kindtest/testdata directory holds the scripts and manifests.

git clone https://gitlab.com/phpboyscout/go/dispatch-kubeflow
cd dispatch-kubeflow

up.sh takes a cluster name and a kubeconfig path, and needs MODULE to name the repository root:

export MODULE="$PWD"
internal/kindtest/testdata/up.sh dkkind /tmp/dkkind.kubeconfig
export KUBECONFIG=/tmp/dkkind.kubeconfig

The script does these things, in order:

1. creates a kind cluster of one control-plane node and two workers (kind.yaml), unless one of that name exists;
2. installs JobSet v0.12.0, Kueue v0.20.1 and Trainer v2.3.0 from manifests stored in the repository, waiting for each to roll out;
3. builds dispatch-admission from this checkout and publish from the go/dispatch version go.mod pins, builds a busybox image for the job to run, and loads all three into the nodes;
4. applies base.yaml, which creates the objects the job needs: a ResourceFlavor, two WorkloadPriorityClasses (low and high), an AdmissionCheck named dispatch-admission, and two ClusterTrainingRuntimes, dispatch-kind and dispatch-kind-init;
5. applies admission.yaml, which deploys the admission controller in the dispatch-system namespace;
6. pins the controllers to the control-plane node, so that pausing a worker later loses only training pods.

When it returns, look at what it made:

kubectl get admissionchecks
kubectl get clustertrainingruntimes
kubectl -n dispatch-system get deploy dispatch-admission
kubectl -n dispatch-system get lease dispatch-admission

The Lease is the admission controller's leader-election lock. The adapter reads it as the controller's heartbeat, which is why it matters to you in a moment. See the Lease heartbeat.

Give the job a queue¶

In Kueue mode a job names a LocalQueue, and the LocalQueue's ClusterQueue has to list dispatch's AdmissionCheck. Create a namespace, then a ClusterQueue and LocalQueue that fit one two-node job (the runtime asks for 100m CPU and 32Mi of memory per node):

kubectl create namespace tutorial

kubectl apply -f - <<'EOF'
apiVersion: kueue.x-k8s.io/v1beta2
kind: ClusterQueue
metadata: {name: tutorial}
spec:
  namespaceSelector: {matchLabels: {kubernetes.io/metadata.name: tutorial}}
  admissionChecksStrategy: {admissionChecks: [{name: dispatch-admission}]}
  resourceGroups:
  - coveredResources: [cpu, memory]
    flavors:
    - name: default-flavor
      resources: [{name: cpu, nominalQuota: 250m}, {name: memory, nominalQuota: 256Mi}]
---
apiVersion: kueue.x-k8s.io/v1beta2
kind: LocalQueue
metadata: {name: lq, namespace: tutorial}
spec: {clusterQueue: tutorial}
EOF

Note that the check is listed with no onFlavors. The adapter refuses a ClusterQueue that limits the check to some flavors; Install the admission controller says why.

Write the program¶

Make a new module beside the checkout and fetch the adapter:

mkdir ../dkdemo && cd ../dkdemo
go mod init example.com/dkdemo
go get gitlab.com/phpboyscout/go/dispatch-kubeflow@main

The module has no release yet, so go get resolves a pseudo-version.

Everything below goes in one main.go. Start with the imports and two helpers:

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "io"
    "log"
    "os"
    "strings"
    "time"

    "gitlab.com/phpboyscout/go/dispatch"
    "gitlab.com/phpboyscout/go/dispatch/model"

    kubeflow "gitlab.com/phpboyscout/go/dispatch-kubeflow"
)

// await polls ok every second until it holds or the limit passes.
func await(what string, limit time.Duration, ok func() bool) {
    for until := time.Now().Add(limit); ; time.Sleep(time.Second) {
        if ok() {
            return
        }

        if time.Now().After(until) {
            log.Fatalf("never saw %s within %s", what, limit)
        }
    }
}

// rankZeroLog reads rank 0's log of the current execution until it holds marker.
func rankZeroLog(ctx context.Context, r dispatch.OutputReader, job model.JobID, marker string) string {
    var out string

    await("rank 0's log", 2*time.Minute, func() bool {
        rc, err := r.Output(ctx, job, 1, dispatch.Stdout, false)
        if err != nil {
            return false
        }

        defer rc.Close()

        b, err := io.ReadAll(rc)
        out = string(b)

        return err == nil && strings.Contains(out, marker)
    })

    return out
}

Open the adapter¶

The adapter's configuration is a JSON section. This one points at the kubeconfig, the tutorial namespace and the LocalQueue, and names the runtime the preflight probe uses:

func main() {
    ctx := context.Background()

    settings, err := json.Marshal(map[string]any{
        "cluster":      map[string]any{"kubeconfig": os.Getenv("KUBECONFIG")},
        "namespace":    "tutorial",
        "probeRuntime": map[string]any{"kind": "ClusterTrainingRuntime", "name": "dispatch-kind"},
        "kueue": map[string]any{
            "localQueue": "lq", "priorityClass": "low", "admissionCheck": "dispatch-admission",
        },
        "timeouts": map[string]any{"request": "10s", "stopJob": "3m"},
    })
    if err != nil {
        log.Fatal(err)
    }

    opened, err := kubeflow.Factory(ctx, dispatch.AdapterConfig{Backend: kubeflow.Name, Settings: settings})
    if err != nil {
        log.Fatal(err)
    }

    backend := opened.(*kubeflow.Backend)
    defer backend.Close()

Factory runs the adapter's preflight before it returns. It checks that the cluster serves the API versions the adapter supports, asks the apiserver whether it may do everything it needs, proves the readmission gate, and creates a TrainJob that cannot be scheduled to prove that the start gate holds. It watches that probe for ten seconds, so expect Factory to take at least that long.

Print what preflight decided:

    caps, err := backend.Capabilities(ctx, nil)
    if err != nil {
        log.Fatal(err)
    }

    for _, c := range []model.Capability{model.ReadmissionGate, model.AttemptInjection, model.HoldCapability} {
        s := caps[c]
        fmt.Printf("%s: %s (%s) %s\n", c, s.Value, s.Source, s.Reason)
    }

All three should say yes and probed. The reasons name what was proven: for the readmission gate, that ClusterQueue tutorial lists AdmissionCheck dispatch-admission, whose controller holds Lease dispatch-system/dispatch-admission. If one says no or unknown, the reason says why, and the adapter is observe-only, which means Start will refuse. Read a preflight result lists the reasons.

Start the job, closed¶

A job is a model.JobSpec. Two extensions are required: the image to run, and the runtime to run it on. The command is a script that prints the execution identity dispatch gives it, then waits:

    job, err := model.NewJobID(time.Now(), nil)
    if err != nil {
        log.Fatal(err)
    }

    spec := model.JobSpec{
        Command: []string{"/bin/sh", "-c",
            `env | grep '^DISPATCH_' | sort; echo tutorial-ready; trap 'exit 143' TERM; while :; do sleep 1; done`},
        Nodes: 2,
        Extensions: model.Extensions{
            model.Image{Ref: "localhost/dispatch-busybox:kind"},
            kubeflow.RuntimeRef{Kind: kubeflow.KindClusterTrainingRuntime, Runtime: "dispatch-kind"},
        },
    }

    ref, err := backend.Start(ctx, model.StartRequest{Job: job, Submission: 1, Attempt: 1, Spec: spec})
    if err != nil {
        log.Fatal(err)
    }

    obs, err := backend.Inspect(ctx, ref)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println(ref.ID, obs.State, "-", obs.Cause.Detail)

Start creates the TrainJob with its start gate closed, and does not return until Kueue has made the TrainJob's Workload. You'll see something like:

tutorial/dj-<lowercase job id>-s1 gated -

The cause detail is empty: dispatch's own gate needs no explanation. A gate someone else added would show here as gated by <their gate>.

Nothing is running. Run kubectl -n tutorial get trainjobs,workloads,pods in another terminal to see a TrainJob and a Workload, and no pods. Read the TrainJob's gate:

kubectl -n tutorial get trainjobs -o jsonpath='{.items[0].metadata.annotations}'

The annotation kueue.x-k8s.io/admission-gated-by is dispatch.phpboyscout.uk/gate. While it is there Kueue will not admit the Workload. The start gate and the readmission gate explains why it works this way.

Start also made a copy of the runtime, a TrainingRuntime with the TrainJob's name, so that a later edit to dispatch-kind cannot change a job already submitted:

kubectl -n tutorial get trainingruntimes

Open it¶

Open removes dispatch's gate. Wait for the job to run, then read rank 0's log through the adapter:

    if err := backend.Open(ctx, ref); err != nil {
        log.Fatal(err)
    }

    await("the job running", 3*time.Minute, func() bool {
        o, err := backend.Inspect(ctx, ref)
        return err == nil && o.State == model.ObjectRunning
    })

    reader := opened.(dispatch.OutputReader)
    fmt.Print(rankZeroLog(ctx, reader, job, "tutorial-ready"))

Between Open and running, Kueue reserved quota for the Workload and held it at the AdmissionCheck. The admission controller saw that, raised the job's admission count A from 0 to 1 in the TrainJob, and set the check Ready. Then Kueue admitted the Workload and Trainer started the pods.

The log shows the execution identity:

DISPATCH_ATTEMPT=<job id>.1.1.1.0
DISPATCH_EPOCH=1000001
DISPATCH_JOB_ID=<job id>
DISPATCH_SUBMISSION=1
tutorial-ready

DISPATCH_ATTEMPT ends in .1.0: A is 1 and J, JobSet's restart count, is 0. DISPATCH_EPOCH is submission 1 followed by R = A + J = 1, padded to six digits. Execution identity: R = A + J says why these two numbers.

Hold it and release it¶

A hold ends the running execution and keeps the job from being admitted again until it is released. Both calls carry a deadline, the actor, and an instance number that names this hold:

    meta := model.HoldMeta{
        Instance: 1,
        CallMeta: model.CallMeta{Deadline: time.Now().Add(time.Minute)},
        By:       model.Actor{Kind: model.ActorHuman, ID: "tutorial"},
        Until:    time.Now().Add(time.Hour),
        Reason:   "the tutorial's hold",
    }

    if err := backend.Hold(ctx, ref, meta); err != nil {
        log.Fatal(err)
    }

    await("the job held", 2*time.Minute, func() bool {
        o, err := backend.Inspect(ctx, ref)
        return err == nil && o.State == model.ObjectHeld && o.HoldInstance == 1
    })

    fmt.Println("held")

In Kueue mode the hold deactivates the job's Workload (spec.active: false), and Kueue ends the execution. Look at the Workload's annotations:

kubectl -n tutorial get workloads -o jsonpath='{.items[0].metadata.annotations}'

You'll see dispatch.phpboyscout.uk/hold-instance, hold-since, hold-who, hold-until and hold-reason. Holds are Workload deactivation covers why the hold lives there and not on the TrainJob.

Now release it. The deadline for a release is its own, so set a fresh one:

    meta.CallMeta.Deadline = time.Now().Add(time.Minute)

    if err := backend.Release(ctx, ref, meta); err != nil {
        log.Fatal(err)
    }

    await("the job running again", 3*time.Minute, func() bool {
        o, err := backend.Inspect(ctx, ref)
        return err == nil && o.State == model.ObjectRunning
    })

    fmt.Print(rankZeroLog(ctx, reader, job, "tutorial-ready"))

Kueue queues the Workload again and reserves quota, and resets the AdmissionCheck to Pending. The controller does not set it Ready while a pod of the previous execution is still there, however briefly; once they have gone it raises A to 2 and approves. The new log's identity differs from the first:

DISPATCH_ATTEMPT=<job id>.1.1.2.0
DISPATCH_EPOCH=1000002

Same job, same submission, A is now 2, so R is 2, and the epoch is higher. Anything written under the first epoch is distinguishable from, and ordered below, anything written under this one. The controller's refusal to approve until the old pods are gone is never two executions at once.

Stop it¶

StopJob deletes everything of the job and checks that it is gone:

    report, err := backend.StopJob(ctx, job, "the tutorial is over")
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("stopped:", len(report.Stopped), "alive:", len(report.Alive), "unproven:", len(report.Unproven))
}

Run the program:

go mod tidy
go run .

StopJob deletes the TrainJob with background propagation, so the TrainJob is gone at once and the cluster's garbage collector removes the JobSet, Jobs, pods and runtime copy afterwards. While they go, Find reports what is left as orphans, and StopJob deletes those too, each under its own UID, on every pass until nothing of the job remains or timeouts.stopJob passes (three minutes here). The report names every object it deleted, so stopped: can count an orphaned JobSet, pods or runtime copy as well as the TrainJob. An object is listed under Stopped when it and every pod beneath it are gone. See background deletion for why it is not foreground.

Clean up¶

cd ../dispatch-kubeflow
internal/kindtest/testdata/down.sh dkkind

down.sh deletes the cluster, the kind Docker network if no other cluster uses it, and the three images up.sh built.

What you built¶

You ran a two-node job through every step the adapter takes: started closed, opened, held, released into a second execution with a higher identity, and stopped. The cluster, its manifests and the controller are the same ones the adapter's test suite runs against.

To run that suite rather than a program of your own, see Run the conformance suite against kind.

Next¶

• Install the admission controller in a cluster of your own.
• Choose Kueue mode or plain mode.
• Run the conformance suite against kind.
• Configuration reference.