Move the nodegroups one level up (#2227)

Author: RobertLucian

pkg/operator/resources/asyncapi/k8s_specs.go

@@ -88,7 +88,7 @@ func gatewayDeploymentSpec(api spec.API, prevDeployment *kapps.Deployment, queue
 				Containers:                    []kcore.Container{container},
 				NodeSelector:                  workloads.NodeSelectors(),
 				Tolerations:                   workloads.GenerateResourceTolerations(),
-				Affinity:                      workloads.GenerateNodeAffinities(api.Pod.NodeGroups),
+				Affinity:                      workloads.GenerateNodeAffinities(api.NodeGroups),
 				Volumes:                       volumes,
 				ServiceAccountName:            workloads.ServiceAccountName,
 			},
@@ -239,7 +239,7 @@ func deploymentSpec(api spec.API, prevDeployment *kapps.Deployment, queueURL str
 				Containers:                    containers,
 				NodeSelector:                  workloads.NodeSelectors(),
 				Tolerations:                   workloads.GenerateResourceTolerations(),
-				Affinity:                      workloads.GenerateNodeAffinities(api.Pod.NodeGroups),
+				Affinity:                      workloads.GenerateNodeAffinities(api.NodeGroups),
 				Volumes:                       volumes,
 				ServiceAccountName:            workloads.ServiceAccountName,
 			},

pkg/operator/resources/job/batchapi/job.go

@@ -145,7 +145,7 @@ func SubmitJob(apiName string, submission *schema.BatchJobSubmission) (*spec.Bat
 			Timeout:         timeout,
 			DeadLetterQueue: deadLetterQueue,
 			TTL:             &kmeta.Duration{Duration: _batchJobTTL},
-			NodeGroups:      apiSpec.Pod.NodeGroups,
+			NodeGroups:      apiSpec.NodeGroups,
 			Probes:          workloads.GetReadinessProbesFromContainers(apiSpec.Pod.Containers),
 		},
 	}

pkg/operator/resources/job/taskapi/k8s_specs.go

@@ -94,7 +94,7 @@ func k8sJobSpec(api *spec.API, job *spec.TaskJob) *kbatch.Job {
 				Containers:         containers,
 				NodeSelector:       workloads.NodeSelectors(),
 				Tolerations:        workloads.GenerateResourceTolerations(),
-				Affinity:           workloads.GenerateNodeAffinities(api.Pod.NodeGroups),
+				Affinity:           workloads.GenerateNodeAffinities(api.NodeGroups),
 				Volumes:            volumes,
 				ServiceAccountName: workloads.ServiceAccountName,
 			},

pkg/operator/resources/realtimeapi/k8s_specs.go

@@ -68,7 +68,7 @@ func deploymentSpec(api *spec.API, prevDeployment *kapps.Deployment) *kapps.Depl
 				Containers:                    containers,
 				NodeSelector:                  workloads.NodeSelectors(),
 				Tolerations:                   workloads.GenerateResourceTolerations(),
-				Affinity:                      workloads.GenerateNodeAffinities(api.Pod.NodeGroups),
+				Affinity:                      workloads.GenerateNodeAffinities(api.NodeGroups),
 				Volumes:                       volumes,
 				ServiceAccountName:            workloads.ServiceAccountName,
 			},

pkg/operator/resources/validations.go

@@ -138,7 +138,7 @@ func validateK8sCompute(api *userconfig.API, maxMemMap map[string]kresource.Quan
 	successfulLoops := 0
 
 	clusterNodeGroupNames := strset.New(config.ClusterConfig.GetNodeGroupNames()...)
-	apiNodeGroupNames := api.Pod.NodeGroups
+	apiNodeGroupNames := api.NodeGroups
 
 	if apiNodeGroupNames != nil {
 		for _, ngName := range apiNodeGroupNames {

pkg/types/spec/validations.go

@@ -52,25 +52,29 @@ func apiValidation(resource userconfig.Resource) *cr.StructValidation {
 	case userconfig.RealtimeAPIKind:
 		structFieldValidations = append(resourceStructValidations,
 			podValidation(userconfig.RealtimeAPIKind),
+			nodegroupsValidation(),
 			networkingValidation(),
 			autoscalingValidation(resource.Kind),
 			updateStrategyValidation(),
 		)
 	case userconfig.AsyncAPIKind:
 		structFieldValidations = append(resourceStructValidations,
 			podValidation(userconfig.AsyncAPIKind),
+			nodegroupsValidation(),
 			networkingValidation(),
 			autoscalingValidation(resource.Kind),
 			updateStrategyValidation(),
 		)
 	case userconfig.BatchAPIKind:
 		structFieldValidations = append(resourceStructValidations,
 			podValidation(userconfig.BatchAPIKind),
+			nodegroupsValidation(),
 			networkingValidation(),
 		)
 	case userconfig.TaskAPIKind:
 		structFieldValidations = append(resourceStructValidations,
 			podValidation(userconfig.TaskAPIKind),
+			nodegroupsValidation(),
 			networkingValidation(),
 		)
 	case userconfig.TrafficSplitterKind:
@@ -144,18 +148,6 @@ func podValidation(kind userconfig.Kind) *cr.StructFieldValidation {
 		StructField: "Pod",
 		StructValidation: &cr.StructValidation{
 			StructFieldValidations: []*cr.StructFieldValidation{
-				{
-					StructField: "NodeGroups",
-					StringListValidation: &cr.StringListValidation{
-						Required:          false,
-						Default:           nil,
-						AllowExplicitNull: true,
-						AllowEmpty:        false,
-						ElementStringValidation: &cr.StringValidation{
-							AlphaNumericDashUnderscore: true,
-						},
-					},
-				},
 				{
 					StructField: "Port",
 					Int32PtrValidation: &cr.Int32PtrValidation{
@@ -267,6 +259,21 @@ func containersValidation(kind userconfig.Kind) *cr.StructFieldValidation {
 	}
 }
 
+func nodegroupsValidation() *cr.StructFieldValidation {
+	return &cr.StructFieldValidation{
+		StructField: "NodeGroups",
+		StringListValidation: &cr.StringListValidation{
+			Required:          false,
+			Default:           nil,
+			AllowExplicitNull: true,
+			AllowEmpty:        false,
+			ElementStringValidation: &cr.StringValidation{
+				AlphaNumericDashUnderscore: true,
+			},
+		},
+	}
+}
+
 func networkingValidation() *cr.StructFieldValidation {
 	return &cr.StructFieldValidation{
 		StructField: "Networking",

pkg/types/userconfig/api.go

@@ -34,6 +34,7 @@ type API struct {
 	Resource
 
 	Pod              *Pod            `json:"pod" yaml:"pod"`
+	NodeGroups       []string        `json:"node_groups" yaml:"node_groups"`
 	APIs             []*TrafficSplit `json:"apis" yaml:"apis"`
 	Networking       *Networking     `json:"networking" yaml:"networking"`
 	Autoscaling      *Autoscaling    `json:"autoscaling" yaml:"autoscaling"`
@@ -44,7 +45,6 @@ type API struct {
 }
 
 type Pod struct {
-	NodeGroups     []string     `json:"node_groups" yaml:"node_groups"`
 	Port           *int32       `json:"port" yaml:"port"`
 	MaxQueueLength int64        `json:"max_queue_length" yaml:"max_queue_length"`
 	MaxConcurrency int64        `json:"max_concurrency" yaml:"max_concurrency"`
@@ -270,6 +270,12 @@ func (api *API) UserStr() string {
 		sb.WriteString(s.Indent(api.Autoscaling.UserStr(), "  "))
 	}
 
+	if api.NodeGroups == nil {
+		sb.WriteString(fmt.Sprintf("%s: null\n", NodeGroupsKey))
+	} else {
+		sb.WriteString(fmt.Sprintf("%s: %s\n", NodeGroupsKey, s.ObjFlatNoQuotes(api.NodeGroups)))
+	}
+
 	if api.UpdateStrategy != nil {
 		sb.WriteString(fmt.Sprintf("%s:\n", UpdateStrategyKey))
 		sb.WriteString(s.Indent(api.UpdateStrategy.UserStr(), "  "))
@@ -288,12 +294,6 @@ func (trafficSplit *TrafficSplit) UserStr() string {
 
 func (pod *Pod) UserStr(kind Kind) string {
 	var sb strings.Builder
-
-	if pod.NodeGroups == nil {
-		sb.WriteString(fmt.Sprintf("%s: null\n", NodeGroupsKey))
-	} else {
-		sb.WriteString(fmt.Sprintf("%s: %s\n", NodeGroupsKey, s.ObjFlatNoQuotes(pod.NodeGroups)))
-	}
 	if pod.Port != nil {
 		sb.WriteString(fmt.Sprintf("%s: %d\n", PortKey, *pod.Port))
 	}
@@ -535,7 +535,6 @@ func (api *API) TelemetryEvent() map[string]interface{} {
 
 	if api.Pod != nil {
 		event["pod._is_defined"] = true
-		event["pod.node_groups._len"] = len(api.Pod.NodeGroups)
 		if api.Pod.Port != nil {
 			event["pod.port"] = *api.Pod.Port
 		}
@@ -576,6 +575,8 @@ func (api *API) TelemetryEvent() map[string]interface{} {
 		event["pod.containers.compute.inf"] = totalCompute.Inf
 	}
 
+	event["node_groups._len"] = len(api.NodeGroups)
+
 	if api.UpdateStrategy != nil {
 		event["update_strategy._is_defined"] = true
 		event["update_strategy.max_surge"] = api.UpdateStrategy.MaxSurge