Contrarian Opinion: Why We Should Use K8s 1.32 Over Nomad 1.9 for 70% of Batch Workloads

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    "k8s.io/client-go/util/retry"
)

// submitK8sIndexedJob creates a Kubernetes 1.32 Indexed Job for batch ML training
// with 10 completions, 2 parallel workers, and native failure retry handling
func submitK8sIndexedJob(ctx context.Context, clientset *kubernetes.Clientset, jobName string) error {
    // Define the Indexed Job spec, a K8s 1.32 GA feature for batch workloads
    job := &batchv1.Job{
        ObjectMeta: metav1.ObjectMeta{
            Name:      jobName,
            Namespace: "batch-ml",
            Labels: map[string]string{
                "app":      "ml-training",
                "version":  "1.32",
                "workload": "batch",
            },
        },
        Spec: batchv1.JobSpec{
            Completions:             intPtr(10), // 10 total training shards
            Parallelism:             intPtr(2),  // 2 concurrent workers
            BackoffLimit:            intPtr(3),  // Retry failed shards 3 times
            CompletionMode:          batchv1.IndexedCompletionMode, // K8s 1.32 GA feature
            PodReplacementPolicy:   batchv1.FailedPodReplacementPolicy,
            Template: corev1.PodTemplateSpec{
                ObjectMeta: metav1.ObjectMeta{
                    Labels: map[string]string{"job-name": jobName},
                },
                Spec: corev1.PodSpec{
                    RestartPolicy: corev1.RestartPolicyOnFailure,
                    Containers: []corev1.Container{
                        {
                            Name:    "ml-trainer",
                            Image:   "us-east-1.docker.io/myorg/ml-trainer:1.3.2",
                            Command: []string{"/bin/sh", "-c"},
                            Args: []string{
                                // Use JOB_COMPLETION_INDEX env var (set by Indexed Job) to shard data
                                "echo Training shard $JOB_COMPLETION_INDEX; python train.py --shard $JOB_COMPLETION_INDEX --total-shards 10",
                            },
                            Resources: corev1.ResourceRequirements{
                                Requests: corev1.ResourceList{
                                    corev1.ResourceCPU:    resourceQuantity("2"),
                                    corev1.ResourceMemory: resourceQuantity("4Gi"),
                                },
                                Limits: corev1.ResourceList{
                                    corev1.ResourceCPU:    resourceQuantity("4"),
                                    corev1.ResourceMemory: resourceQuantity("8Gi"),
                                },
                            },
                            Env: []corev1.EnvVar{
                                {
                                    Name: "JOB_COMPLETION_INDEX",
                                    ValueFrom: &corev1.EnvVarSource{
                                        JobCompletionIndex: &corev1.Empty{},
                                    },
                                },
                            },
                        },
                    },
                },
            },
        },
    }

    // Retry up to 3 times on conflict (e.g., job already exists with newer version)
    err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
        _, err := clientset.BatchV1().Jobs("batch-ml").Create(ctx, job, metav1.CreateOptions{})
        return err
    })
    if err != nil {
        return fmt.Errorf("failed to create job %s: %w", jobName, err)
    }

    log.Printf("Successfully submitted K8s 1.32 Indexed Job: %s", jobName)
    return nil
}

// Helper functions to avoid nil pointers
func intPtr(i int32) *int32 { return &i }
func resourceQuantity(s string) resource.Quantity {
    q, _ := resource.ParseQuantity(s)
    return q
}

func main() {
    // Load kubeconfig from default path (~/.kube/config)
    config, err := clientcmd.BuildConfigFromFlags("", clientcmd.RecommendedHomeFile)
    if err != nil {
        log.Fatalf("Failed to load kubeconfig: %v", err)
    }

    // Create Kubernetes clientset
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        log.Fatalf("Failed to create clientset: %v", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // Submit job with timestamp to avoid name conflicts
    jobName := fmt.Sprintf("ml-train-batch-%d", time.Now().Unix())
    if err := submitK8sIndexedJob(ctx, clientset, jobName); err != nil {
        log.Fatalf("Job submission failed: %v", err)
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    nomad "github.com/hashicorp/nomad/api"
)

// submitNomadParameterizedJob submits a Nomad 1.9 Parameterized Job for batch ML training
// with 10 completions, 2 parallel workers, and retry handling
func submitNomadParameterizedJob(ctx context.Context, client *nomad.Client, jobID string) error {
    // Define the Parameterized Job spec for Nomad 1.9
    // Note: Nomad 1.9 does not have native Indexed Job support, so we use parameterized jobs with meta
    job := &nomad.Job{
        ID:          &jobID,
        Name:        &jobID,
        Type:        stringToPtr("batch"),
        Datacenters: []string{"dc1"},
        Meta: map[string]string{
            "workload": "batch-ml",
            "version":  "1.9",
        },
        TaskGroups: []*nomad.TaskGroup{
            {
                Name: stringToPtr("ml-train-group"),
                Tasks: []*nomad.Task{
                    {
                        Name:   "ml-trainer",
                        Driver: stringToPtr("docker"),
                        Config: map[string]interface{}{
                            "image": "us-east-1.docker.io/myorg/ml-trainer:1.3.2",
                            "args": []string{
                                "sh", "-c",
                                // Use NOMAD_META_SHARD_INDEX passed via dispatch meta
                                "echo Training shard $NOMAD_META_SHARD_INDEX; python train.py --shard $NOMAD_META_SHARD_INDEX --total-shards 10",
                            },
                        },
                        Resources: &nomad.Resources{
                            CPU:      intToPtr(2000), // 2 CPU cores
                            MemoryMB: intToPtr(4096), // 4GB RAM
                        },
                        Env: map[string]string{
                            // Meta is passed via dispatch, not natively set per job
                        },
                    },
                },
                Count: intToPtr(1), // Nomad uses dispatch to create instances, not native parallelism for parameterized jobs
            },
        },
        // Parameterized job config for Nomad 1.9
        ParameterizedJob: &nomad.ParameterizedJobConfig{
            Payload: "optional",
            Meta: map[string]string{
                "shard_index": "", // Passed via dispatch
            },
        },
        // Retry config for failed tasks
        ReschedulePolicy: &nomad.ReschedulePolicy{
            Attempts:      intToPtr(3),
            Interval:      timeToPtr(10 * time.Minute),
            Delay:         timeToPtr(30 * time.Second),
            DelayFunction: stringToPtr("exponential"),
        },
    }

    // Register the parameterized job first
    _, _, err := client.Jobs().Register(job, nil)
    if err != nil {
        return fmt.Errorf("failed to register Nomad parameterized job %s: %w", jobID, err)
    }

    // Dispatch 10 instances (shards) manually, since Nomad 1.9 lacks native Indexed Job completion tracking
    for i := 0; i < 10; i++ {
        meta := map[string]string{
            "shard_index": fmt.Sprintf("%d", i),
        }
        _, _, err := client.Jobs().Dispatch(jobID, nil, meta, nil)
        if err != nil {
            return fmt.Errorf("failed to dispatch shard %d: %w", i, err)
        }
        log.Printf("Dispatched Nomad shard %d for job %s", i, jobID)
    }

    log.Printf("Successfully submitted Nomad 1.9 Parameterized Job: %s", jobID)
    return nil
}

// Helper functions to avoid nil pointers
func stringToPtr(s string) *string { return &s }
func intToPtr(i int) *int { return &i }
func timeToPtr(t time.Duration) *time.Duration { return &t }

func main() {
    // Load Nomad config from default path (~/.nomad.d/nomad.hcl)
    client, err := nomad.NewClient(nomad.DefaultConfig())
    if err != nil {
        log.Fatalf("Failed to create Nomad client: %v", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    // Submit job with timestamp to avoid name conflicts
    jobID := fmt.Sprintf("ml-train-batch-%d", time.Now().Unix())
    if err := submitNomadParameterizedJob(ctx, client, jobID); err != nil {
        log.Fatalf("Job submission failed: %v", err)
    }
}

package main

import (
    "context"
    "fmt"
    "log"
    "sort"
    "sync"
    "time"

    batchv1 "k8s.io/api/batch/v1"
    corev1 "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/apimachinery/pkg/api/resource"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
)

// benchmarkK8sBatchStartup benchmarks startup time for 100 K8s 1.32 batch jobs
// Measures time from job creation to first pod running
func benchmarkK8sBatchStartup(ctx context.Context, clientset *kubernetes.Clientset, namespace string) ([]time.Duration, error) {
    var (
        wg      sync.WaitGroup
        mu      sync.Mutex
        latencies []time.Duration
    )

    // Submit 100 batch jobs concurrently
    for i := 0; i < 100; i++ {
        wg.Add(1)
        go func(shard int) {
            defer wg.Done()
            jobName := fmt.Sprintf("bench-job-%d-%d", time.Now().Unix(), shard)

            // Create a simple batch job
            job := &batchv1.Job{
                ObjectMeta: metav1.ObjectMeta{
                    Name:      jobName,
                    Namespace: namespace,
                },
                Spec: batchv1.JobSpec{
                    Completions:           intPtr(1),
                    Parallelism:           intPtr(1),
                    BackoffLimit:          intPtr(1),
                    CompletionMode:        batchv1.NonIndexedCompletionMode,
                    Template: corev1.PodTemplateSpec{
                        ObjectMeta: metav1.ObjectMeta{
                            Labels: map[string]string{"benchmark": "startup"},
                        },
                        Spec: corev1.PodSpec{
                            RestartPolicy: corev1.RestartPolicyNever,
                            Containers: []corev1.Container{
                                {
                                    Name:    "pause",
                                    Image:   "gcr.io/google-containers/pause:3.9",
                                    Command: []string{"sleep", "30"},
                                },
                            },
                        },
                    },
                },
            }

            // Record start time
            start := time.Now()
            _, err := clientset.BatchV1().Jobs(namespace).Create(ctx, job, metav1.CreateOptions{})
            if err != nil {
                log.Printf("Failed to create job %s: %v", jobName, err)
                return
            }

            // Watch for pod to enter Running phase
            watcher, err := clientset.CoreV1().Pods(namespace).Watch(ctx, metav1.ListOptions{
                LabelSelector: fmt.Sprintf("job-name=%s", jobName),
            })
            if err != nil {
                log.Printf("Failed to watch pods for job %s: %v", jobName, err)
                return
            }
            defer watcher.Stop()

            // Wait for first pod to be running
            for event := range watcher.ResultChan() {
                pod, ok := event.Object.(*corev1.Pod)
                if !ok {
                    continue
                }
                if pod.Status.Phase == corev1.PodRunning {
                    latency := time.Since(start)
                    mu.Lock()
                    latencies = append(latencies, latency)
                    mu.Unlock()
                    // Clean up job
                    clientset.BatchV1().Jobs(namespace).Delete(ctx, jobName, metav1.DeleteOptions{})
                    break
                }
            }
        }(i)
    }

    wg.Wait()
    return latencies, nil
}

// Helper functions
func intPtr(i int32) *int32 { return &i }
func resourceQuantity(s string) resource.Quantity { q, _ := resource.ParseQuantity(s); return q }

func main() {
    config, err := clientcmd.BuildConfigFromFlags("", clientcmd.RecommendedHomeFile)
    if err != nil {
        log.Fatalf("Failed to load kubeconfig: %v", err)
    }

    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        log.Fatalf("Failed to create clientset: %v", err)
    }

    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
    defer cancel()

    latencies, err := benchmarkK8sBatchStartup(ctx, clientset, "default")
    if err != nil {
        log.Fatalf("Benchmark failed: %v", err)
    }

    // Calculate p50, p99 startup latency
    if len(latencies) == 0 {
        log.Fatal("No latencies recorded")
    }
    // Sort latencies
    sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })
    p50 := latencies[len(latencies)/2]
    p99 := latencies[int(float64(len(latencies))*0.99)]
    avg := time.Duration(0)
    for _, l := range latencies {
        avg += l
    }
    avg = avg / time.Duration(len(latencies))

    log.Printf("K8s 1.32 Batch Startup Benchmark (100 jobs):")
    log.Printf("p50: %v, p99: %v, avg: %v", p50, p99, avg)
}

# Snippet: K8s 1.32 Indexed Job Spec
apiVersion: batch/v1
kind: Job
metadata:
  name: ml-train-sharded
spec:
  completions: 10
  parallelism: 2
  completionMode: Indexed
  template:
    spec:
      containers:
      - name: trainer
        image: myorg/ml-train:1.3
        env:
        - name: JOB_COMPLETION_INDEX
          valueFrom:
            jobCompletionIndex: {}

# Snippet: K8s 1.32 Batch PriorityClass
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: batch-low-priority
value: 100
globalDefault: false
description: "Low priority for batch workloads, preempted by serving jobs"

# Snippet: Tekton PipelineRun for Batch Test Suite
apiVersion: tekton.dev/v1beta1
kind: PipelineRun
metadata:
  name: batch-test-run
spec:
  pipelineRef:
    name: parallel-test-pipeline
  params:
  - name: total-shards
    value: "10"
  taskRunSpecs:
  - pipelineTaskName: run-tests
    taskServiceAccountName: test-sa