diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index e8f2ab7..65bbcd6 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -4,6 +4,12 @@ kind: ClusterRole metadata: name: manager-role rules: +- apiGroups: + - "" + resources: + - nodes + verbs: + - list - apiGroups: - "" resources: diff --git a/config/rbac/spindle_job_service_account.yaml b/config/rbac/spindle_job_service_account.yaml index f62b79d..5a4f10a 100644 --- a/config/rbac/spindle_job_service_account.yaml +++ b/config/rbac/spindle_job_service_account.yaml @@ -7,6 +7,8 @@ metadata: name: spindle-job-runner namespace: system automountServiceAccountToken: false +imagePullSecrets: +- name: atcr-login --- # Note: No Role or RoleBinding created intentionally # Job pods should have no permissions to read Secrets, list Pods, etc. diff --git a/internal/controller/spindleset_controller.go b/internal/controller/spindleset_controller.go index 983ec72..9d51559 100644 --- a/internal/controller/spindleset_controller.go +++ b/internal/controller/spindleset_controller.go @@ -55,6 +55,7 @@ type SpindleSetReconciler struct { // +kubebuilder:rbac:groups=loom.j5t.io,resources=spindlesets/finalizers,verbs=update // +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=batch,resources=jobs/status,verbs=get +// +kubebuilder:rbac:groups="",resources=nodes,verbs=list // +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch // +kubebuilder:rbac:groups="",resources=pods/log,verbs=get // +kubebuilder:rbac:groups="",resources=secrets,verbs=get;list;watch @@ -345,6 +346,12 @@ func (r *SpindleSetReconciler) ensurePipelineJobs(ctx context.Context, spindleSe } } + // List nodes for profile selection (to validate nodeSelector labels exist) + var nodeList corev1.NodeList + if err := r.Client.List(ctx, &nodeList); err != nil { + return fmt.Errorf("failed to list nodes: %w", err) + } + // Convert workflow steps to jobbuilder format and create Jobs for each workflow for _, workflowSpec := range pipelineRun.Workflows { // Check if Job already exists @@ -396,7 +403,7 @@ func (r *SpindleSetReconciler) ensurePipelineJobs(ctx context.Context, spindleSe } // Create the Job - job, err := jobbuilder.BuildJob(jobConfig) + job, err := jobbuilder.BuildJob(jobConfig, &nodeList) if err != nil { return fmt.Errorf("failed to build job for workflow %s: %w", workflowSpec.Name, err) } diff --git a/internal/jobbuilder/job_template.go b/internal/jobbuilder/job_template.go index 94e8a7f..53a1fdc 100644 --- a/internal/jobbuilder/job_template.go +++ b/internal/jobbuilder/job_template.go @@ -61,18 +61,48 @@ type WorkflowConfig struct { Namespace string } -// selectResourceProfile selects the first resource profile matching the workflow architecture. +// nodeMatchesSelector returns true if at least one node has all the labels in selector. +func nodeMatchesSelector(nodes *corev1.NodeList, selector map[string]string) bool { + if nodes == nil { + return false + } + for _, node := range nodes.Items { + if labelsMatch(node.Labels, selector) { + return true + } + } + return false +} + +// labelsMatch returns true if nodeLabels contains all key-value pairs from selector. +func labelsMatch(nodeLabels, selector map[string]string) bool { + for key, value := range selector { + if nodeLabels[key] != value { + return false + } + } + return true +} + +// selectResourceProfile selects the first resource profile matching the workflow architecture +// and whose nodeSelector labels all exist on at least one available node. // Returns the profile's resources and nodeSelector, or default values if no match is found. -func selectResourceProfile(profiles []loomv1alpha1.ResourceProfile, architecture string) (corev1.ResourceRequirements, map[string]string) { +func selectResourceProfile(profiles []loomv1alpha1.ResourceProfile, architecture string, nodes *corev1.NodeList) (corev1.ResourceRequirements, map[string]string) { // Iterate through profiles to find first match for _, profile := range profiles { // Check if profile's nodeSelector has the matching architecture - if arch, ok := profile.NodeSelector["kubernetes.io/arch"]; ok && arch == architecture { + arch, ok := profile.NodeSelector["kubernetes.io/arch"] + if !ok || arch != architecture { + continue + } + + // Check if ALL nodeSelector labels exist on at least one node + if nodeMatchesSelector(nodes, profile.NodeSelector) { return profile.Resources, profile.NodeSelector } } - // No profile matched - return defaults + // No profile matched - return defaults with just architecture selector return corev1.ResourceRequirements{ Requests: corev1.ResourceList{ corev1.ResourceCPU: resource.MustParse("500m"), @@ -82,11 +112,12 @@ func selectResourceProfile(profiles []loomv1alpha1.ResourceProfile, architecture corev1.ResourceCPU: resource.MustParse("2"), corev1.ResourceMemory: resource.MustParse("4Gi"), }, - }, nil + }, map[string]string{"kubernetes.io/arch": architecture} } // BuildJob creates a Kubernetes Job specification for running a spindle workflow. -func BuildJob(config WorkflowConfig) (*batchv1.Job, error) { +// The nodes parameter is used to validate that resource profile nodeSelectors can be satisfied. +func BuildJob(config WorkflowConfig, nodes *corev1.NodeList) (*batchv1.Job, error) { if config.WorkflowName == "" { return nil, fmt.Errorf("workflow name is required") } @@ -106,8 +137,8 @@ func BuildJob(config WorkflowConfig) (*batchv1.Job, error) { return nil, fmt.Errorf("failed to marshal workflow spec: %w", err) } - // Select resource profile based on workflow architecture - resources, profileNodeSelector := selectResourceProfile(config.Template.ResourceProfiles, config.Architecture) + // Select resource profile based on workflow architecture and available nodes + resources, profileNodeSelector := selectResourceProfile(config.Template.ResourceProfiles, config.Architecture, nodes) // Build architecture-based node affinity archAffinity := BuildArchitectureAffinity(config.Architecture) diff --git a/internal/jobbuilder/job_template_test.go b/internal/jobbuilder/job_template_test.go index 9558fc2..60e872e 100644 --- a/internal/jobbuilder/job_template_test.go +++ b/internal/jobbuilder/job_template_test.go @@ -1,25 +1,42 @@ package jobbuilder import ( + "fmt" "testing" corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/api/resource" loomv1alpha1 "tangled.org/evan.jarrett.net/loom/api/v1alpha1" ) +// Helper function to create a mock node list for tests +func makeNodeList(nodes ...map[string]string) *corev1.NodeList { + list := &corev1.NodeList{} + for i, labels := range nodes { + list.Items = append(list.Items, corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: fmt.Sprintf("node-%d", i), + Labels: labels, + }, + }) + } + return list +} + func TestSelectResourceProfile(t *testing.T) { tests := []struct { name string profiles []loomv1alpha1.ResourceProfile architecture string + nodes *corev1.NodeList wantCPU string wantMemory string wantLabels map[string]string }{ { - name: "select arm64 profile", + name: "select arm64 profile when matching node exists", profiles: []loomv1alpha1.ResourceProfile{ { NodeSelector: map[string]string{ @@ -53,14 +70,18 @@ func TestSelectResourceProfile(t *testing.T) { }, }, architecture: "arm64", - wantCPU: "1", - wantMemory: "2Gi", + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64"}, + map[string]string{"kubernetes.io/arch": "amd64"}, + ), + wantCPU: "1", + wantMemory: "2Gi", wantLabels: map[string]string{ "kubernetes.io/arch": "arm64", }, }, { - name: "select amd64 profile", + name: "select amd64 profile when matching node exists", profiles: []loomv1alpha1.ResourceProfile{ { NodeSelector: map[string]string{ @@ -86,14 +107,56 @@ func TestSelectResourceProfile(t *testing.T) { }, }, architecture: "amd64", - wantCPU: "4", - wantMemory: "8Gi", + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "amd64"}, + ), + wantCPU: "4", + wantMemory: "8Gi", wantLabels: map[string]string{ "kubernetes.io/arch": "amd64", }, }, { - name: "select first matching profile with additional labels", + name: "skip profile when additional labels dont exist on nodes", + profiles: []loomv1alpha1.ResourceProfile{ + { + NodeSelector: map[string]string{ + "kubernetes.io/arch": "arm64", + "node-tier": "large", + }, + Resources: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("4"), + corev1.ResourceMemory: resource.MustParse("8Gi"), + }, + }, + }, + { + NodeSelector: map[string]string{ + "kubernetes.io/arch": "arm64", + }, + Resources: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("1"), + corev1.ResourceMemory: resource.MustParse("2Gi"), + }, + }, + }, + }, + architecture: "arm64", + // Node exists but does NOT have node-tier=large label + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64"}, + ), + // Should skip first profile and use second one + wantCPU: "1", + wantMemory: "2Gi", + wantLabels: map[string]string{ + "kubernetes.io/arch": "arm64", + }, + }, + { + name: "select profile with additional labels when matching node exists", profiles: []loomv1alpha1.ResourceProfile{ { NodeSelector: map[string]string{ @@ -120,15 +183,19 @@ func TestSelectResourceProfile(t *testing.T) { }, }, architecture: "arm64", - wantCPU: "4", - wantMemory: "8Gi", + // Node has BOTH labels + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64", "node-tier": "large"}, + ), + wantCPU: "4", + wantMemory: "8Gi", wantLabels: map[string]string{ "kubernetes.io/arch": "arm64", "node-tier": "large", }, }, { - name: "fallback to defaults when no profile matches", + name: "fallback to defaults when no architecture profile matches", profiles: []loomv1alpha1.ResourceProfile{ { NodeSelector: map[string]string{ @@ -143,23 +210,33 @@ func TestSelectResourceProfile(t *testing.T) { }, }, architecture: "arm64", - wantCPU: "500m", - wantMemory: "1Gi", - wantLabels: nil, + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64"}, + ), + wantCPU: "500m", + wantMemory: "1Gi", + wantLabels: map[string]string{ + "kubernetes.io/arch": "arm64", + }, }, { name: "fallback to defaults when no profiles configured", profiles: []loomv1alpha1.ResourceProfile{}, architecture: "amd64", - wantCPU: "500m", - wantMemory: "1Gi", - wantLabels: nil, + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "amd64"}, + ), + wantCPU: "500m", + wantMemory: "1Gi", + wantLabels: map[string]string{ + "kubernetes.io/arch": "amd64", + }, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - gotResources, gotLabels := selectResourceProfile(tt.profiles, tt.architecture) + gotResources, gotLabels := selectResourceProfile(tt.profiles, tt.architecture, tt.nodes) // Check CPU request gotCPU := gotResources.Requests[corev1.ResourceCPU] @@ -177,7 +254,7 @@ func TestSelectResourceProfile(t *testing.T) { // Check labels if len(gotLabels) != len(tt.wantLabels) { - t.Errorf("selectResourceProfile() labels count = %d, want %d", len(gotLabels), len(tt.wantLabels)) + t.Errorf("selectResourceProfile() labels count = %d, want %d (got: %v)", len(gotLabels), len(tt.wantLabels), gotLabels) } for k, wantV := range tt.wantLabels { if gotV, ok := gotLabels[k]; !ok || gotV != wantV { @@ -192,13 +269,14 @@ func TestBuildJob(t *testing.T) { tests := []struct { name string config WorkflowConfig + nodes *corev1.NodeList wantCPU string wantMemory string wantNodeSelector map[string]string wantErr bool }{ { - name: "use arm64 profile with additional labels", + name: "use arm64 profile with additional labels when node matches", config: WorkflowConfig{ WorkflowName: "test-workflow", PipelineID: "test-pipeline", @@ -224,6 +302,9 @@ func TestBuildJob(t *testing.T) { }, }, }, + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64", "node-tier": "large"}, + ), wantCPU: "2", wantMemory: "4Gi", wantNodeSelector: map[string]string{ @@ -269,6 +350,10 @@ func TestBuildJob(t *testing.T) { }, }, }, + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64"}, + map[string]string{"kubernetes.io/arch": "amd64"}, + ), wantCPU: "4", wantMemory: "8Gi", wantNodeSelector: map[string]string{ @@ -277,7 +362,7 @@ func TestBuildJob(t *testing.T) { wantErr: false, }, { - name: "profile with multiple labels", + name: "fallback to simpler profile when labels dont match", config: WorkflowConfig{ WorkflowName: "test-workflow", PipelineID: "test-pipeline", @@ -294,6 +379,17 @@ func TestBuildJob(t *testing.T) { "node-tier": "large", "custom-label": "custom-value", }, + Resources: corev1.ResourceRequirements{ + Requests: corev1.ResourceList{ + corev1.ResourceCPU: resource.MustParse("4"), + corev1.ResourceMemory: resource.MustParse("8Gi"), + }, + }, + }, + { + NodeSelector: map[string]string{ + "kubernetes.io/arch": "arm64", + }, Resources: corev1.ResourceRequirements{ Requests: corev1.ResourceList{ corev1.ResourceCPU: resource.MustParse("1"), @@ -304,12 +400,14 @@ func TestBuildJob(t *testing.T) { }, }, }, + // Node has arm64 but NOT the other labels + nodes: makeNodeList( + map[string]string{"kubernetes.io/arch": "arm64"}, + ), wantCPU: "1", wantMemory: "2Gi", wantNodeSelector: map[string]string{ "kubernetes.io/arch": "arm64", - "node-tier": "large", - "custom-label": "custom-value", }, wantErr: false, }, @@ -317,7 +415,7 @@ func TestBuildJob(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - job, err := BuildJob(tt.config) + job, err := BuildJob(tt.config, tt.nodes) if (err != nil) != tt.wantErr { t.Errorf("BuildJob() error = %v, wantErr %v", err, tt.wantErr) return @@ -343,7 +441,7 @@ func TestBuildJob(t *testing.T) { // Check nodeSelector gotNodeSelector := job.Spec.Template.Spec.NodeSelector if len(gotNodeSelector) != len(tt.wantNodeSelector) { - t.Errorf("BuildJob() nodeSelector count = %d, want %d", len(gotNodeSelector), len(tt.wantNodeSelector)) + t.Errorf("BuildJob() nodeSelector count = %d, want %d (got: %v)", len(gotNodeSelector), len(tt.wantNodeSelector), gotNodeSelector) } for k, wantV := range tt.wantNodeSelector { if gotV, ok := gotNodeSelector[k]; !ok || gotV != wantV {