diff --git a/.github/workflows/dryrun.yml b/.github/workflows/dryrun.yml index 56803f1..ce31d57 100644 --- a/.github/workflows/dryrun.yml +++ b/.github/workflows/dryrun.yml @@ -8,7 +8,9 @@ on: pull_request: paths: - 'internal/emitter/**' + - 'internal/parser/**' - 'internal/splat/**' + - 'internal/k8senc/**' - 'internal/*/types.go' - 'scripts/dryrun/**' - '.github/workflows/dryrun.yml' diff --git a/internal/jobset/tosplat.go b/internal/jobset/tosplat.go new file mode 100644 index 0000000..d6ab688 --- /dev/null +++ b/internal/jobset/tosplat.go @@ -0,0 +1,92 @@ +package jobset + +import ( + corev1 "k8s.io/api/core/v1" + + "github.com/InsightSoftmax/BAMMM/internal/k8senc" + "github.com/InsightSoftmax/BAMMM/internal/splat" +) + +// DefaultImage is the placeholder image the Kueue/YuniKorn JobSet emitters use +// for script and executable tasks that carry no source container. The parser +// uses it to recognize that such a container is really an inlined script. +const DefaultImage = "ubuntu:22.04" + +// ToTasks converts a JobSet's replicatedJobs into SPLAT tasks, plus the shared +// job-level volumes and the maximum in-pod retry (BackoffLimit) observed. It is +// the inverse of the Kueue and YuniKorn JobSet emitters; scheduler-specific +// metadata (queue labels, gang annotations) stays with the caller. +func ToTasks(js *JobSet) (tasks []splat.Task, volumes []splat.Volume, maxRetries int) { + seen := map[string]splat.Volume{} + for i := range js.Spec.ReplicatedJobs { + rj := &js.Spec.ReplicatedJobs[i] + spec := &rj.Template.Spec + if bl := spec.BackoffLimit; bl != nil && int(*bl) > maxRetries { + maxRetries = int(*bl) + } + + task := splat.Task{Name: rj.Name, Replicas: replicasOf(spec.Parallelism, spec.Completions)} + pod := &spec.Template.Spec + if len(pod.Containers) > 0 { + c := pod.Containers[0] + task.Resources = k8senc.ResourcesFromContainer(&c) + task.Execution = executionOf(&c) + k8senc.VolumesFromPod(&c, pod, seen) + } + task.Placement = placementOf(pod) + tasks = append(tasks, task) + } + return tasks, k8senc.SortVolumes(seen), maxRetries +} + +// replicasOf recovers a task's replica count from the wrapped Job's parallelism +// (the emitters set parallelism/completions only when replicas exceed 1). +func replicasOf(parallelism, completions *int32) int { + switch { + case parallelism != nil && *parallelism > 0: + return int(*parallelism) + case completions != nil && *completions > 0: + return int(*completions) + default: + return 1 + } +} + +// executionOf inverts the emitters' container: an inlined "/bin/bash -c