Repository: armadaproject/armada
Pull request #3455 (2024-03-12 10:31:16)

Scheduler and submission config refactoring and cleanup
Move config relating to job submission from `SchedulingConfig` into a new struct `SubmissionConfig`. Misc. other config refactoring and cleanup. Improved docstrings/comments.


Patch:
diff --git a/config/armada/config.yaml b/config/armada/config.yaml
index b6ee4fc2..80f44dff 100644
--- a/config/armada/config.yaml
+++ b/config/armada/config.yaml
@@ -31,52 +31,43 @@ eventsApiRedis:
   password: ""
   db: 1
   poolSize: 1000
+# Scheduling config used by the job submitChecker. Only needs the following subset of config options.
+# You may also wish to configure indexedNodeLabels and indexedTaints to speed up checking.
+#
+# This config must be consistent with the scheduling config used by the scheduler.
+# You may want to insert the scheduling config used for the scheduler automatically, e.g., using PyYAML, to guarantee consistency.
 scheduling:
-  enableAssertions: true
-  fairnessModel: "AssetFairness"
-  dominantResourceFairnessResourcesToConsider:
-    - "cpu"
-    - "memory"
-    - "nvidia.com/gpu"
-  resourceScarcity:
-    cpu: 1.0
-  preemption:
-    nodeEvictionProbability: 1.0
-    nodeOversubscriptionEvictionProbability: 1.0
-    protectedFractionOfFairShare: 1.0
-    setNodeIdSelector: true
-    nodeIdLabel: kubernetes.io/hostname
-    setNodeName: false
-    priorityClasses:
-      armada-default:
-        priority: 1000
-        preemptible: false
-        maximumResourceFractionPerQueue:
-          memory: 1.0
-          cpu: 1.0
-      armada-preemptible:
-        priority: 1000
-        preemptible: true
-    defaultPriorityClass: armada-default
-    priorityClassNameOverride: armada-default
-  executorTimeout: 60m
-  maxQueueLookback: 1000
-  maxExtraNodesToConsider: 1
-  maximumResourceFractionToSchedule:
-    memory: 1.0
-    cpu: 1.0
-  maximumSchedulingRate: 100.0
-  maximumSchedulingBurst: 1000
-  maximumPerQueueSchedulingRate: 50.0
-  maximumPerQueueSchedulingBurst: 1000
-  maxJobSchedulingContextsPerExecutor: 10000
-  lease:
-    expireAfter: 15m
-    expiryLoopInterval: 5s
+  executorTimeout: "60m"
+  executorUpdateFrequency: "1m"
+  priorityClasses:
+    armada-default:
+      priority: 1000
+      preemptible: false
+      maximumResourceFractionPerQueue:
+        memory: 1.0
+        cpu: 1.0
+    armada-preemptible:
+      priority: 1000
+      preemptible: true
+  indexedResources:
+    - name: "nvidia.com/gpu"
+      resolution: "1"
+    - name: "cpu"
+      resolution: "100m"
+    - name: "memory"
+      resolution: "100Mi"
+    - name: "ephemeral-storage"
+      resolution: "1Gi"
+submission:
+  allowedPriorityClassNames:
+    armada-default: true
+    armada-preemptible: true
+    armada-resilient: true
+  defaultPriorityClassName: "armada-default"
   defaultJobLimits:
-    cpu: 1
-    memory: 1Gi
-    ephemeral-storage: 8Gi
+    cpu: "1"
+    memory: "1Gi"
+    ephemeral-storage: "8Gi"
   defaultJobTolerations:
     - key: "armadaproject.io/armada"
       operator: "Equal"
@@ -93,18 +84,20 @@ scheduling:
         operator: "Equal"
         value: "true"
         effect: "NoSchedule"
-  maxRetries: 5
+  defaultJobTolerationsByResourceRequest:
+    - "nvidia.com/gpu":
+        - key: "armadaproject.io/gpu"
+          operator: "Equal"
+          value: "true"
+          effect: "NoSchedule"
   maxPodSpecSizeBytes: 65535
   minJobResources:
-    memory: 1Mi
-  indexedResources:
-    - name: "cpu"
-      resolution: "100m"
-    - name: "memory"
-      resolution: "1Mi"
-  minTerminationGracePeriod: 1s
-  maxTerminationGracePeriod: 300s
-  executorUpdateFrequency: 1m
+    memory: "1Mi"
+  minTerminationGracePeriod: "1s"
+  maxTerminationGracePeriod: "5m"
+  defaultActiveDeadline: "72h" # 3 days.
+  defaultActiveDeadlineByResourceRequest:
+    nvidia.com/gpu: "336h" # 14 days.
 eventRetention:
   expiryEnabled: true
   retentionDuration: 336h
diff --git a/config/scheduler/config.yaml b/config/scheduler/config.yaml
index 9845b8cb..71e84baf 100644
--- a/config/scheduler/config.yaml
+++ b/config/scheduler/config.yaml
@@ -74,38 +74,27 @@ grpc:
     permitWithoutStream: true
   tls:
     enabled: false
+# You may want to configure indexedNodeLabels and indexedTaints to speed up scheduling.
 scheduling:
-  executorTimeout: 10m
-  executorUpdateFrequency: 1m
-  enableAssertions: true
-  fairnessModel: "AssetFairness"
-  dominantResourceFairnessResourcesToConsider:
-    - "cpu"
-    - "memory"
-    - "nvidia.com/gpu"
-  resourceScarcity:
-    cpu: 1.0
-  preemption:
-    alwaysAttemptScheduling: false
-    enabled: true
-    nodeEvictionProbability: 1.0
-    nodeOversubscriptionEvictionProbability: 1.0
-    protectedFractionOfFairShare: 1.0
-    nodeIdLabel: kubernetes.io/hostname
-    priorityClasses:
-      armada-default:
-        priority: 1000
-        preemptible: false
-        maximumResourceFractionPerQueue:
-          memory: 1.0
-          cpu: 1.0
-      armada-preemptible:
-        priority: 1000
-        preemptible: true
-    defaultPriorityClass: armada-default
-    priorityClassNameOverride: armada-default
-  maxQueueLookback: 1000
-  maxExtraNodesToConsider: 1
+  disableScheduling: false
+  enableAssertions: false
+  nodeEvictionProbability: 1.0
+  nodeOversubscriptionEvictionProbability: 1.0
+  protectedFractionOfFairShare: 1.0
+  nodeIdLabel: "kubernetes.io/hostname"
+  priorityClasses:
+    armada-default:
+      priority: 1000
+      preemptible: false
+      maximumResourceFractionPerQueue:
+        memory: 1.0
+        cpu: 1.0
+    armada-preemptible:
+      priority: 1000
+      preemptible: true
+  defaultPriorityClassName: "armada-default"
+  priorityClassNameOverride: "armada-default"
+  maxQueueLookback: 100000
   maximumResourceFractionToSchedule:
     memory: 1.0
     cpu: 1.0
@@ -113,36 +102,26 @@ scheduling:
   maximumSchedulingBurst: 1000
   maximumPerQueueSchedulingRate: 50.0
   maximumPerQueueSchedulingBurst: 1000
-  maxUnacknowledgedJobsPerExecutor: 2500
   maxJobSchedulingContextsPerExecutor: 10000
-  defaultJobLimits:
-    cpu: 1
-    memory: 1Gi
-    ephemeral-storage: 8Gi
-  defaultJobTolerations:
-    - key: "armadaproject.io/armada"
-      operator: "Equal"
-      value: "true"
-      effect: "NoSchedule"
-  defaultJobTolerationsByPriorityClass:
-    armada-default:
-      - key: "armadaproject.io/pc-armada-default"
-        operator: "Equal"
-        value: "true"
-        effect: "NoSchedule"
-    armada-preemptible:
-      - key: "armadaproject.io/pc-armada-preemptible"
-        operator: "Equal"
-        value: "true"
-        effect: "NoSchedule"
-  maxRetries: 5
+  maxRetries: 3
+  dominantResourceFairnessResourcesToConsider:
+    - "cpu"
+    - "memory"
+    - "nvidia.com/gpu"
+    - "ephemeral-storage"
   indexedResources:
+    - name: "nvidia.com/gpu"
+      resolution: "1"
     - name: "cpu"
       resolution: "100m"
     - name: "memory"
-      resolution: "1Mi"
-  gangIdAnnotation: armadaproject.io/gangId
-  gangCardinalityAnnotation: armadaproject.io/gangCardinality
+      resolution: "100Mi"
+    - name: "ephemeral-storage"
+      resolution: "1Gi"
+  executorTimeout: "10m"
+  maxUnacknowledgedJobsPerExecutor: 2500
+  alwaysAttemptScheduling: false
+  executorUpdateFrequency: "1m"
   failureEstimatorConfig:
     # Optimised default parameters.
     numInnerIterations: 10
diff --git a/internal/armada/configuration/constants.go b/internal/armada/configuration/constants.go
index d7c1c30d..49e0051f 100644
--- a/internal/armada/configuration/constants.go
+++ b/internal/armada/configuration/constants.go
@@ -1,7 +1,7 @@
 package configuration
 
 const (
-	// GangIdAnnotation Jobs with equal value for this annotation make up a gang.
+	// GangIdAnnotation maps to a unique id of the gang the job is part of; jobs with equal value make up a gang.
 	// All jobs in a gang are guaranteed to be scheduled onto the same cluster at the same time.
 	GangIdAnnotation = "armadaproject.io/gangId"
 	// GangCardinalityAnnotation All jobs in a gang must specify the total number of jobs in the gang via this annotation.
@@ -14,16 +14,10 @@ const (
 	// Specifically, if provided, all gang jobs are scheduled onto nodes for which the value of the provided label is equal.
 	// Used to ensure, e.g., that all gang jobs are scheduled onto the same cluster or rack.
 	GangNodeUniformityLabelAnnotation = "armadaproject.io/gangNodeUniformityLabel"
-	// Armada normally tries to re-schedule jobs for which a pod fails to start.
-	// Pods for which this annotation has value "true" are not retried.
+	// GangNumJobsScheduledAnnotation is set by the scheduler and indicates how many gang jobs were scheduled.
+	// For example, a gang composed of 4 jobs may only have a subset be scheduled if  GangMinimumCardinalityAnnotation < 4.
+	GangNumJobsScheduledAnnotation = "armadaproject.io/numGangJobsScheduled"
+	// FailFastAnnotation, if set to true, ensures Armada does not re-schedule jobs that fail to start.
 	// Instead, the job the pod is part of fails immediately.
 	FailFastAnnotation = "armadaproject.io/failFast"
 )
-
-const (
-	RuntimeGangCardinality = "runtime_gang_cardinality"
-)
-
-var ReturnLeaseRequestTrackedAnnotations = map[string]struct{}{
-	FailFastAnnotation: {},
-}
diff --git a/internal/armada/configuration/types.go b/internal/armada/configuration/types.go
index 4cbcdb5f..ea15750e 100644
--- a/internal/armada/configuration/types.go
+++ b/internal/armada/configuration/types.go
@@ -34,12 +34,17 @@ type ArmadaConfig struct {
 	SchedulerApiConnection client.ApiConnectionDetails
 
 	CancelJobsBatchSize int
-	Redis               redis.UniversalOptions
-	EventsApiRedis      redis.UniversalOptions
-	Scheduling          SchedulingConfig
-	Pulsar              PulsarConfig
-	Postgres            PostgresConfig // Used for Pulsar submit API deduplication
-	QueryApi            QueryApiConfig
+
+	Redis          redis.UniversalOptions
+	EventsApiRedis redis.UniversalOptions
+	Pulsar         PulsarConfig
+	Postgres       PostgresConfig // Used for Pulsar submit API deduplication
+	QueryApi       QueryApiConfig
+
+	// Config relating to job submission.
+	Submission SubmissionConfig
+	// Scheduling config used by the submitChecker.
+	Scheduling SchedulingConfig
 }
 
 type PulsarConfig struct {
@@ -80,12 +85,97 @@ type PulsarConfig struct {
 	ReceiverQueueSize int
 }
 
+// SubmissionConfig contains config relating to job submission.
+type SubmissionConfig struct {
+	// The priorityClassName field on submitted pod must be either empty or in this list.
+	// These names should correspond to priority classes defined in schedulingConfig.
+	AllowedPriorityClassNames map[string]bool
+	// Priority class name assigned to pods that do not specify one.
+	// Must be an entry in PriorityClasses above.
+	DefaultPriorityClassName string
+	// Default job resource limits added to pods.
+	DefaultJobLimits armadaresource.ComputeResources
+	// Tolerations added to all submitted pods.
+	DefaultJobTolerations []v1.Toleration
+	// Tolerations added to all submitted pods of a given priority class.
+	DefaultJobTolerationsByPriorityClass map[string][]v1.Toleration
+	// Tolerations added to all submitted pods requesting a non-zero amount of some resource.
+	DefaultJobTolerationsByResourceRequest map[string][]v1.Toleration
+	// Pods of size greater than this are rejected at submission.
+	MaxPodSpecSizeBytes uint
+	// Jobs requesting less than this amount of resources are rejected at submission.
+	MinJobResources v1.ResourceList
+	// Default value of GangNodeUniformityLabelAnnotation if not set on submitted jobs.
+	// TODO(albin): We should add a label to nodes in the nodeDb indicating which cluster it came from.
+	//              If we do, we can default to that label if the uniformity label is empty.
+	DefaultGangNodeUniformityLabel string
+	// Minimum allowed termination grace period for pods submitted to Armada.
+	// Should normally be set to a positive value, e.g., "10m".
+	// Since a zero grace period causes Kubernetes to force delete pods, which may causes issues with container resource cleanup.
+	//
+	// The grace period of pods that either
+	// - do not set a grace period, or
+	// - explicitly set a grace period of 0 seconds,
+	// is automatically set to MinTerminationGracePeriod.
+	MinTerminationGracePeriod time.Duration
+	// Max allowed grace period.
+	// Should normally not be set greater than single-digit minutes,
+	// since cancellation and preemption may need to wait for this amount of time.
+	MaxTerminationGracePeriod time.Duration
+	// Default activeDeadline for all pods that don't explicitly set activeDeadlineSeconds.
+	// Is trumped by DefaultActiveDeadlineByResourceRequest.
+	DefaultActiveDeadline time.Duration
+	// Default activeDeadline for pods with at least one container requesting a given resource.
+	// For example, if
+	// DefaultActiveDeadlineByResourceRequest: map[string]time.Duration{"gpu": time.Second},
+	// then all pods requesting a non-zero amount of gpu and don't explicitly set activeDeadlineSeconds
+	// will have activeDeadlineSeconds set to 1.
+	// Trumps DefaultActiveDeadline.
+	DefaultActiveDeadlineByResourceRequest map[string]time.Duration
+}
+
+// SchedulingConfig contains config controlling the Armada scheduler.
+//
+// The Armada scheduler is in charge of assigning pods to cluster and nodes.
+// The Armada scheduler is part of the Armada control plane.
+//
+// Features:
+// 1. Queuing and fairly dividing resources between users.
+// 2. Fair preemption, including between jobs of equal priority to balance resource allocation.
+// 3. Gang scheduling, optional across clusters, and with lower and upper bounds on the number of jobs scheduled.
+//
+// Note that Armada still relies on kube-scheduler for binding of pods to nodes.
+// This is achieved by adding to each pod created by Armada a node selector that matches only the intended node.
 type SchedulingConfig struct {
 	// Set to true to disable scheduling
 	DisableScheduling bool
 	// Set to true to enable scheduler assertions. This results in some performance loss.
 	EnableAssertions bool
-	Preemption       PreemptionConfig
+	// If using PreemptToFairShare,
+	// the probability of evicting jobs on a node to balance resource usage.
+	// TODO(albin): Remove.
+	NodeEvictionProbability float64
+	// If using PreemptToFairShare,
+	// the probability of evicting jobs on oversubscribed nodes, i.e.,
+	// nodes on which the total resource requests are greater than the available resources.
+	// TODO(albin): Remove.
+	NodeOversubscriptionEvictionProbability float64
+	// Only queues allocated more than this fraction of their fair share are considered for preemption.
+	ProtectedFractionOfFairShare float64 `validate:"gte=0"`
+	// Armada adds a node selector term to every scheduled pod using this label with the node name as value.
+	// This to force kube-scheduler to schedule pods on the node chosen by Armada.
+	// For example, if NodeIdLabel is "kubernetes.io/hostname" and armada schedules a pod on node "myNode",
+	// then Armada adds "kubernetes.io/hostname": "myNode" to the pod node selector before sending it to the executor.
+	NodeIdLabel string `validate:"required"`
+	// Map from priority class names to priority classes.
+	// Must be consistent with Kubernetes priority classes.
+	// I.e., priority classes defined here must be defined in all executor clusters and should map to the same priority.
+	PriorityClasses map[string]types.PriorityClass `validate:"dive"`
+	// Jobs with no priority class are assigned this priority class when ingested by the scheduler.
+	// Must be a key in the PriorityClasses map above.
+	DefaultPriorityClassName string
+	// If set, override the priority class name of pods with this value when sending to an executor.
+	PriorityClassNameOverride *string
 	// Number of jobs to load from the database at a time.
 	MaxQueueLookback uint
 	// In each invocation of the scheduler, no more jobs are scheduled once this limit has been exceeded.
@@ -124,83 +214,36 @@ type SchedulingConfig struct {
 	// This setting limits the number of such contexts to store.
 	// Contexts associated with the most recent scheduling attempt for each queue and cluster are always stored.
 	MaxJobSchedulingContextsPerExecutor uint
-	DefaultJobLimits                    armadaresource.ComputeResources
-	// Set of tolerations added to all submitted pods.
-	DefaultJobTolerations []v1.Toleration
-	// Set of tolerations added to all submitted pods of a given priority class.
-	DefaultJobTolerationsByPriorityClass map[string][]v1.Toleration
-	// Set of tolerations added to all submitted pods with a given resource request.
-	DefaultJobTolerationsByResourceRequest map[string][]v1.Toleration
 	// Maximum number of times a job is retried before considered failed.
 	MaxRetries uint
 	// List of resource names, e.g., []string{"cpu", "memory"}, to consider when computing DominantResourceFairness.
 	DominantResourceFairnessResourcesToConsider []string
-	MaxPodSpecSizeBytes                         uint
-	MinJobResources                             v1.ResourceList
 	// Once a node has been found on which a pod can be scheduled,
 	// the scheduler will consider up to the next maxExtraNodesToConsider nodes.
 	// The scheduler selects the node with the best score out of the considered nodes.
 	// In particular, the score expresses whether preemption is necessary to schedule a pod.
 	// Hence, a larger MaxExtraNodesToConsider would reduce the expected number of preemptions.
+	// TODO(albin): Remove. It's unused.
 	MaxExtraNodesToConsider uint
-	// Resources, e.g., "cpu", "memory", and "nvidia.com/gpu",
-	// for which the scheduler creates indexes for efficient lookup.
-	// Applies only to the new scheduler.
+	// Resources, e.g., "cpu", "memory", and "nvidia.com/gpu", for which the scheduler creates indexes for efficient lookup.
+	// This list must contain at least one resource. Adding more than one resource is not required, but may speed up scheduling.
+	// Ideally, this list contains all resources that frequently constrain which nodes a job can be scheduled onto.
 	IndexedResources []IndexedResource
 	// Node labels that the scheduler creates indexes for efficient lookup of.
-	// Should include node labels frequently used for scheduling.
-	// Since the scheduler can efficiently sort out nodes for which these labels
-	// are not set correctly when looking for a node a pod can be scheduled on.
+	// Should include node labels frequently used by node selectors on submitted jobs.
 	//
 	// If not set, no labels are indexed.
-	//
-	// Applies only to the new scheduler.
 	IndexedNodeLabels []string
 	// Taint keys that the scheduler creates indexes for efficient lookup of.
-	// Should include taints frequently used for scheduling.
-	// Since the scheduler can efficiently sort out nodes for which these taints
-	// are not set correctly when looking for a node a pod can be scheduled on.
+	// Should include keys of taints frequently used in tolerations on submitted jobs.
 	//
 	// If not set, all taints are indexed.
-	//
-	// Applies only to the new scheduler.
 	IndexedTaints []string
-	// WellKnownNodeTypes defines a set of well-known node types; these are used
-	// to define "home" and "away" nodes for a given priority class.
+	// WellKnownNodeTypes defines a set of well-known node types used to define "home" and "away" nodes for a given priority class.
 	WellKnownNodeTypes []WellKnownNodeType `validate:"dive"`
-	// Default value of GangNodeUniformityLabelAnnotation if none is provided.
-	DefaultGangNodeUniformityLabel string
-	// Kubernetes pods may specify a termination grace period.
-	// When Pods are cancelled/preempted etc., they are first sent a SIGTERM.
-	// If a pod has not exited within its termination grace period,
-	// it is killed forcefully by Kubernetes sending it a SIGKILL.
-	//
-	// This is the minimum allowed termination grace period.
-	// It should normally be set to a positive value, e.g., 1 second.
-	// Since a zero grace period causes Kubernetes to force delete pods,
-	// which may causes issues where resources associated with the pod, e.g.,
-	// containers, are not cleaned up correctly.
-	//
-	// The grace period of pods that either
-	// - do not set a grace period, or
-	// - explicitly set a grace period of 0 seconds,
-	// is automatically set to MinTerminationGracePeriod.
-	MinTerminationGracePeriod time.Duration
-	// Max allowed grace period.
-	// Should normally not be set greater than single-digit minutes,
-	// since cancellation and preemption may need to wait for this amount of time.
-	MaxTerminationGracePeriod time.Duration
-	// If an executor hasn't heartbeated in this time period, it will be considered stale
+	// Executor that haven't heartbeated in this time period are considered stale.
+	// No new jobs are scheduled onto stale executors.
 	ExecutorTimeout time.Duration
-	// Default activeDeadline for all pods that don't explicitly set activeDeadlineSeconds.
-	// Is trumped by DefaultActiveDeadlineByResourceRequest.
-	DefaultActiveDeadline time.Duration
-	// Default activeDeadline for pods with at least one container requesting a given resource.
-	// For example, if
-	// DefaultActiveDeadlineByResourceRequest: map[string]time.Duration{"gpu": time.Second},
-	// then all pods requesting a non-zero amount of gpu and don't explicitly set activeDeadlineSeconds
-	// will have activeDeadlineSeconds set to 1. Trumps DefaultActiveDeadline.
-	DefaultActiveDeadlineByResourceRequest map[string]time.Duration
 	// Maximum number of jobs that can be assigned to a executor but not yet acknowledged, before
 	// the scheduler is excluded from consideration by the scheduler.
 	MaxUnacknowledgedJobsPerExecutor uint
@@ -208,7 +251,7 @@ type SchedulingConfig struct {
 	AlwaysAttemptScheduling bool
 	// The frequency at which the scheduler updates the cluster state.
 	ExecutorUpdateFrequency time.Duration
-	// Controls node and queue success probability estimation.
+	// Controls node and queue failure probability estimation.
 	FailureEstimatorConfig FailureEstimatorConfig
 }
 
@@ -230,7 +273,7 @@ func SchedulingConfigValidation(sl validator.StructLevel) {
 		wellKnownNodeTypes[wellKnownNodeType.Name] = true
 	}
 
-	for priorityClassName, priorityClass := range c.Preemption.PriorityClasses {
+	for priorityClassName, priorityClass := range c.PriorityClasses {
 		if len(priorityClass.AwayNodeTypes) > 0 && !priorityClass.Preemptible {
 			fieldName := fmt.Sprintf("Preemption.PriorityClasses[%s].Preemptible", priorityClassName)
 			sl.ReportError(priorityClass.Preemptible, fieldName, "", AwayNodeTypesWithoutPreemptionErrorMessage, "")
@@ -245,10 +288,16 @@ func SchedulingConfigValidation(sl validator.StructLevel) {
 	}
 }
 
+// IndexedResource represents a resource the scheduler indexes for efficient lookup.
 type IndexedResource struct {
-	// Resource name. E.g., "cpu", "memory", or "nvidia.com/gpu".
+	// Resource name, e.g., "cpu", "memory", or "nvidia.com/gpu".
 	Name string
-	// See NodeDb docs.
+	// Resolution with which Armada tracks this resource; larger values indicate lower resolution.
+	// In particular, the allocatable resources on each node are rounded to a multiple of the resolution.
+	// Lower resolution speeds up scheduling by improving node lookup speed but may prevent scheduling jobs,
+	// since the allocatable resources may be rounded down to be a multiple of the resolution.
+	//
+	// See NodeDb docs for more details.
 	Resolution resource.Quantity
 }
 
@@ -261,37 +310,6 @@ type WellKnownNodeType struct {
 	Taints []v1.Taint
 }
 
-// TODO: Remove. Move PriorityClasses and DefaultPriorityClass into SchedulingConfig.
-type PreemptionConfig struct {
-	// If using PreemptToFairShare,
-	// the probability of evicting jobs on a node to balance resource usage.
-	NodeEvictionProbability float64
-	// If using PreemptToFairShare,
-	// the probability of evicting jobs on oversubscribed nodes, i.e.,
-	// nodes on which the total resource requests are greater than the available resources.
-	NodeOversubscriptionEvictionProbability float64
-	// Only queues allocated more than this fraction of their fair share are considered for preemption.
-	ProtectedFractionOfFairShare float64
-	// If true, the Armada scheduler will add to scheduled pods a node selector
-	// NodeIdLabel: <value of label on node selected by scheduler>.
-	// If true, NodeIdLabel must be non-empty.
-	SetNodeIdSelector bool
-	// Label used with SetNodeIdSelector. Must be non-empty if SetNodeIdSelector is true.
-	NodeIdLabel string `validate:"required"`
-	// If true, the Armada scheduler will set the node name of the selected node directly on scheduled pods,
-	// thus bypassing kube-scheduler entirely.
-	SetNodeName bool
-	// Map from priority class names to priority classes.
-	// Must be consistent with Kubernetes priority classes.
-	// I.e., priority classes defined here must be defined in all executor clusters and should map to the same priority.
-	PriorityClasses map[string]types.PriorityClass `validate:"dive"`
-	// Priority class assigned to pods that do not specify one.
-	// Must be an entry in PriorityClasses above.
-	DefaultPriorityClass string
-	// If set, override the priority class name of pods with this value when sending to an executor.
-	PriorityClassNameOverride *string
-}
-
 // FailureEstimatorConfig contains config controlling node and queue success probability estimation.
 // See the internal/scheduler/failureestimator package for details.
 type FailureEstimatorConfig struct {
diff --git a/internal/armada/server.go b/internal/armada/server.go
index 8b49ca20..f4a80b43 100644
--- a/internal/armada/server.go
+++ b/internal/armada/server.go
@@ -39,8 +39,6 @@ import (
 
 func Serve(ctx *armadacontext.Context, config *configuration.ArmadaConfig, healthChecks *health.MultiChecker) error {
 	log.Info("Armada server starting")
-	log.Infof("Armada priority classes: %v", config.Scheduling.Preemption.PriorityClasses)
-	log.Infof("Default priority class: %s", config.Scheduling.Preemption.DefaultPriorityClass)
 	defer log.Info("Armada server shutting down")
 
 	// We call startupCompleteCheck.MarkComplete() when all services have been started.
@@ -58,13 +56,10 @@ func Serve(ctx *armadacontext.Context, config *configuration.ArmadaConfig, healt
 	// we add all services to a slice and start them together at the end of this function.
 	var services []func() error
 
-	err := validateCancelJobsBatchSizeConfig(config)
-	if err != nil {
+	if err := validateCancelJobsBatchSizeConfig(config); err != nil {
 		return err
 	}
-
-	err = validatePreemptionConfig(config.Scheduling.Preemption)
-	if err != nil {
+	if err := validateSubmissionConfig(config.Submission); err != nil {
 		return err
 	}
 
@@ -180,7 +175,7 @@ func Serve(ctx *armadacontext.Context, config *configuration.ArmadaConfig, healt
 		Producer:              producer,
 		QueueRepository:       queueRepository,
 		JobRepository:         jobRepository,
-		SchedulingConfig:      config.Scheduling,
+		SubmissionConfig:      config.Submission,
 		MaxAllowedMessageSize: config.Pulsar.MaxAllowedMessageSize,
 		GangIdAnnotation:      configuration.GangIdAnnotation,
 		SubmitChecker:         pulsarSchedulerSubmitChecker,
@@ -308,15 +303,16 @@ func validateCancelJobsBatchSizeConfig(config *configuration.ArmadaConfig) error
 	return nil
 }
 
-func validatePreemptionConfig(config configuration.PreemptionConfig) error {
-	// Check that the default priority class is in the priority class map.
-	if config.DefaultPriorityClass != "" {
-		_, ok := config.PriorityClasses[config.DefaultPriorityClass]
-		if !ok {
-			return errors.WithStack(fmt.Errorf("default priority class was set to %s, but no such priority class has been configured", config.DefaultPriorityClass))
+func validateSubmissionConfig(config configuration.SubmissionConfig) error {
+	// Check that the default priority class is allowed to be submitted.
+	if config.DefaultPriorityClassName != "" {
+		if !config.AllowedPriorityClassNames[config.DefaultPriorityClassName] {
+			return errors.WithStack(fmt.Errorf(
+				"defaultPriorityClassName %s is not allowed; allowedPriorityClassNames is %v",
+				config.DefaultPriorityClassName, config.AllowedPriorityClassNames,
+			))
 		}
 	}
-
 	return nil
 }
 
diff --git a/internal/armada/server/applydefaults.go b/internal/armada/server/applydefaults.go
index 510110c7..b54b20e9 100644
--- a/internal/armada/server/applydefaults.go
+++ b/internal/armada/server/applydefaults.go
@@ -10,14 +10,14 @@ import (
 	armadaresource "github.com/armadaproject/armada/internal/common/resource"
 )
 
-func applyDefaultsToAnnotations(annotations map[string]string, config configuration.SchedulingConfig) {
+func applyDefaultsToAnnotations(annotations map[string]string, config configuration.SubmissionConfig) {
 	if annotations == nil {
 		return
 	}
 	applyDefaultNodeUniformityLabelAnnotation(annotations, config)
 }
 
-func applyDefaultNodeUniformityLabelAnnotation(annotations map[string]string, config configuration.SchedulingConfig) {
+func applyDefaultNodeUniformityLabelAnnotation(annotations map[string]string, config configuration.SubmissionConfig) {
 	if _, ok := annotations[configuration.GangIdAnnotation]; ok {
 		if _, ok := annotations[configuration.GangNodeUniformityLabelAnnotation]; !ok {
 			annotations[configuration.GangNodeUniformityLabelAnnotation] = config.DefaultGangNodeUniformityLabel
@@ -25,7 +25,7 @@ func applyDefaultNodeUniformityLabelAnnotation(annotations map[string]string, co
 	}
 }
 
-func applyDefaultsToPodSpec(spec *v1.PodSpec, config configuration.SchedulingConfig) {
+func applyDefaultsToPodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) {
 	if spec == nil {
 		return
 	}
@@ -36,7 +36,7 @@ func applyDefaultsToPodSpec(spec *v1.PodSpec, config configuration.SchedulingCon
 	applyDefaultTerminationGracePeriodToPodSpec(spec, config)
 }
 
-func applyDefaultRequestsAndLimitsToPodSpec(spec *v1.PodSpec, config configuration.SchedulingConfig) {
+func applyDefaultRequestsAndLimitsToPodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) {
 	for i := range spec.Containers {
 		c := &spec.Containers[i]
 		if c.Resources.Limits == nil {
@@ -56,7 +56,7 @@ func applyDefaultRequestsAndLimitsToPodSpec(spec *v1.PodSpec, config configurati
 	}
 }
 
-func applyDefaultTolerationsToPodSpec(spec *v1.PodSpec, config configuration.SchedulingConfig) {
+func applyDefaultTolerationsToPodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) {
 	spec.Tolerations = append(spec.Tolerations, config.DefaultJobTolerations...)
 	if config.DefaultJobTolerationsByPriorityClass != nil {
 		if tolerations, ok := config.DefaultJobTolerationsByPriorityClass[spec.PriorityClassName]; ok {
@@ -77,9 +77,9 @@ func applyDefaultTolerationsToPodSpec(spec *v1.PodSpec, config configuration.Sch
 	}
 }
 
-func applyDefaultPriorityClassNameToPodSpec(spec *v1.PodSpec, config configuration.SchedulingConfig) {
+func applyDefaultPriorityClassNameToPodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) {
 	if spec.PriorityClassName == "" {
-		spec.PriorityClassName = config.Preemption.DefaultPriorityClass
+		spec.PriorityClassName = config.DefaultPriorityClassName
 	}
 }
 
@@ -87,7 +87,7 @@ func applyDefaultPriorityClassNameToPodSpec(spec *v1.PodSpec, config configurati
 // of the pod equal to the minimum if
 // - the pod does not explicitly set a termination period, or
 // - the pod explicitly sets a termination period of 0.
-func applyDefaultTerminationGracePeriodToPodSpec(spec *v1.PodSpec, config configuration.SchedulingConfig) {
+func applyDefaultTerminationGracePeriodToPodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) {
 	if config.MinTerminationGracePeriod.Seconds() == 0 {
 		return
 	}
@@ -103,7 +103,7 @@ func applyDefaultTerminationGracePeriodToPodSpec(spec *v1.PodSpec, config config
 	}
 }
 
-func applyDefaultActiveDeadlineSecondsToPodSpec(spec *v1.PodSpec, config configuration.SchedulingConfig) {
+func applyDefaultActiveDeadlineSecondsToPodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) {
 	if spec.ActiveDeadlineSeconds != nil {
 		return
 	}
diff --git a/internal/armada/server/submit_to_log.go b/internal/armada/server/submit_to_log.go
index 75642992..882fee14 100644
--- a/internal/armada/server/submit_to_log.go
+++ b/internal/armada/server/submit_to_log.go
@@ -44,12 +44,11 @@ import (
 // PulsarSubmitServer is a service that accepts API calls according to the original Armada submit API
 // and publishes messages to Pulsar based on those calls.
 // TODO: Consider returning a list of message ids of the messages generated
-// TODO: Include job set as the message key for each message
 type PulsarSubmitServer struct {
 	Producer         pulsar.Producer
 	QueueRepository  repository.QueueRepository
 	JobRepository    repository.JobRepository
-	SchedulingConfig configuration.SchedulingConfig
+	SubmissionConfig configuration.SubmissionConfig
 	// Maximum size of Pulsar messages
 	MaxAllowedMessageSize uint
 	// Used for job submission deduplication.
@@ -93,7 +92,7 @@ func (srv *PulsarSubmitServer) SubmitJobs(grpcCtx context.Context, req *api.JobS
 
 		return nil, st.Err()
 	}
-	if responseItems, err := commonvalidation.ValidateApiJobs(apiJobs, srv.SchedulingConfig); err != nil {
+	if responseItems, err := commonvalidation.ValidateApiJobs(apiJobs, srv.SubmissionConfig); err != nil {
 		details := &api.JobSubmitResponse{
 			JobResponseItems: responseItems,
 		}
@@ -918,9 +917,9 @@ func (srv *PulsarSubmitServer) createJobsObjects(request *api.JobSubmitRequest,
 			namespace = "default"
 		}
 		fillContainerRequestsAndLimits(podSpec.Containers)
-		applyDefaultsToAnnotations(item.Annotations, srv.SchedulingConfig)
-		applyDefaultsToPodSpec(podSpec, srv.SchedulingConfig)
-		if err := validation.ValidatePodSpec(podSpec, srv.SchedulingConfig); err != nil {
+		applyDefaultsToAnnotations(item.Annotations, srv.SubmissionConfig)
+		applyDefaultsToPodSpec(podSpec, srv.SubmissionConfig)
+		if err := validation.ValidatePodSpec(podSpec, srv.SubmissionConfig); err != nil {
 			response := &api.JobSubmitResponseItem{
 				JobId: jobId,
 				Error: fmt.Sprintf("[createJobs] error validating the %d-th job of job set %s: %v", i, request.JobSetId, err),
diff --git a/internal/armada/validation/job.go b/internal/armada/validation/job.go
index 6c55b499..c0314aa8 100644
--- a/internal/armada/validation/job.go
+++ b/internal/armada/validation/job.go
@@ -77,11 +77,11 @@ func ValidateApiJob(job *api.Job, config configuration.SchedulingConfig) error {
 	if err := ValidateApiJobPodSpecs(job); err != nil {
 		return err
 	}
-	if err := validatePodSpecPriorityClass(job.PodSpec, true, config.Preemption.PriorityClasses); err != nil {
+	if err := validatePodSpecPriorityClass(job.PodSpec, true, config.PriorityClasses); err != nil {
 		return err
 	}
 	for _, podSpec := range job.PodSpecs {
-		if err := validatePodSpecPriorityClass(podSpec, true, config.Preemption.PriorityClasses); err != nil {
+		if err := validatePodSpecPriorityClass(podSpec, true, config.PriorityClasses); err != nil {
 			return err
 		}
 	}
diff --git a/internal/armada/validation/podspec.go b/internal/armada/validation/podspec.go
index c90a28f7..eaa54fef 100644
--- a/internal/armada/validation/podspec.go
+++ b/internal/armada/validation/podspec.go
@@ -10,13 +10,13 @@ import (
 	"github.com/armadaproject/armada/internal/common/types"
 )
 
-func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig configuration.SchedulingConfig) error {
+func ValidatePodSpec(spec *v1.PodSpec, config configuration.SubmissionConfig) error {
 	if spec == nil {
 		return errors.Errorf("empty pod spec")
 	}
 
-	if uint(spec.Size()) > schedulingConfig.MaxPodSpecSizeBytes {
-		return errors.Errorf("pod spec has a size of %v bytes which is greater than the maximum allowed size of %v", spec.Size(), schedulingConfig.MaxPodSpecSizeBytes)
+	if uint(spec.Size()) > config.MaxPodSpecSizeBytes {
+		return errors.Errorf("pod spec has a size of %v bytes which is greater than the maximum allowed size of %v", spec.Size(), config.MaxPodSpecSizeBytes)
 	}
 
 	if len(spec.Containers) == 0 {
@@ -28,7 +28,7 @@ func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig configuration.Scheduling
 		return err
 	}
 
-	err = validateTerminationGracePeriod(spec, schedulingConfig)
+	err = validateTerminationGracePeriod(spec, config)
 	if err != nil {
 		return err
 	}
@@ -40,12 +40,10 @@ func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig configuration.Scheduling
 		if len(container.Resources.Requests) == 0 {
 			return errors.Errorf("container %v has no resource requests specified", container.Name)
 		}
-		err = validateContainerResource(container.Resources.Limits, schedulingConfig.MinJobResources, container.Name, "limit")
-		if err != nil {
+		if err := validateContainerResource(container.Resources.Limits, config.MinJobResources, container.Name, "limit"); err != nil {
 			return err
 		}
-		err = validateContainerResource(container.Resources.Requests, schedulingConfig.MinJobResources, container.Name, "request")
-		if err != nil {
+		if err := validateContainerResource(container.Resources.Requests, config.MinJobResources, container.Name, "request"); err != nil {
 			return err
 		}
 		if !resourceListEquals(container.Resources.Requests, container.Resources.Limits) {
@@ -55,14 +53,14 @@ func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig configuration.Scheduling
 	return validatePorts(spec)
 }
 
-func validateTerminationGracePeriod(spec *v1.PodSpec, config configuration.SchedulingConfig) error {
+func validateTerminationGracePeriod(spec *v1.PodSpec, config configuration.SubmissionConfig) error {
 	specHasTerminationGracePeriod := spec.TerminationGracePeriodSeconds != nil
 	var terminationGracePeriodSeconds int64
 	var exceedsBounds bool
 	if specHasTerminationGracePeriod {
 		terminationGracePeriodSeconds = *spec.TerminationGracePeriodSeconds
-		exceedsBounds = (terminationGracePeriodSeconds < int64(config.MinTerminationGracePeriod.Seconds()) ||
-			terminationGracePeriodSeconds > int64(config.MaxTerminationGracePeriod.Seconds()))
+		exceedsBounds = terminationGracePeriodSeconds < int64(config.MinTerminationGracePeriod.Seconds()) ||
+			terminationGracePeriodSeconds > int64(config.MaxTerminationGracePeriod.Seconds())
 	}
 	if exceedsBounds {
 		return errors.Errorf("terminationGracePeriodSeconds of %v must be in [%d, %d]s, or omitted",
@@ -125,9 +123,7 @@ func validateRequiredNodeAffinity(required *v1.NodeSelector) error {
 	if required == nil {
 		return nil
 	}
-
-	_, err := nodeaffinity.NewNodeSelector(required)
-	if err != nil {
+	if _, err := nodeaffinity.NewNodeSelector(required); err != nil {
 		return errors.Errorf("invalid RequiredDuringSchedulingIgnoredDuringExecution node affinity: %v", err)
 	}
 	return nil
diff --git a/internal/common/validation/job.go b/internal/common/validation/job.go
index 6c55b499..e3b4d11c 100644
--- a/internal/common/validation/job.go
+++ b/internal/common/validation/job.go
@@ -9,7 +9,7 @@ import (
 	"github.com/armadaproject/armada/pkg/api"
 )
 
-func ValidateApiJobs(jobs []*api.Job, config configuration.SchedulingConfig) ([]*api.JobSubmitResponseItem, error) {
+func ValidateApiJobs(jobs []*api.Job, config configuration.SubmissionConfig) ([]*api.JobSubmitResponseItem, error) {
 	if _, err := validateGangs(jobs); err != nil {
 		return nil, err
 	}
@@ -73,15 +73,15 @@ func validateGangs(jobs []*api.Job) (map[string]schedulercontext.GangInfo, error
 	return gangDetailsByGangId, nil
 }
 
-func ValidateApiJob(job *api.Job, config configuration.SchedulingConfig) error {
+func ValidateApiJob(job *api.Job, config configuration.SubmissionConfig) error {
 	if err := ValidateApiJobPodSpecs(job); err != nil {
 		return err
 	}
-	if err := validatePodSpecPriorityClass(job.PodSpec, true, config.Preemption.PriorityClasses); err != nil {
+	if err := validatePodSpecPriorityClass(job.PodSpec, config.AllowedPriorityClassNames); err != nil {
 		return err
 	}
 	for _, podSpec := range job.PodSpecs {
-		if err := validatePodSpecPriorityClass(podSpec, true, config.Preemption.PriorityClasses); err != nil {
+		if err := validatePodSpecPriorityClass(podSpec, config.AllowedPriorityClassNames); err != nil {
 			return err
 		}
 	}
diff --git a/internal/common/validation/podspec.go b/internal/common/validation/podspec.go
index d1c556a1..b06ca421 100644
--- a/internal/common/validation/podspec.go
+++ b/internal/common/validation/podspec.go
@@ -1,35 +1,34 @@
 package validation
 
 import (
+	"fmt"
+
 	"github.com/pkg/errors"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/component-helpers/scheduling/corev1/nodeaffinity"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
 	"github.com/armadaproject/armada/internal/common/armadaerrors"
-	"github.com/armadaproject/armada/internal/common/types"
 )
 
-func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig *configuration.SchedulingConfig) error {
+func ValidatePodSpec(spec *v1.PodSpec, config *configuration.SubmissionConfig) error {
 	if spec == nil {
 		return errors.Errorf("empty pod spec")
 	}
 
-	if uint(spec.Size()) > schedulingConfig.MaxPodSpecSizeBytes {
-		return errors.Errorf("pod spec has a size of %v bytes which is greater than the maximum allowed size of %v", spec.Size(), schedulingConfig.MaxPodSpecSizeBytes)
+	if uint(spec.Size()) > config.MaxPodSpecSizeBytes {
+		return errors.Errorf("pod spec has a size of %v bytes which is greater than the maximum allowed size of %v", spec.Size(), config.MaxPodSpecSizeBytes)
 	}
 
 	if len(spec.Containers) == 0 {
 		return errors.Errorf("pod spec has no containers")
 	}
 
-	err := validateAffinity(spec.Affinity)
-	if err != nil {
+	if err := validateAffinity(spec.Affinity); err != nil {
 		return err
 	}
 
-	err = validateTerminationGracePeriod(spec, schedulingConfig)
-	if err != nil {
+	if err := validateTerminationGracePeriod(spec, config); err != nil {
 		return err
 	}
 
@@ -40,12 +39,10 @@ func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig *configuration.Schedulin
 		if len(container.Resources.Requests) == 0 {
 			return errors.Errorf("container %v has no resource requests specified", container.Name)
 		}
-		err = validateContainerResource(container.Resources.Limits, schedulingConfig.MinJobResources, container.Name, "limit")
-		if err != nil {
+		if err := validateContainerResource(container.Resources.Limits, config.MinJobResources, container.Name, "limit"); err != nil {
 			return err
 		}
-		err = validateContainerResource(container.Resources.Requests, schedulingConfig.MinJobResources, container.Name, "request")
-		if err != nil {
+		if err := validateContainerResource(container.Resources.Requests, config.MinJobResources, container.Name, "request"); err != nil {
 			return err
 		}
 		if !resourceListEquals(container.Resources.Requests, container.Resources.Limits) {
@@ -55,7 +52,7 @@ func ValidatePodSpec(spec *v1.PodSpec, schedulingConfig *configuration.Schedulin
 	return validatePorts(spec)
 }
 
-func validateTerminationGracePeriod(spec *v1.PodSpec, config *configuration.SchedulingConfig) error {
+func validateTerminationGracePeriod(spec *v1.PodSpec, config *configuration.SubmissionConfig) error {
 	specHasTerminationGracePeriod := spec.TerminationGracePeriodSeconds != nil
 	var terminationGracePeriodSeconds int64
 	var exceedsBounds bool
@@ -161,24 +158,14 @@ func validatePorts(podSpec *v1.PodSpec) error {
 	return nil
 }
 
-func validatePodSpecPriorityClass(podSpec *v1.PodSpec, preemptionEnabled bool, allowedPriorityClasses map[string]types.PriorityClass) error {
+func validatePodSpecPriorityClass(podSpec *v1.PodSpec, allowedPriorityClassNames map[string]bool) error {
 	priorityClassName := podSpec.PriorityClassName
-	if priorityClassName != "" {
-		if !preemptionEnabled {
-			return errors.WithStack(&armadaerrors.ErrInvalidArgument{
-				Name:    "PriorityClassName",
-				Value:   podSpec.PriorityClassName,
-				Message: "Preemption is disabled in Server config",
-			})
-		}
-		if _, exists := allowedPriorityClasses[priorityClassName]; !exists {
-			return errors.WithStack(&armadaerrors.ErrInvalidArgument{
-				Name:    "PriorityClassName",
-				Value:   podSpec.PriorityClassName,
-				Message: "Specified Priority Class is not supported in Server config",
-			})
-		}
+	if !allowedPriorityClassNames[priorityClassName] {
+		return errors.WithStack(&armadaerrors.ErrInvalidArgument{
+			Name:    "PriorityClassName",
+			Value:   podSpec.PriorityClassName,
+			Message: fmt.Sprintf("priorityClassName not in %v", allowedPriorityClassNames),
+		})
 	}
-
 	return nil
 }
diff --git a/internal/scheduler/constraints/constraints.go b/internal/scheduler/constraints/constraints.go
index d8f6cb6a..816b7425 100644
--- a/internal/scheduler/constraints/constraints.go
+++ b/internal/scheduler/constraints/constraints.go
@@ -92,8 +92,8 @@ func NewSchedulingConstraints(
 	config configuration.SchedulingConfig,
 	queues []queue.Queue,
 ) SchedulingConstraints {
-	priorityClassSchedulingConstraintsByPriorityClassName := make(map[string]PriorityClassSchedulingConstraints, len(config.Preemption.PriorityClasses))
-	for name, priorityClass := range config.Preemption.PriorityClasses {
+	priorityClassSchedulingConstraintsByPriorityClassName := make(map[string]PriorityClassSchedulingConstraints, len(config.PriorityClasses))
+	for name, priorityClass := range config.PriorityClasses {
 		maximumResourceFractionPerQueue := priorityClass.MaximumResourceFractionPerQueue
 		if m, ok := priorityClass.MaximumResourceFractionPerQueueByPool[pool]; ok {
 			// Use pool-specific config is available.
diff --git a/internal/scheduler/pool_assigner.go b/internal/scheduler/pool_assigner.go
index 8e8cf943..83e6f60a 100644
--- a/internal/scheduler/pool_assigner.go
+++ b/internal/scheduler/pool_assigner.go
@@ -58,10 +58,10 @@ func NewPoolAssigner(executorTimeout time.Duration,
 	}
 	return &DefaultPoolAssigner{
 		executorTimeout:        executorTimeout,
-		priorityClasses:        schedulingConfig.Preemption.PriorityClasses,
+		priorityClasses:        schedulingConfig.PriorityClasses,
 		executorsByPool:        map[string][]*executor{},
 		poolByExecutorId:       map[string]string{},
-		priorities:             types.AllowedPriorities(schedulingConfig.Preemption.PriorityClasses),
+		priorities:             types.AllowedPriorities(schedulingConfig.PriorityClasses),
 		indexedResources:       schedulingConfig.IndexedResources,
 		indexedTaints:          schedulingConfig.IndexedTaints,
 		wellKnownNodeTypes:     schedulingConfig.WellKnownNodeTypes,
diff --git a/internal/scheduler/queue_scheduler.go b/internal/scheduler/queue_scheduler.go
index ed3a85c6..85524c02 100644
--- a/internal/scheduler/queue_scheduler.go
+++ b/internal/scheduler/queue_scheduler.go
@@ -97,7 +97,7 @@ func (sch *QueueScheduler) Schedule(ctx *armadacontext.Context) (*SchedulerResul
 				if pctx := jctx.PodSchedulingContext; pctx.IsSuccessful() {
 					scheduledJobs = append(scheduledJobs, jctx)
 					nodeIdByJobId[jctx.JobId] = pctx.NodeId
-					additionalAnnotationsByJobId[jctx.JobId] = map[string]string{configuration.RuntimeGangCardinality: strconv.Itoa(numScheduled)}
+					additionalAnnotationsByJobId[jctx.JobId] = map[string]string{configuration.GangNumJobsScheduledAnnotation: strconv.Itoa(numScheduled)}
 				} else if jctx.ShouldFail {
 					failedJobs = append(failedJobs, jctx)
 				}
diff --git a/internal/scheduler/schedulerapp.go b/internal/scheduler/schedulerapp.go
index 9d7f6f35..af695e7c 100644
--- a/internal/scheduler/schedulerapp.go
+++ b/internal/scheduler/schedulerapp.go
@@ -154,9 +154,9 @@ func Run(config schedulerconfig.Configuration) error {
 		jobRepository,
 		executorRepository,
 		legacyExecutorRepository,
-		types.AllowedPriorities(config.Scheduling.Preemption.PriorityClasses),
-		config.Scheduling.Preemption.NodeIdLabel,
-		config.Scheduling.Preemption.PriorityClassNameOverride,
+		types.AllowedPriorities(config.Scheduling.PriorityClasses),
+		config.Scheduling.NodeIdLabel,
+		config.Scheduling.PriorityClassNameOverride,
 		config.Pulsar.MaxAllowedMessageSize,
 	)
 	if err != nil {
@@ -206,8 +206,8 @@ func Run(config schedulerconfig.Configuration) error {
 		return errors.WithMessage(err, "error creating scheduling algo")
 	}
 	jobDb := jobdb.NewJobDb(
-		config.Scheduling.Preemption.PriorityClasses,
-		config.Scheduling.Preemption.DefaultPriorityClass,
+		config.Scheduling.PriorityClasses,
+		config.Scheduling.DefaultPriorityClassName,
 		config.InternedStringsCacheSize,
 	)
 	schedulingRoundMetrics := NewSchedulerMetrics(config.Metrics.Metrics)
@@ -252,7 +252,7 @@ func Run(config schedulerconfig.Configuration) error {
 		config.SchedulePeriod,
 		config.ExecutorTimeout,
 		config.Scheduling.MaxRetries+1,
-		config.Scheduling.Preemption.NodeIdLabel,
+		config.Scheduling.NodeIdLabel,
 		schedulingRoundMetrics,
 		schedulerMetrics,
 		failureEstimator,
diff --git a/internal/scheduler/scheduling_algo.go b/internal/scheduler/scheduling_algo.go
index cc51ab69..3a5d1893 100644
--- a/internal/scheduler/scheduling_algo.go
+++ b/internal/scheduler/scheduling_algo.go
@@ -68,8 +68,11 @@ func NewFairSchedulingAlgo(
 	queueRepository repository.QueueRepository,
 	schedulingContextRepository *SchedulingContextRepository,
 ) (*FairSchedulingAlgo, error) {
-	if _, ok := config.Preemption.PriorityClasses[config.Preemption.DefaultPriorityClass]; !ok {
-		return nil, errors.Errorf("default priority class %s is missing from priority class mapping %v", config.Preemption.DefaultPriorityClass, config.Preemption.PriorityClasses)
+	if _, ok := config.PriorityClasses[config.DefaultPriorityClassName]; !ok {
+		return nil, errors.Errorf(
+			"defaultPriorityClassName %s does not correspond to a priority class; priorityClasses is %v",
+			config.DefaultPriorityClassName, config.PriorityClasses,
+		)
 	}
 	return &FairSchedulingAlgo{
 		schedulingConfig:            config,
@@ -334,7 +337,7 @@ func (l *FairSchedulingAlgo) scheduleOnExecutors(
 	executors []*schedulerobjects.Executor,
 ) (*SchedulerResult, *schedulercontext.SchedulingContext, error) {
 	nodeDb, err := nodedb.NewNodeDb(
-		l.schedulingConfig.Preemption.PriorityClasses,
+		l.schedulingConfig.PriorityClasses,
 		l.schedulingConfig.MaxExtraNodesToConsider,
 		l.schedulingConfig.IndexedResources,
 		l.schedulingConfig.IndexedTaints,
@@ -371,8 +374,8 @@ func (l *FairSchedulingAlgo) scheduleOnExecutors(
 	sctx := schedulercontext.NewSchedulingContext(
 		executorId,
 		pool,
-		l.schedulingConfig.Preemption.PriorityClasses,
-		l.schedulingConfig.Preemption.DefaultPriorityClass,
+		l.schedulingConfig.PriorityClasses,
+		l.schedulingConfig.DefaultPriorityClassName,
 		fairnessCostProvider,
 		l.limiter,
 		totalResources,
@@ -413,9 +416,9 @@ func (l *FairSchedulingAlgo) scheduleOnExecutors(
 	scheduler := NewPreemptingQueueScheduler(
 		sctx,
 		constraints,
-		l.schedulingConfig.Preemption.NodeEvictionProbability,
-		l.schedulingConfig.Preemption.NodeOversubscriptionEvictionProbability,
-		l.schedulingConfig.Preemption.ProtectedFractionOfFairShare,
+		l.schedulingConfig.NodeEvictionProbability,
+		l.schedulingConfig.NodeOversubscriptionEvictionProbability,
+		l.schedulingConfig.ProtectedFractionOfFairShare,
 		NewSchedulerJobRepositoryAdapter(fsctx.txn),
 		nodeDb,
 		fsctx.nodeIdByJobId,
diff --git a/internal/scheduler/simulator/simulator.go b/internal/scheduler/simulator/simulator.go
index 3973e827..2e8d5dbe 100644
--- a/internal/scheduler/simulator/simulator.go
+++ b/internal/scheduler/simulator/simulator.go
@@ -99,8 +99,8 @@ func NewSimulator(clusterSpec *ClusterSpec, workloadSpec *WorkloadSpec, scheduli
 		return nil, err
 	}
 	jobDb := jobdb.NewJobDb(
-		schedulingConfig.Preemption.PriorityClasses,
-		schedulingConfig.Preemption.DefaultPriorityClass,
+		schedulingConfig.PriorityClasses,
+		schedulingConfig.DefaultPriorityClassName,
 		1024,
 	)
 	randomSeed := workloadSpec.RandomSeed
@@ -223,7 +223,7 @@ func (s *Simulator) setupClusters() error {
 		totalResourcesForPool := schedulerobjects.ResourceList{}
 		for executorGroupIndex, executorGroup := range pool.ClusterGroups {
 			nodeDb, err := nodedb.NewNodeDb(
-				s.schedulingConfig.Preemption.PriorityClasses,
+				s.schedulingConfig.PriorityClasses,
 				s.schedulingConfig.MaxExtraNodesToConsider,
 				s.schedulingConfig.IndexedResources,
 				s.schedulingConfig.IndexedTaints,
@@ -247,7 +247,7 @@ func (s *Simulator) setupClusters() error {
 							Labels:         maps.Clone(nodeTemplate.Labels),
 							TotalResources: nodeTemplate.TotalResources.DeepCopy(),
 							AllocatableByPriorityAndResource: schedulerobjects.NewAllocatableByPriorityAndResourceType(
-								types.AllowedPriorities(s.schedulingConfig.Preemption.PriorityClasses),
+								types.AllowedPriorities(s.schedulingConfig.PriorityClasses),
 								nodeTemplate.TotalResources,
 							),
 						}
@@ -433,8 +433,8 @@ func (s *Simulator) handleScheduleEvent(ctx *armadacontext.Context) error {
 			sctx := schedulercontext.NewSchedulingContext(
 				fmt.Sprintf("%s-%d", pool.Name, i),
 				pool.Name,
-				s.schedulingConfig.Preemption.PriorityClasses,
-				s.schedulingConfig.Preemption.DefaultPriorityClass,
+				s.schedulingConfig.PriorityClasses,
+				s.schedulingConfig.DefaultPriorityClassName,
 				fairnessCostProvider,
 				s.limiter,
 				totalResources,
@@ -472,9 +472,9 @@ func (s *Simulator) handleScheduleEvent(ctx *armadacontext.Context) error {
 			sch := scheduler.NewPreemptingQueueScheduler(
 				sctx,
 				constraints,
-				s.schedulingConfig.Preemption.NodeEvictionProbability,
-				s.schedulingConfig.Preemption.NodeOversubscriptionEvictionProbability,
-				s.schedulingConfig.Preemption.ProtectedFractionOfFairShare,
+				s.schedulingConfig.NodeEvictionProbability,
+				s.schedulingConfig.NodeOversubscriptionEvictionProbability,
+				s.schedulingConfig.ProtectedFractionOfFairShare,
 				scheduler.NewSchedulerJobRepositoryAdapter(txn),
 				nodeDb,
 				// TODO: Necessary to support partial eviction.
@@ -661,7 +661,7 @@ func (s *Simulator) handleEventSequence(ctx *armadacontext.Context, es *armadaev
 }
 
 func (s *Simulator) handleSubmitJob(txn *jobdb.Txn, e *armadaevents.SubmitJob, time time.Time, eventSequence *armadaevents.EventSequence) (*jobdb.Job, bool, error) {
-	schedulingInfo, err := scheduleringester.SchedulingInfoFromSubmitJob(e, time, s.schedulingConfig.Preemption.PriorityClasses)
+	schedulingInfo, err := scheduleringester.SchedulingInfoFromSubmitJob(e, time, s.schedulingConfig.PriorityClasses)
 	if err != nil {
 		return nil, false, err
 	}
@@ -815,7 +815,7 @@ func (s *Simulator) unbindRunningJob(job *jobdb.Job) error {
 	} else if node == nil {
 		return errors.Errorf("node %s not found", run.NodeId())
 	}
-	node, err = nodeDb.UnbindJobFromNode(s.schedulingConfig.Preemption.PriorityClasses, job, node)
+	node, err = nodeDb.UnbindJobFromNode(s.schedulingConfig.PriorityClasses, job, node)
 	if err != nil {
 		return err
 	}
diff --git a/internal/scheduler/submitcheck.go b/internal/scheduler/submitcheck.go
index 67183d07..9b7e2bd5 100644
--- a/internal/scheduler/submitcheck.go
+++ b/internal/scheduler/submitcheck.go
@@ -45,7 +45,6 @@ type SubmitScheduleChecker interface {
 type SubmitChecker struct {
 	executorTimeout           time.Duration
 	priorityClasses           map[string]types.PriorityClass
-	gangIdAnnotation          string
 	executorById              map[string]minimalExecutor
 	priorities                []int32
 	indexedResources          []configuration.IndexedResource
@@ -71,10 +70,9 @@ func NewSubmitChecker(
 	}
 	return &SubmitChecker{
 		executorTimeout:           executorTimeout,
-		priorityClasses:           schedulingConfig.Preemption.PriorityClasses,
-		gangIdAnnotation:          configuration.GangIdAnnotation,
+		priorityClasses:           schedulingConfig.PriorityClasses,
 		executorById:              map[string]minimalExecutor{},
-		priorities:                types.AllowedPriorities(schedulingConfig.Preemption.PriorityClasses),
+		priorities:                types.AllowedPriorities(schedulingConfig.PriorityClasses),
 		indexedResources:          schedulingConfig.IndexedResources,
 		indexedTaints:             schedulingConfig.IndexedTaints,
 		indexedNodeLabels:         schedulingConfig.IndexedNodeLabels,


Test patch:
diff --git a/internal/armada/server/applydefaults_test.go b/internal/armada/server/applydefaults_test.go
index 62293a45..bf4e1995 100644
--- a/internal/armada/server/applydefaults_test.go
+++ b/internal/armada/server/applydefaults_test.go
@@ -13,7 +13,7 @@ import (
 
 func TestApplyDefaultsToAnnotations(t *testing.T) {
 	tests := map[string]struct {
-		Config      configuration.SchedulingConfig
+		Config      configuration.SubmissionConfig
 		Annotations map[string]string
 		Expected    map[string]string
 	}{
@@ -22,7 +22,7 @@ func TestApplyDefaultsToAnnotations(t *testing.T) {
 			Expected:    make(map[string]string),
 		},
 		"DefaultNodeUniformityLabelAnnotation no change for non-gang jobs": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultGangNodeUniformityLabel: "foo",
 			},
 			Annotations: make(map[string]string),
@@ -38,7 +38,7 @@ func TestApplyDefaultsToAnnotations(t *testing.T) {
 			},
 		},
 		"DefaultNodeUniformityLabelAnnotation": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultGangNodeUniformityLabel: "foo",
 			},
 			Annotations: map[string]string{
@@ -60,15 +60,13 @@ func TestApplyDefaultsToAnnotations(t *testing.T) {
 
 func TestApplyDefaultsToPodSpec(t *testing.T) {
 	tests := map[string]struct {
-		Config   configuration.SchedulingConfig
+		Config   configuration.SubmissionConfig
 		PodSpec  v1.PodSpec
 		Expected v1.PodSpec
 	}{
 		"DefaultPriorityClassName": {
-			Config: configuration.SchedulingConfig{
-				Preemption: configuration.PreemptionConfig{
-					DefaultPriorityClass: "pc",
-				},
+			Config: configuration.SubmissionConfig{
+				DefaultPriorityClassName: "pc",
 			},
 			PodSpec: v1.PodSpec{},
 			Expected: v1.PodSpec{
@@ -76,7 +74,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultJobLimits": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultJobLimits: map[string]resource.Quantity{
 					"cpu":    resource.MustParse("10"),
 					"memory": resource.MustParse("1Gi"),
@@ -115,7 +113,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultJobTolerations": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultJobTolerations: []v1.Toleration{{Key: "foo"}, {Key: "bar"}},
 			},
 			PodSpec: v1.PodSpec{
@@ -126,7 +124,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultJobTolerationsByPriorityClass": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultJobTolerationsByPriorityClass: map[string][]v1.Toleration{
 					"pc-1": {{Key: "foo"}, {Key: "bar"}},
 					"pc-2": {{Key: "oof"}, {Key: "rab"}},
@@ -142,7 +140,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultJobTolerationsByResourceRequest": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultJobTolerationsByResourceRequest: map[string][]v1.Toleration{
 					"gpu": {{Key: "foo"}, {Key: "bar"}},
 				},
@@ -177,7 +175,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultJobTolerationsByResourceRequest explicit zero resource": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultJobTolerationsByResourceRequest: map[string][]v1.Toleration{
 					"gpu": {{Key: "foo"}, {Key: "bar"}},
 				},
@@ -212,10 +210,8 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultPriorityClassName + DefaultJobTolerationsByPriorityClass": {
-			Config: configuration.SchedulingConfig{
-				Preemption: configuration.PreemptionConfig{
-					DefaultPriorityClass: "pc",
-				},
+			Config: configuration.SubmissionConfig{
+				DefaultPriorityClassName: "pc",
 				DefaultJobTolerationsByPriorityClass: map[string][]v1.Toleration{
 					"pc": {{Key: "foo"}, {Key: "bar"}},
 				},
@@ -227,7 +223,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultActiveDeadlineSeconds": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultActiveDeadline: time.Second,
 			},
 			Expected: v1.PodSpec{
@@ -235,7 +231,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultActiveDeadlineSecondsByResource": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultActiveDeadlineByResourceRequest: map[string]time.Duration{
 					"memory": 2 * time.Minute,
 					"gpu":    time.Minute,
@@ -272,7 +268,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultActiveDeadlineSeconds + DefaultActiveDeadlineSecondsByResource": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultActiveDeadline: time.Second,
 				DefaultActiveDeadlineByResourceRequest: map[string]time.Duration{
 					"gpu": time.Minute,
@@ -307,7 +303,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultActiveDeadlineSecondsByResource trumps DefaultActiveDeadlineSeconds": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultActiveDeadline: time.Minute,
 				DefaultActiveDeadlineByResourceRequest: map[string]time.Duration{
 					"gpu": time.Second,
@@ -340,7 +336,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"DefaultActiveDeadlineSecondsByResource explicit zero resource": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				DefaultActiveDeadlineByResourceRequest: map[string]time.Duration{
 					"gpu": time.Second,
 				},
@@ -371,7 +367,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"MinTerminationGracePeriod": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				MinTerminationGracePeriod: time.Second,
 			},
 			Expected: v1.PodSpec{
@@ -379,7 +375,7 @@ func TestApplyDefaultsToPodSpec(t *testing.T) {
 			},
 		},
 		"MinTerminationGracePeriod convert 0 to 1": {
-			Config: configuration.SchedulingConfig{
+			Config: configuration.SubmissionConfig{
 				MinTerminationGracePeriod: time.Second,
 			},
 			PodSpec: v1.PodSpec{
diff --git a/internal/armada/validation/podspec_test.go b/internal/armada/validation/podspec_test.go
index 27c9ab91..ee7f25f9 100644
--- a/internal/armada/validation/podspec_test.go
+++ b/internal/armada/validation/podspec_test.go
@@ -14,16 +14,16 @@ import (
 )
 
 func Test_ValidatePodSpec_checkForMissingValues(t *testing.T) {
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
 
-	assert.Error(t, ValidatePodSpec(nil, schedulingConfig))
-	assert.Error(t, ValidatePodSpec(&v1.PodSpec{}, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(nil, config))
+	assert.Error(t, ValidatePodSpec(&v1.PodSpec{}, config))
 	assert.Error(t, ValidatePodSpec(&v1.PodSpec{
 		Containers: []v1.Container{{}},
-	}, schedulingConfig))
+	}, config))
 }
 
 func Test_ValidatePodSpec_checkForResources(t *testing.T) {
@@ -34,7 +34,7 @@ func Test_ValidatePodSpec_checkForResources(t *testing.T) {
 	resources1 := v1.ResourceList{"cpu": cpu, "memory": memory}
 	resources2 := v1.ResourceList{"cpu": cpu2, "memory": memory}
 
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -46,7 +46,7 @@ func Test_ValidatePodSpec_checkForResources(t *testing.T) {
 				Requests: resources2,
 			},
 		}},
-	}, schedulingConfig))
+	}, config))
 
 	assert.NoError(t, ValidatePodSpec(&v1.PodSpec{
 		Containers: []v1.Container{{
@@ -55,15 +55,13 @@ func Test_ValidatePodSpec_checkForResources(t *testing.T) {
 				Requests: resources1,
 			},
 		}},
-	}, schedulingConfig))
+	}, config))
 }
 
 func Test_ValidatePodSpec_terminationGracePeriod(t *testing.T) {
-	schedulingConfig := configuration.SchedulingConfig{
-		Preemption: configuration.PreemptionConfig{
-			DefaultPriorityClass: "high",
-			PriorityClasses:      map[string]types.PriorityClass{"high": {Priority: 0}},
-		},
+	config := configuration.SubmissionConfig{
+		DefaultPriorityClassName:  "high",
+		AllowedPriorityClassNames: map[string]bool{"high": true},
 		MinTerminationGracePeriod: time.Duration(30 * time.Second),
 		MaxTerminationGracePeriod: time.Duration(300 * time.Second),
 	}
@@ -80,13 +78,13 @@ func Test_ValidatePodSpec_terminationGracePeriod(t *testing.T) {
 		PriorityClassName: "high",
 	}
 
-	assert.Error(t, validateTerminationGracePeriod(podspecOutsideRange, schedulingConfig))
-	assert.NoError(t, validateTerminationGracePeriod(podspecWithinRange, schedulingConfig))
-	assert.NoError(t, validateTerminationGracePeriod(podspecNoSetting, schedulingConfig))
+	assert.Error(t, validateTerminationGracePeriod(podspecOutsideRange, config))
+	assert.NoError(t, validateTerminationGracePeriod(podspecWithinRange, config))
+	assert.NoError(t, validateTerminationGracePeriod(podspecNoSetting, config))
 }
 
 func Test_ValidatePodSpec_checkForPortConfiguration(t *testing.T) {
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -120,12 +118,12 @@ func Test_ValidatePodSpec_checkForPortConfiguration(t *testing.T) {
 			},
 		},
 	}
-	assert.Error(t, ValidatePodSpec(portsUniqueToContainer, schedulingConfig))
-	assert.Error(t, ValidatePodSpec(portExposeOverMultipleContainers, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(portsUniqueToContainer, config))
+	assert.Error(t, ValidatePodSpec(portExposeOverMultipleContainers, config))
 }
 
 func Test_ValidatePodSpec_WhenPreferredAffinitySet_Fails(t *testing.T) {
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -152,11 +150,11 @@ func Test_ValidatePodSpec_WhenPreferredAffinitySet_Fails(t *testing.T) {
 		},
 	}
 
-	assert.Error(t, ValidatePodSpec(podSpec, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(podSpec, config))
 }
 
 func Test_ValidatePodSpec_WhenValidRequiredAffinitySet_Succeeds(t *testing.T) {
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -182,11 +180,11 @@ func Test_ValidatePodSpec_WhenValidRequiredAffinitySet_Succeeds(t *testing.T) {
 		},
 	}
 
-	assert.Nil(t, ValidatePodSpec(podSpec, schedulingConfig))
+	assert.Nil(t, ValidatePodSpec(podSpec, config))
 }
 
 func Test_ValidatePodSpec_WhenInvalidRequiredAffinitySet_Fails(t *testing.T) {
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -212,34 +210,34 @@ func Test_ValidatePodSpec_WhenInvalidRequiredAffinitySet_Fails(t *testing.T) {
 		},
 	}
 
-	assert.Error(t, ValidatePodSpec(podSpec, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(podSpec, config))
 }
 
 func Test_ValidatePodSpec_WhenExceedsMaxSize_Fails(t *testing.T) {
 	spec := minimalValidPodSpec()
 	specSize := uint(spec.Size())
 
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: specSize,
 	}
-	assert.NoError(t, ValidatePodSpec(spec, schedulingConfig))
+	assert.NoError(t, ValidatePodSpec(spec, config))
 
-	schedulingConfig.MaxPodSpecSizeBytes -= 1
-	assert.Error(t, ValidatePodSpec(spec, schedulingConfig))
+	config.MaxPodSpecSizeBytes -= 1
+	assert.Error(t, ValidatePodSpec(spec, config))
 }
 
 func Test_ValidatePodSpec_WhenResourcesAboveMinimum_Succeeds(t *testing.T) {
 	spec := minimalValidPodSpec()
 
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MaxPodSpecSizeBytes: uint(spec.Size()),
 		MinJobResources: v1.ResourceList{
 			"memory": resource.MustParse("100Mi"),
 		},
 	}
 
-	assert.NoError(t, ValidatePodSpec(spec, schedulingConfig))
+	assert.NoError(t, ValidatePodSpec(spec, config))
 }
 
 func Test_ValidatePodSpec_WhenResourcesBelowMinimum_Fails(t *testing.T) {
@@ -258,14 +256,14 @@ func Test_ValidatePodSpec_WhenResourcesBelowMinimum_Fails(t *testing.T) {
 		},
 	}
 
-	schedulingConfig := configuration.SchedulingConfig{
+	config := configuration.SubmissionConfig{
 		MaxPodSpecSizeBytes: uint(spec.Size()),
 		MinJobResources: v1.ResourceList{
 			"memory": resource.MustParse("100Mi"),
 		},
 	}
 
-	assert.Error(t, ValidatePodSpec(spec, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(spec, config))
 }
 
 func minimalValidPodSpec() *v1.PodSpec {
diff --git a/internal/common/validation/podspec_test.go b/internal/common/validation/podspec_test.go
index bffa1f27..a569fb7a 100644
--- a/internal/common/validation/podspec_test.go
+++ b/internal/common/validation/podspec_test.go
@@ -10,20 +10,19 @@ import (
 	"k8s.io/utils/pointer"
 
 	"github.com/armadaproject/armada/internal/armada/configuration"
-	"github.com/armadaproject/armada/internal/common/types"
 )
 
 func Test_ValidatePodSpec_checkForMissingValues(t *testing.T) {
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
 
-	assert.Error(t, ValidatePodSpec(nil, schedulingConfig))
-	assert.Error(t, ValidatePodSpec(&v1.PodSpec{}, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(nil, config))
+	assert.Error(t, ValidatePodSpec(&v1.PodSpec{}, config))
 	assert.Error(t, ValidatePodSpec(&v1.PodSpec{
 		Containers: []v1.Container{{}},
-	}, schedulingConfig))
+	}, config))
 }
 
 func Test_ValidatePodSpec_checkForResources(t *testing.T) {
@@ -34,7 +33,7 @@ func Test_ValidatePodSpec_checkForResources(t *testing.T) {
 	resources1 := v1.ResourceList{"cpu": cpu, "memory": memory}
 	resources2 := v1.ResourceList{"cpu": cpu2, "memory": memory}
 
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -46,7 +45,7 @@ func Test_ValidatePodSpec_checkForResources(t *testing.T) {
 				Requests: resources2,
 			},
 		}},
-	}, schedulingConfig))
+	}, config))
 
 	assert.NoError(t, ValidatePodSpec(&v1.PodSpec{
 		Containers: []v1.Container{{
@@ -55,15 +54,13 @@ func Test_ValidatePodSpec_checkForResources(t *testing.T) {
 				Requests: resources1,
 			},
 		}},
-	}, schedulingConfig))
+	}, config))
 }
 
 func Test_ValidatePodSpec_terminationGracePeriod(t *testing.T) {
-	schedulingConfig := &configuration.SchedulingConfig{
-		Preemption: configuration.PreemptionConfig{
-			DefaultPriorityClass: "high",
-			PriorityClasses:      map[string]types.PriorityClass{"high": {Priority: 0}},
-		},
+	config := &configuration.SubmissionConfig{
+		DefaultPriorityClassName:  "high",
+		AllowedPriorityClassNames: map[string]bool{"high": true},
 		MinTerminationGracePeriod: time.Duration(30 * time.Second),
 		MaxTerminationGracePeriod: time.Duration(300 * time.Second),
 	}
@@ -80,13 +77,13 @@ func Test_ValidatePodSpec_terminationGracePeriod(t *testing.T) {
 		PriorityClassName: "high",
 	}
 
-	assert.Error(t, validateTerminationGracePeriod(podspecOutsideRange, schedulingConfig))
-	assert.NoError(t, validateTerminationGracePeriod(podspecWithinRange, schedulingConfig))
-	assert.NoError(t, validateTerminationGracePeriod(podspecNoSetting, schedulingConfig))
+	assert.Error(t, validateTerminationGracePeriod(podspecOutsideRange, config))
+	assert.NoError(t, validateTerminationGracePeriod(podspecWithinRange, config))
+	assert.NoError(t, validateTerminationGracePeriod(podspecNoSetting, config))
 }
 
 func Test_ValidatePodSpec_checkForPortConfiguration(t *testing.T) {
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -120,12 +117,12 @@ func Test_ValidatePodSpec_checkForPortConfiguration(t *testing.T) {
 			},
 		},
 	}
-	assert.Error(t, ValidatePodSpec(portsUniqueToContainer, schedulingConfig))
-	assert.Error(t, ValidatePodSpec(portExposeOverMultipleContainers, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(portsUniqueToContainer, config))
+	assert.Error(t, ValidatePodSpec(portExposeOverMultipleContainers, config))
 }
 
 func Test_ValidatePodSpec_WhenPreferredAffinitySet_Fails(t *testing.T) {
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -152,11 +149,11 @@ func Test_ValidatePodSpec_WhenPreferredAffinitySet_Fails(t *testing.T) {
 		},
 	}
 
-	assert.Error(t, ValidatePodSpec(podSpec, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(podSpec, config))
 }
 
 func Test_ValidatePodSpec_WhenValidRequiredAffinitySet_Succeeds(t *testing.T) {
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -182,11 +179,11 @@ func Test_ValidatePodSpec_WhenValidRequiredAffinitySet_Succeeds(t *testing.T) {
 		},
 	}
 
-	assert.Nil(t, ValidatePodSpec(podSpec, schedulingConfig))
+	assert.Nil(t, ValidatePodSpec(podSpec, config))
 }
 
 func Test_ValidatePodSpec_WhenInvalidRequiredAffinitySet_Fails(t *testing.T) {
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: 65535,
 	}
@@ -212,34 +209,34 @@ func Test_ValidatePodSpec_WhenInvalidRequiredAffinitySet_Fails(t *testing.T) {
 		},
 	}
 
-	assert.Error(t, ValidatePodSpec(podSpec, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(podSpec, config))
 }
 
 func Test_ValidatePodSpec_WhenExceedsMaxSize_Fails(t *testing.T) {
 	spec := minimalValidPodSpec()
 	specSize := uint(spec.Size())
 
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MinJobResources:     v1.ResourceList{},
 		MaxPodSpecSizeBytes: specSize,
 	}
-	assert.NoError(t, ValidatePodSpec(spec, schedulingConfig))
+	assert.NoError(t, ValidatePodSpec(spec, config))
 
-	schedulingConfig.MaxPodSpecSizeBytes -= 1
-	assert.Error(t, ValidatePodSpec(spec, schedulingConfig))
+	config.MaxPodSpecSizeBytes -= 1
+	assert.Error(t, ValidatePodSpec(spec, config))
 }
 
 func Test_ValidatePodSpec_WhenResourcesAboveMinimum_Succeeds(t *testing.T) {
 	spec := minimalValidPodSpec()
 
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MaxPodSpecSizeBytes: uint(spec.Size()),
 		MinJobResources: v1.ResourceList{
 			"memory": resource.MustParse("100Mi"),
 		},
 	}
 
-	assert.NoError(t, ValidatePodSpec(spec, schedulingConfig))
+	assert.NoError(t, ValidatePodSpec(spec, config))
 }
 
 func Test_ValidatePodSpec_WhenResourcesBelowMinimum_Fails(t *testing.T) {
@@ -258,14 +255,14 @@ func Test_ValidatePodSpec_WhenResourcesBelowMinimum_Fails(t *testing.T) {
 		},
 	}
 
-	schedulingConfig := &configuration.SchedulingConfig{
+	config := &configuration.SubmissionConfig{
 		MaxPodSpecSizeBytes: uint(spec.Size()),
 		MinJobResources: v1.ResourceList{
 			"memory": resource.MustParse("100Mi"),
 		},
 	}
 
-	assert.Error(t, ValidatePodSpec(spec, schedulingConfig))
+	assert.Error(t, ValidatePodSpec(spec, config))
 }
 
 func minimalValidPodSpec() *v1.PodSpec {
@@ -287,27 +284,19 @@ func minimalValidPodSpec() *v1.PodSpec {
 
 func Test_ValidatePodSpecPriorityClass(t *testing.T) {
 	validPriorityClass := &v1.PodSpec{PriorityClassName: "some-priority-class"}
-	allowedPriorityClasses := map[string]types.PriorityClass{"some-priority-class": {Priority: 10}}
+	allowedPriorityClasses := map[string]bool{"some-priority-class": true}
 	assert.NoError(
 		t,
-		validatePodSpecPriorityClass(validPriorityClass, true, allowedPriorityClasses),
+		validatePodSpecPriorityClass(validPriorityClass, allowedPriorityClasses),
 		"validation should pass when specified priority class is configured to be allowed and preemption is enabled",
 	)
 
-	err := validatePodSpecPriorityClass(validPriorityClass, false, allowedPriorityClasses)
-	assert.Error(
-		t,
-		err,
-		"validation should fail if priority class is specified and disabled",
-	)
-	validateInvalidArgumentErrorMessage(t, err, "Preemption is disabled in Server config")
-
 	invalidPriorityClass := &v1.PodSpec{PriorityClassName: "some-other-priority-class"}
-	err = validatePodSpecPriorityClass(invalidPriorityClass, true, allowedPriorityClasses)
+	err := validatePodSpecPriorityClass(invalidPriorityClass, allowedPriorityClasses)
 	assert.Error(
 		t,
 		err,
 		"validation should fail if specified priority class is not configured to be allowed",
 	)
-	validateInvalidArgumentErrorMessage(t, err, "Specified Priority Class is not supported in Server config")
+	validateInvalidArgumentErrorMessage(t, err, "priorityClassName not in map[some-priority-class:true]")
 }
diff --git a/internal/scheduler/configuration/configuration_test.go b/internal/scheduler/configuration/configuration_test.go
index b12fc437..00b17af4 100644
--- a/internal/scheduler/configuration/configuration_test.go
+++ b/internal/scheduler/configuration/configuration_test.go
@@ -23,16 +23,14 @@ func TestSchedulingConfigValidate(t *testing.T) {
 					Taints: []v1.Taint{{Key: "other-key", Value: "true", Effect: v1.TaintEffectNoSchedule}},
 				},
 			},
-			Preemption: configuration.PreemptionConfig{
-				PriorityClasses: map[string]types.PriorityClass{
-					"armada-preemptible-away": {
-						Priority: 100,
+			PriorityClasses: map[string]types.PriorityClass{
+				"armada-preemptible-away": {
+					Priority: 100,
 
-						AwayNodeTypes: []types.AwayNodeType{
-							{
-								Priority:              50,
-								WellKnownNodeTypeName: "not-gpu",
-							},
+					AwayNodeTypes: []types.AwayNodeType{
+						{
+							Priority:              50,
+							WellKnownNodeTypeName: "not-gpu",
 						},
 					},
 				},
diff --git a/internal/scheduler/constraints/constraints_test.go b/internal/scheduler/constraints/constraints_test.go
index 630302a5..c2b17cf2 100644
--- a/internal/scheduler/constraints/constraints_test.go
+++ b/internal/scheduler/constraints/constraints_test.go
@@ -53,9 +53,7 @@ func TestConstraints(t *testing.T) {
 			configuration.SchedulingConfig{
 				MaximumResourceFractionToSchedule: map[string]float64{"cpu": 0.1, "memory": 0.1},
 				MaxQueueLookback:                  1000,
-				Preemption: configuration.PreemptionConfig{
-					PriorityClasses: map[string]types.PriorityClass{"priority-class-1": {MaximumResourceFractionPerQueueByPool: map[string]map[string]float64{"pool-1": {"cpu": 0.9, "memory": 0.9}}}},
-				},
+				PriorityClasses:                   map[string]types.PriorityClass{"priority-class-1": {MaximumResourceFractionPerQueueByPool: map[string]map[string]float64{"pool-1": {"cpu": 0.9, "memory": 0.9}}}},
 			},
 			[]queue.Queue{{Name: "queue-1", ResourceLimitsByPriorityClassName: map[string]api.PriorityClassResourceLimits{"priority-class-1": {MaximumResourceFraction: map[string]float64{"cpu": 0.9, "memory": 0.9}}}}},
 		)),
@@ -111,9 +109,7 @@ func TestConstraints(t *testing.T) {
 				configuration.SchedulingConfig{
 					MaximumResourceFractionToSchedule: map[string]float64{"cpu": 0.1, "memory": 0.1},
 					MaxQueueLookback:                  1000,
-					Preemption: configuration.PreemptionConfig{
-						PriorityClasses: map[string]types.PriorityClass{"priority-class-1": {MaximumResourceFractionPerQueueByPool: map[string]map[string]float64{"pool-1": {"cpu": 0.00000001, "memory": 0.9}}}},
-					},
+					PriorityClasses:                   map[string]types.PriorityClass{"priority-class-1": {MaximumResourceFractionPerQueueByPool: map[string]map[string]float64{"pool-1": {"cpu": 0.00000001, "memory": 0.9}}}},
 				},
 				[]queue.Queue{},
 			))
@@ -127,9 +123,7 @@ func TestConstraints(t *testing.T) {
 			configuration.SchedulingConfig{
 				MaximumResourceFractionToSchedule: map[string]float64{"cpu": 0.1, "memory": 0.1},
 				MaxQueueLookback:                  1000,
-				Preemption: configuration.PreemptionConfig{
-					PriorityClasses: map[string]types.PriorityClass{"priority-class-1": {MaximumResourceFractionPerQueueByPool: map[string]map[string]float64{"pool-1": {"cpu": 0.00000001, "memory": 0.9}}}},
-				},
+				PriorityClasses:                   map[string]types.PriorityClass{"priority-class-1": {MaximumResourceFractionPerQueueByPool: map[string]map[string]float64{"pool-1": {"cpu": 0.00000001, "memory": 0.9}}}},
 			},
 			[]queue.Queue{{Name: "queue-1", ResourceLimitsByPriorityClassName: map[string]api.PriorityClassResourceLimits{"priority-class-1": {MaximumResourceFraction: map[string]float64{"cpu": 0.9, "memory": 0.9}}}}},
 		)),
@@ -241,7 +235,7 @@ func makeSchedulingConfig() configuration.SchedulingConfig {
 	return configuration.SchedulingConfig{
 		MaximumResourceFractionToSchedule: map[string]float64{"cpu": 0.1, "memory": 0.1},
 		MaxQueueLookback:                  1000,
-		Preemption:                        configuration.PreemptionConfig{PriorityClasses: map[string]types.PriorityClass{"priority-class-1": {}}},
+		PriorityClasses:                   map[string]types.PriorityClass{"priority-class-1": {}},
 	}
 }
 
diff --git a/internal/scheduler/gang_scheduler_test.go b/internal/scheduler/gang_scheduler_test.go
index 7e13e675..8d02a070 100644
--- a/internal/scheduler/gang_scheduler_test.go
+++ b/internal/scheduler/gang_scheduler_test.go
@@ -452,7 +452,7 @@ func TestGangScheduler(t *testing.T) {
 		"AwayNodeTypes": {
 			SchedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible-away": {
 						Priority:    30000,
 						Preemptible: true,
@@ -471,7 +471,7 @@ func TestGangScheduler(t *testing.T) {
 						},
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible-away"
+				config.DefaultPriorityClassName = "armada-preemptible-away"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name: "node-type-a",
@@ -520,7 +520,7 @@ func TestGangScheduler(t *testing.T) {
 		"Home-away scheduling": {
 			SchedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible": {
 						Priority:    30000,
 						Preemptible: true,
@@ -533,7 +533,7 @@ func TestGangScheduler(t *testing.T) {
 						},
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible"
+				config.DefaultPriorityClassName = "armada-preemptible"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name: "node-type-a",
@@ -572,7 +572,7 @@ func TestGangScheduler(t *testing.T) {
 				nodesById[node.Id] = node
 			}
 			nodeDb, err := nodedb.NewNodeDb(
-				tc.SchedulingConfig.Preemption.PriorityClasses,
+				tc.SchedulingConfig.PriorityClasses,
 				tc.SchedulingConfig.MaxExtraNodesToConsider,
 				tc.SchedulingConfig.IndexedResources,
 				tc.SchedulingConfig.IndexedTaints,
@@ -605,8 +605,8 @@ func TestGangScheduler(t *testing.T) {
 			sctx := schedulercontext.NewSchedulingContext(
 				"executor",
 				"pool",
-				tc.SchedulingConfig.Preemption.PriorityClasses,
-				tc.SchedulingConfig.Preemption.DefaultPriorityClass,
+				tc.SchedulingConfig.PriorityClasses,
+				tc.SchedulingConfig.DefaultPriorityClassName,
 				fairnessCostProvider,
 				rate.NewLimiter(
 					rate.Limit(tc.SchedulingConfig.MaximumSchedulingRate),
@@ -639,7 +639,7 @@ func TestGangScheduler(t *testing.T) {
 			var actualScheduledIndices []int
 			scheduledGangs := 0
 			for i, gang := range tc.Gangs {
-				jctxs := schedulercontext.JobSchedulingContextsFromJobs(tc.SchedulingConfig.Preemption.PriorityClasses, gang)
+				jctxs := schedulercontext.JobSchedulingContextsFromJobs(tc.SchedulingConfig.PriorityClasses, gang)
 				gctx := schedulercontext.NewGangSchedulingContext(jctxs)
 				ok, reason, err := sch.Schedule(armadacontext.Background(), gctx)
 				require.NoError(t, err)
diff --git a/internal/scheduler/preempting_queue_scheduler_test.go b/internal/scheduler/preempting_queue_scheduler_test.go
index 915fa626..63c108b8 100644
--- a/internal/scheduler/preempting_queue_scheduler_test.go
+++ b/internal/scheduler/preempting_queue_scheduler_test.go
@@ -33,11 +33,11 @@ import (
 func TestEvictOversubscribed(t *testing.T) {
 	config := testfixtures.TestSchedulingConfig()
 
-	priorities := types.AllowedPriorities(config.Preemption.PriorityClasses)
+	priorities := types.AllowedPriorities(config.PriorityClasses)
 
 	jobs := append(
-		testfixtures.N1Cpu4GiJobs("A", config.Preemption.DefaultPriorityClass, 20),
-		testfixtures.N1Cpu4GiJobs("A", config.Preemption.DefaultPriorityClass, 20)...,
+		testfixtures.N1Cpu4GiJobs("A", config.DefaultPriorityClassName, 20),
+		testfixtures.N1Cpu4GiJobs("A", config.DefaultPriorityClassName, 20)...,
 	)
 
 	node := testfixtures.Test32CpuNode(priorities)
@@ -47,7 +47,7 @@ func TestEvictOversubscribed(t *testing.T) {
 	err = nodeDb.CreateAndInsertWithJobDbJobsWithTxn(nodeDbTxn, jobs, node)
 	require.NoError(t, err)
 
-	jobDb := jobdb.NewJobDb(config.Preemption.PriorityClasses, config.Preemption.DefaultPriorityClass, 1024)
+	jobDb := jobdb.NewJobDb(config.PriorityClasses, config.DefaultPriorityClassName, 1024)
 	jobDbTxn := jobDb.WriteTxn()
 	err = jobDbTxn.Upsert(jobs)
 	require.NoError(t, err)
@@ -55,8 +55,8 @@ func TestEvictOversubscribed(t *testing.T) {
 	evictor := NewOversubscribedEvictor(
 		NewSchedulerJobRepositoryAdapter(jobDbTxn),
 		nodeDb,
-		config.Preemption.PriorityClasses,
-		config.Preemption.DefaultPriorityClass,
+		config.PriorityClasses,
+		config.DefaultPriorityClassName,
 		1,
 		nil,
 	)
@@ -1481,7 +1481,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 		"home-away preemption, away jobs first": {
 			SchedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible-away": {
 						Priority:    30000,
 						Preemptible: true,
@@ -1493,7 +1493,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 						Preemptible: true,
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible"
+				config.DefaultPriorityClassName = "armada-preemptible"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name:   "gpu",
@@ -1547,7 +1547,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 		"home-away preemption, home jobs first": {
 			SchedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible-away": {
 						Priority:    30000,
 						Preemptible: true,
@@ -1559,7 +1559,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 						Preemptible: true,
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible"
+				config.DefaultPriorityClassName = "armada-preemptible"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name:   "gpu",
@@ -1613,7 +1613,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 		"home-away preemption, mixed nodes, away jobs first": {
 			SchedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible-away": {
 						Priority:    30000,
 						Preemptible: true,
@@ -1625,7 +1625,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 						Preemptible: true,
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible"
+				config.DefaultPriorityClassName = "armada-preemptible"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name:   "gpu",
@@ -1680,7 +1680,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 		"home-away preemption, mixed nodes, home jobs first": {
 			SchedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible-away": {
 						Priority:    30000,
 						Preemptible: true,
@@ -1692,7 +1692,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 						Preemptible: true,
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible"
+				config.DefaultPriorityClassName = "armada-preemptible"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name:   "gpu",
@@ -1756,9 +1756,9 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 			}
 			nodeDbTxn.Commit()
 
-			priorities := types.AllowedPriorities(tc.SchedulingConfig.Preemption.PriorityClasses)
+			priorities := types.AllowedPriorities(tc.SchedulingConfig.PriorityClasses)
 
-			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.Preemption.PriorityClasses, tc.SchedulingConfig.Preemption.DefaultPriorityClass, 1024)
+			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.PriorityClasses, tc.SchedulingConfig.DefaultPriorityClassName, 1024)
 			jobDbTxn := jobDb.WriteTxn()
 
 			// Accounting across scheduling rounds.
@@ -1813,7 +1813,7 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 							nodeId := nodeIdByJobId[job.GetId()]
 							node, err := nodeDb.GetNode(nodeId)
 							require.NoError(t, err)
-							node, err = nodeDb.UnbindJobFromNode(tc.SchedulingConfig.Preemption.PriorityClasses, job, node)
+							node, err = nodeDb.UnbindJobFromNode(tc.SchedulingConfig.PriorityClasses, job, node)
 							require.NoError(t, err)
 							err = nodeDb.Upsert(node)
 							require.NoError(t, err)
@@ -1848,8 +1848,8 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 				sctx := schedulercontext.NewSchedulingContext(
 					"executor",
 					"pool",
-					tc.SchedulingConfig.Preemption.PriorityClasses,
-					tc.SchedulingConfig.Preemption.DefaultPriorityClass,
+					tc.SchedulingConfig.PriorityClasses,
+					tc.SchedulingConfig.DefaultPriorityClassName,
 					fairnessCostProvider,
 					limiter,
 					tc.TotalResources,
@@ -1876,9 +1876,9 @@ func TestPreemptingQueueScheduler(t *testing.T) {
 				sch := NewPreemptingQueueScheduler(
 					sctx,
 					constraints,
-					tc.SchedulingConfig.Preemption.NodeEvictionProbability,
-					tc.SchedulingConfig.Preemption.NodeOversubscriptionEvictionProbability,
-					tc.SchedulingConfig.Preemption.ProtectedFractionOfFairShare,
+					tc.SchedulingConfig.NodeEvictionProbability,
+					tc.SchedulingConfig.NodeOversubscriptionEvictionProbability,
+					tc.SchedulingConfig.ProtectedFractionOfFairShare,
 					NewSchedulerJobRepositoryAdapter(jobDbTxn),
 					nodeDb,
 					nodeIdByJobId,
@@ -2181,7 +2181,7 @@ func BenchmarkPreemptingQueueScheduler(b *testing.B) {
 			}
 			txn.Commit()
 
-			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.Preemption.PriorityClasses, tc.SchedulingConfig.Preemption.DefaultPriorityClass, 1024)
+			jobDb := jobdb.NewJobDb(tc.SchedulingConfig.PriorityClasses, tc.SchedulingConfig.DefaultPriorityClassName, 1024)
 			jobDbTxn := jobDb.WriteTxn()
 			var queuedJobs []*jobdb.Job
 			for _, jobs := range jobsByQueue {
@@ -2212,8 +2212,8 @@ func BenchmarkPreemptingQueueScheduler(b *testing.B) {
 			sctx := schedulercontext.NewSchedulingContext(
 				"executor",
 				"pool",
-				tc.SchedulingConfig.Preemption.PriorityClasses,
-				tc.SchedulingConfig.Preemption.DefaultPriorityClass,
+				tc.SchedulingConfig.PriorityClasses,
+				tc.SchedulingConfig.DefaultPriorityClassName,
 				fairnessCostProvider,
 				limiter,
 				nodeDb.TotalResources(),
@@ -2233,9 +2233,9 @@ func BenchmarkPreemptingQueueScheduler(b *testing.B) {
 			sch := NewPreemptingQueueScheduler(
 				sctx,
 				constraints,
-				tc.SchedulingConfig.Preemption.NodeEvictionProbability,
-				tc.SchedulingConfig.Preemption.NodeOversubscriptionEvictionProbability,
-				tc.SchedulingConfig.Preemption.ProtectedFractionOfFairShare,
+				tc.SchedulingConfig.NodeEvictionProbability,
+				tc.SchedulingConfig.NodeOversubscriptionEvictionProbability,
+				tc.SchedulingConfig.ProtectedFractionOfFairShare,
 				NewSchedulerJobRepositoryAdapter(jobDbTxn),
 				nodeDb,
 				nil,
@@ -2281,8 +2281,8 @@ func BenchmarkPreemptingQueueScheduler(b *testing.B) {
 				sctx := schedulercontext.NewSchedulingContext(
 					"executor",
 					"pool",
-					tc.SchedulingConfig.Preemption.PriorityClasses,
-					tc.SchedulingConfig.Preemption.DefaultPriorityClass,
+					tc.SchedulingConfig.PriorityClasses,
+					tc.SchedulingConfig.DefaultPriorityClassName,
 					fairnessCostProvider,
 					limiter,
 					nodeDb.TotalResources(),
@@ -2295,9 +2295,9 @@ func BenchmarkPreemptingQueueScheduler(b *testing.B) {
 				sch := NewPreemptingQueueScheduler(
 					sctx,
 					constraints,
-					tc.SchedulingConfig.Preemption.NodeEvictionProbability,
-					tc.SchedulingConfig.Preemption.NodeOversubscriptionEvictionProbability,
-					tc.SchedulingConfig.Preemption.ProtectedFractionOfFairShare,
+					tc.SchedulingConfig.NodeEvictionProbability,
+					tc.SchedulingConfig.NodeOversubscriptionEvictionProbability,
+					tc.SchedulingConfig.ProtectedFractionOfFairShare,
 					NewSchedulerJobRepositoryAdapter(jobDbTxn),
 					nodeDb,
 					nil,
diff --git a/internal/scheduler/queue_scheduler_test.go b/internal/scheduler/queue_scheduler_test.go
index 412ed6dc..9eed4775 100644
--- a/internal/scheduler/queue_scheduler_test.go
+++ b/internal/scheduler/queue_scheduler_test.go
@@ -548,7 +548,7 @@ func TestQueueScheduler(t *testing.T) {
 			jobRepo := NewInMemoryJobRepository()
 			jobRepo.EnqueueMany(
 				schedulercontext.JobSchedulingContextsFromJobs(
-					tc.SchedulingConfig.Preemption.PriorityClasses,
+					tc.SchedulingConfig.PriorityClasses,
 					legacySchedulerJobs,
 				),
 			)
@@ -561,8 +561,8 @@ func TestQueueScheduler(t *testing.T) {
 			sctx := schedulercontext.NewSchedulingContext(
 				"executor",
 				"pool",
-				tc.SchedulingConfig.Preemption.PriorityClasses,
-				tc.SchedulingConfig.Preemption.DefaultPriorityClass,
+				tc.SchedulingConfig.PriorityClasses,
+				tc.SchedulingConfig.DefaultPriorityClassName,
 				fairnessCostProvider,
 				rate.NewLimiter(
 					rate.Limit(tc.SchedulingConfig.MaximumSchedulingRate),
@@ -730,7 +730,7 @@ func TestQueueScheduler(t *testing.T) {
 
 func NewNodeDb(config configuration.SchedulingConfig) (*nodedb.NodeDb, error) {
 	nodeDb, err := nodedb.NewNodeDb(
-		config.Preemption.PriorityClasses,
+		config.PriorityClasses,
 		config.MaxExtraNodesToConsider,
 		config.IndexedResources,
 		config.IndexedTaints,
diff --git a/internal/scheduler/scheduling_algo_test.go b/internal/scheduler/scheduling_algo_test.go
index 082994c7..b3cac9d6 100644
--- a/internal/scheduler/scheduling_algo_test.go
+++ b/internal/scheduler/scheduling_algo_test.go
@@ -474,9 +474,9 @@ func TestSchedule(t *testing.T) {
 			} else {
 				assert.Equal(t, tc.expectedScheduledIndices, actualScheduledIndices)
 			}
-			// Sanity check: we've set `RuntimeGangCardinality` for all scheduled jobs.
+			// Sanity check: we've set `GangNumJobsScheduledAnnotation` for all scheduled jobs.
 			for _, job := range scheduledJobs {
-				assert.Contains(t, schedulerResult.AdditionalAnnotationsByJobId[job.Id()], configuration.RuntimeGangCardinality)
+				assert.Contains(t, schedulerResult.AdditionalAnnotationsByJobId[job.Id()], configuration.GangNumJobsScheduledAnnotation)
 			}
 
 			// Check that we failed the correct number of excess jobs when a gang schedules >= minimum cardinality
@@ -553,7 +553,7 @@ func BenchmarkNodeDbConstruction(b *testing.B) {
 				b.StartTimer()
 
 				nodeDb, err := nodedb.NewNodeDb(
-					schedulingConfig.Preemption.PriorityClasses,
+					schedulingConfig.PriorityClasses,
 					schedulingConfig.MaxExtraNodesToConsider,
 					schedulingConfig.IndexedResources,
 					schedulingConfig.IndexedTaints,
diff --git a/internal/scheduler/simulator/simulator_test.go b/internal/scheduler/simulator/simulator_test.go
index ec74a206..67489faf 100644
--- a/internal/scheduler/simulator/simulator_test.go
+++ b/internal/scheduler/simulator/simulator_test.go
@@ -389,7 +389,7 @@ func TestSimulator(t *testing.T) {
 			},
 			schedulingConfig: func() configuration.SchedulingConfig {
 				config := testfixtures.TestSchedulingConfig()
-				config.Preemption.PriorityClasses = map[string]types.PriorityClass{
+				config.PriorityClasses = map[string]types.PriorityClass{
 					"armada-preemptible-away": {
 						Priority:    30000,
 						Preemptible: true,
@@ -401,7 +401,7 @@ func TestSimulator(t *testing.T) {
 						Preemptible: true,
 					},
 				}
-				config.Preemption.DefaultPriorityClass = "armada-preemptible"
+				config.DefaultPriorityClassName = "armada-preemptible"
 				config.WellKnownNodeTypes = []configuration.WellKnownNodeType{
 					{
 						Name:   "gpu-whale",
diff --git a/internal/scheduler/simulator/test_utils.go b/internal/scheduler/simulator/test_utils.go
index 5cf39725..c44ba04a 100644
--- a/internal/scheduler/simulator/test_utils.go
+++ b/internal/scheduler/simulator/test_utils.go
@@ -45,19 +45,18 @@ func GetOneQueue10JobWorkload() *WorkloadSpec {
 
 func GetBasicSchedulingConfig() configuration.SchedulingConfig {
 	return configuration.SchedulingConfig{
-		Preemption: configuration.PreemptionConfig{
-			NodeEvictionProbability: 1.0,
-			PriorityClasses: map[string]types.PriorityClass{
-				"armada-default": {
-					Priority:    30000,
-					Preemptible: false,
-				},
-				"armada-preemptible": {
-					Priority:    30000,
-					Preemptible: true,
-				},
+		NodeEvictionProbability: 1.0,
+		PriorityClasses: map[string]types.PriorityClass{
+			"armada-default": {
+				Priority:    30000,
+				Preemptible: false,
+			},
+			"armada-preemptible": {
+				Priority:    30000,
+				Preemptible: true,
 			},
 		},
+		DefaultPriorityClassName: "armada-default",
 		MaximumResourceFractionToSchedule: map[string]float64{
 			"memory": 0.025,
 			"cpu":    0.025,
diff --git a/internal/scheduler/simulator/testdata/configs/basicSchedulingConfig.yaml b/internal/scheduler/simulator/testdata/configs/basicSchedulingConfig.yaml
index 6153509c..16eecf71 100644
--- a/internal/scheduler/simulator/testdata/configs/basicSchedulingConfig.yaml
+++ b/internal/scheduler/simulator/testdata/configs/basicSchedulingConfig.yaml
@@ -18,12 +18,12 @@ indexedResources:
     resolution: "1Mi"
   - name: "nvidia.com/gpu"
     resolution: "1"
-preemption:
-  nodeEvictionProbability: 1.0
-  priorityClasses:
-    armada-default:
-      priority: 30000
-      preemptible: false
-    armada-preemptible:
-      priority: 30000
-      preemptible: true
+nodeEvictionProbability: 1.0
+priorityClasses:
+  armada-default:
+    priority: 30000
+    preemptible: false
+  armada-preemptible:
+    priority: 30000
+    preemptible: true
+defaultPriorityClassName: "armada-default"
diff --git a/internal/scheduler/testfixtures/testfixtures.go b/internal/scheduler/testfixtures/testfixtures.go
index 4979966e..3e2d62cd 100644
--- a/internal/scheduler/testfixtures/testfixtures.go
+++ b/internal/scheduler/testfixtures/testfixtures.go
@@ -124,12 +124,10 @@ func Repeat[T any](v T, n int) []T {
 
 func TestSchedulingConfig() configuration.SchedulingConfig {
 	return configuration.SchedulingConfig{
-		Preemption: configuration.PreemptionConfig{
-			PriorityClasses:                         maps.Clone(TestPriorityClasses),
-			DefaultPriorityClass:                    TestDefaultPriorityClass,
-			NodeEvictionProbability:                 1.0,
-			NodeOversubscriptionEvictionProbability: 1.0,
-		},
+		PriorityClasses:                             maps.Clone(TestPriorityClasses),
+		DefaultPriorityClassName:                    TestDefaultPriorityClass,
+		NodeEvictionProbability:                     1.0,
+		NodeOversubscriptionEvictionProbability:     1.0,
 		MaximumSchedulingRate:                       math.Inf(1),
 		MaximumSchedulingBurst:                      math.MaxInt,
 		MaximumPerQueueSchedulingRate:               math.Inf(1),
@@ -151,17 +149,17 @@ func WithMaxUnacknowledgedJobsPerExecutorConfig(v uint, config configuration.Sch
 }
 
 func WithProtectedFractionOfFairShareConfig(v float64, config configuration.SchedulingConfig) configuration.SchedulingConfig {
-	config.Preemption.ProtectedFractionOfFairShare = v
+	config.ProtectedFractionOfFairShare = v
 	return config
 }
 
 func WithNodeEvictionProbabilityConfig(p float64, config configuration.SchedulingConfig) configuration.SchedulingConfig {
-	config.Preemption.NodeEvictionProbability = p
+	config.NodeEvictionProbability = p
 	return config
 }
 
 func WithNodeOversubscriptionEvictionProbabilityConfig(p float64, config configuration.SchedulingConfig) configuration.SchedulingConfig {
-	config.Preemption.NodeOversubscriptionEvictionProbability = p
+	config.NodeOversubscriptionEvictionProbability = p
 	return config
 }
 
@@ -177,12 +175,12 @@ func WithRoundLimitsPoolConfig(limits map[string]map[string]float64, config conf
 
 func WithPerPriorityLimitsConfig(limits map[string]map[string]float64, config configuration.SchedulingConfig) configuration.SchedulingConfig {
 	for priorityClassName, limit := range limits {
-		priorityClass, ok := config.Preemption.PriorityClasses[priorityClassName]
+		priorityClass, ok := config.PriorityClasses[priorityClassName]
 		if !ok {
 			panic(fmt.Sprintf("no priority class with name %s", priorityClassName))
 		}
 		// We need to make a copy to avoid mutating the priorityClasses, which are used by other tests too.
-		config.Preemption.PriorityClasses[priorityClassName] = types.PriorityClass{
+		config.PriorityClasses[priorityClassName] = types.PriorityClass{
 			Priority:                              priorityClass.Priority,
 			Preemptible:                           priorityClass.Preemptible,
 			MaximumResourceFractionPerQueue:       limit,