Implement Gang-Scheduling

Tiltfile

@@ -7,7 +7,7 @@
 analytics_settings(False)
 
 # Use the ACTIVE_DEPLOYMENTS env var to select which Cortex bundles to deploy.
-ACTIVE_DEPLOYMENTS_ENV = os.getenv('ACTIVE_DEPLOYMENTS', 'nova,manila,cinder,ironcore,pods')
+ACTIVE_DEPLOYMENTS_ENV = os.getenv('ACTIVE_DEPLOYMENTS', 'nova,manila,cinder,ironcore,pods,podgroupsets')
 if ACTIVE_DEPLOYMENTS_ENV == "":
     ACTIVE_DEPLOYMENTS = [] # Catch "".split(",") = [""]
 else:
@@ -55,6 +55,7 @@ bundle_charts = [
     ('helm/bundles/cortex-cinder', 'cortex-cinder'),
     ('helm/bundles/cortex-ironcore', 'cortex-ironcore'),
     ('helm/bundles/cortex-pods', 'cortex-pods'),
+    ('helm/bundles/cortex-podgroupsets', 'cortex-podgroupsets'),
 ]
 dep_charts = {
     'cortex-crds': [
@@ -80,6 +81,10 @@ dep_charts = {
         ('helm/library/cortex-postgres', 'cortex-postgres'),
         ('dist/chart', 'cortex'),
     ],
+    'cortex-podgroupsets': [
+        ('helm/library/cortex-postgres', 'cortex-postgres'),
+        ('dist/chart', 'cortex'),
+    ],
 }
 
 for (bundle_chart_path, bundle_chart_name) in bundle_charts:
@@ -120,6 +125,10 @@ k8s_yaml(helm('./helm/bundles/cortex-crds', name='cortex-crds', set=[
     'cortex.crd.pods.enable=true',
     'cortex.rbac.pods.enable=true',
     'cortex.namePrefix=cortex-pods',
+
+    'cortex.crd.podgroupsets.enable=true',
+    'cortex.rbac.podgroupsets.enable=true',
+    'cortex.namePrefix=cortex-podgroupsets',
 ]))
 
 if 'nova' in ACTIVE_DEPLOYMENTS:
@@ -198,6 +207,13 @@ if 'pods' in ACTIVE_DEPLOYMENTS:
     k8s_yaml('samples/pods/pod.yaml')
     k8s_resource('test-pod', labels=['Cortex-Pods'])
 
+if 'podgroupsets' in ACTIVE_DEPLOYMENTS:
+    print("Activating Cortex PodGroupSets bundle")
+    k8s_yaml(helm('./helm/bundles/cortex-podgroupsets', name='cortex-podgroupsets', values=tilt_values),)
+    k8s_resource('cortex-podgroupsets-controller-manager', labels=['Cortex-PodGroupSets'])
+    # Deploy example resources
+    k8s_yaml('samples/podgroupsets/podgroupset.yaml')
+
 ########### Dev Dependencies
 local('sh helm/sync.sh helm/dev/cortex-prometheus-operator')
 k8s_yaml(helm('./helm/dev/cortex-prometheus-operator', name='cortex-prometheus-operator')) # Operator
@@ -232,4 +248,4 @@ k8s_resource('cortex-plutono', port_forwards=[
     port_forward(5000, 3000, name='plutono'),
 ], links=[
     link('http://localhost:5000/d/cortex/cortex?orgId=1', 'cortex dashboard'),
-], labels=['Monitoring'])
+], labels=['Monitoring'])

api/delegation/podgroupsets/messages.go

@@ -0,0 +1,39 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package podgroupsets
+
+import (
+	"log/slog"
+
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+)
+
+type PodGroupSetPipelineRequest struct {
+	PodGroupSet v1alpha1.PodGroupSet `json:"podGroupSet"`
+	Nodes       []corev1.Node        `json:"nodes"`
+}
+
+func (r PodGroupSetPipelineRequest) GetSubjects() []string {
+	hosts := make([]string, len(r.Nodes))
+	for i, host := range r.Nodes {
+		hosts[i] = host.Name
+	}
+	return hosts
+}
+
+func (r PodGroupSetPipelineRequest) GetWeights() map[string]float64 {
+	weights := make(map[string]float64, len(r.Nodes))
+	for _, node := range r.Nodes {
+		weights[node.Name] = 0.0
+	}
+	return weights
+}
+
+func (r PodGroupSetPipelineRequest) GetTraceLogArgs() []slog.Attr {
+	return []slog.Attr{
+		slog.String("podGroupSet", r.PodGroupSet.Name),
+		slog.String("namespace", r.PodGroupSet.Namespace),
+	}
+}

api/v1alpha1/common_types.go

@@ -25,4 +25,7 @@ const (
 	// SchedulingDomainPods indicates scheduling related to Kubernetes pods,
 	// which are the smallest deployable units in a Kubernetes cluster.
 	SchedulingDomainPods SchedulingDomain = "pods"
+	// SchedulingDomainPodGroupSets indicates scheduling related to PodGroupSets,
+	// which allows for gang scheduling of multiple pods.
+	SchedulingDomainPodGroupSets SchedulingDomain = "podgroupsets"
 )

api/v1alpha1/decision_types.go

@@ -39,6 +39,9 @@ type DecisionSpec struct {
 	// If the type is "pod", this field contains the pod reference.
 	// +kubebuilder:validation:Optional
 	PodRef *corev1.ObjectReference `json:"podRef,omitempty"`
+	// If the type is "podgroupset", this field contains the pod group set reference.
+	// +kubebuilder:validation:Optional
+	PodGroupSetRef *corev1.ObjectReference `json:"podGroupSetRef,omitempty"`
 }
 
 type StepResult struct {
@@ -68,6 +71,10 @@ type DecisionResult struct {
 	// The first element of the ordered hosts is considered the target host.
 	// +kubebuilder:validation:Optional
 	TargetHost *string `json:"targetHost,omitempty"`
+	// TargetPlacements contains the mapping of resource names to target hosts.
+	// This is used for bulk scheduling decisions like PodGroupSets.
+	// +kubebuilder:validation:Optional
+	TargetPlacements map[string]string `json:"targetPlacements,omitempty"`
 }
 
 const (

api/v1alpha1/pipeline_types.go

@@ -72,6 +72,8 @@ const (
 	// Pipeline containing descheduler steps for generating descheduling
 	// recommendations.
 	PipelineTypeDescheduler PipelineType = "descheduler"
+	// Pipeline containing gang scheduling logic.
+	PipelineTypeGang PipelineType = "gang"
 )
 
 type PipelineSpec struct {

api/v1alpha1/podgroupset_types.go

@@ -0,0 +1,55 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package v1alpha1
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+type PodGroupSpec struct {
+	Replicas int32          `json:"replicas"`
+	PodSpec  corev1.PodSpec `json:"podSpec"`
+}
+
+type PodGroup struct {
+	Name string       `json:"name"`
+	Spec PodGroupSpec `json:"spec"`
+}
+
+type PodGroupSetSpec struct {
+	PodGroups []PodGroup `json:"podGroups"`
+}
+
+type PodGroupSetStatus struct {
+	// The current status conditions of the pod group set.
+	// +kubebuilder:validation:Optional
+	Conditions []metav1.Condition `json:"conditions,omitempty" patchStrategy:"merge" patchMergeKey:"type"`
+}
+
+// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
+// +kubebuilder:resource:scope=Namespaced
+
+// PodGroupSet is the Schema for the podgroupsets API
+type PodGroupSet struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   PodGroupSetSpec   `json:"spec,omitempty"`
+	Status PodGroupSetStatus `json:"status,omitempty"`
+}
+
+// +kubebuilder:object:root=true
+
+// PodGroupSetList contains a list of PodGroupSet
+type PodGroupSetList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []PodGroupSet `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&PodGroupSet{}, &PodGroupSetList{})
+}

cmd/main.go

@@ -43,6 +43,7 @@ import (
 	decisionsmachines "github.com/cobaltcore-dev/cortex/internal/scheduling/decisions/machines"
 	decisionsmanila "github.com/cobaltcore-dev/cortex/internal/scheduling/decisions/manila"
 	decisionsnova "github.com/cobaltcore-dev/cortex/internal/scheduling/decisions/nova"
+	decisionpodgroupsets "github.com/cobaltcore-dev/cortex/internal/scheduling/decisions/podgroupsets"
 	decisionpods "github.com/cobaltcore-dev/cortex/internal/scheduling/decisions/pods"
 	deschedulingnova "github.com/cobaltcore-dev/cortex/internal/scheduling/descheduling/nova"
 	cindere2e "github.com/cobaltcore-dev/cortex/internal/scheduling/e2e/cinder"
@@ -382,6 +383,18 @@ func main() {
 			os.Exit(1)
 		}
 	}
+	if slices.Contains(config.EnabledControllers, "podgroupsets-decisions-pipeline-controller") {
+		controller := &decisionpodgroupsets.DecisionPipelineController{
+			Monitor: pipelineMonitor,
+			Conf:    config,
+		}
+		// Inferred through the base controller.
+		controller.Client = multiclusterClient
+		if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil {
+			setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler")
+			os.Exit(1)
+		}
+	}
 	if slices.Contains(config.EnabledControllers, "explanation-controller") {
 		// Setup a controller which will reconcile the history and explanation for
 		// decision resources.