feat: KEP-2437 - PodGroup Creation for Volcano Scheduler

Makefile

@@ -111,6 +111,15 @@ scheduler-plugins-crd: ## Copy the CRDs from the Scheduler Plugins repository to
 	mkdir -p $(EXTERNAL_CRDS_DIR)/scheduler-plugins/
 	cp -f $(SCHEDULER_PLUGINS_ROOT)/manifests/coscheduling/* $(EXTERNAL_CRDS_DIR)/scheduler-plugins
 
+VOLCANO_APIS_ROOT = $(shell go list -m -f "{{.Dir}}" volcano.sh/apis)
+VOLCANO_VERSION = $(shell basename $(VOLCANO_APIS_ROOT) | cut -d'@' -f2)
+VOLCANO_CRD_URL = https://raw.githubusercontent.com/volcano-sh/volcano/$(VOLCANO_VERSION)/config/crd/volcano/bases/scheduling.volcano.sh_podgroups.yaml
+
+.PHONY: volcano-crd
+volcano-crd: ## Copy the CRDs from Volcano repository to the manifests/external-crds directory.
+	mkdir -p $(EXTERNAL_CRDS_DIR)/volcano/
+	curl -sSL $(VOLCANO_CRD_URL) -o $(EXTERNAL_CRDS_DIR)/volcano/scheduling.volcano.sh_podgroups.yaml
+
 # Instructions for code generation.
 .PHONY: manifests
 manifests: controller-gen ## Generate manifests.
@@ -155,7 +164,7 @@ test: ## Run Go unit test.
 	go test $(shell go list ./... | grep -Ev '/(test|cmd|hack|pkg/apis|pkg/client|pkg/util/testing)') -coverprofile cover.out
 
 .PHONY: test-integration
-test-integration: ginkgo envtest jobset-operator-crd scheduler-plugins-crd ## Run Go integration test.
+test-integration: ginkgo envtest jobset-operator-crd scheduler-plugins-crd volcano-crd ## Run Go integration test.
 	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(K8S_VERSION) -p path)" $(GINKGO) -v ./test/integration/...
 
 .PHONY: test-python

api/openapi-spec/swagger.json

@@ -13504,6 +13504,20 @@
           }
         }
       },
+      "scheduling.v1beta1.NetworkTopologySpec": {
+        "type": "object",
+        "properties": {
+          "highestTierAllowed": {
+            "description": "HighestTierAllowed specifies the highest tier that a job allowed to cross when scheduling.",
+            "type": "integer",
+            "format": "int32"
+          },
+          "mode": {
+            "description": "Mode specifies the mode of the network topology constrain.",
+            "type": "string"
+          }
+        }
+      },
       "trainer.v1alpha1.ClusterTrainingRuntime": {
         "description": "ClusterTrainingRuntime represents a training runtime which can be referenced as part of `runtimeRef` API in TrainJob. This resource is a cluster-scoped and can be referenced by TrainJob that created in *any* namespace.",
         "type": "object",
@@ -13876,6 +13890,14 @@
                 "$ref": "#/components/schemas/trainer.v1alpha1.CoschedulingPodGroupPolicySource"
               }
             ]
+          },
+          "volcano": {
+            "description": "Volcano plugin for gang-scheduling.",
+            "allOf": [
+              {
+                "$ref": "#/components/schemas/trainer.v1alpha1.VolcanoPodGroupPolicySource"
+              }
+            ]
           }
         }
       },
@@ -13890,6 +13912,14 @@
                 "$ref": "#/components/schemas/trainer.v1alpha1.CoschedulingPodGroupPolicySource"
               }
             ]
+          },
+          "volcano": {
+            "description": "Volcano plugin for gang-scheduling.",
+            "allOf": [
+              {
+                "$ref": "#/components/schemas/trainer.v1alpha1.VolcanoPodGroupPolicySource"
+              }
+            ]
           }
         }
       },
@@ -14477,6 +14507,20 @@
             ]
           }
         }
+      },
+      "trainer.v1alpha1.VolcanoPodGroupPolicySource": {
+        "description": "VolcanoPodGroupPolicySource represents configuration for the Volcano gang-scheduler.",
+        "type": "object",
+        "properties": {
+          "networkTopology": {
+            "description": "NetworkTopology defines the NetworkTopology config, this field works in conjunction with network topology feature and hyperNode CRD.",
+            "allOf": [
+              {
+                "$ref": "#/components/schemas/scheduling.v1beta1.NetworkTopologySpec"
+              }
+            ]
+          }
+        }
       }
     }
   }

api/python_api/kubeflow_trainer_api/models/__init__.py

@@ -352,6 +352,7 @@
 from kubeflow_trainer_api.models.jobset_v1alpha2_replicated_job_status import JobsetV1alpha2ReplicatedJobStatus
 from kubeflow_trainer_api.models.jobset_v1alpha2_startup_policy import JobsetV1alpha2StartupPolicy
 from kubeflow_trainer_api.models.jobset_v1alpha2_success_policy import JobsetV1alpha2SuccessPolicy
+from kubeflow_trainer_api.models.scheduling_v1beta1_network_topology_spec import SchedulingV1beta1NetworkTopologySpec
 from kubeflow_trainer_api.models.trainer_v1alpha1_cluster_training_runtime import TrainerV1alpha1ClusterTrainingRuntime
 from kubeflow_trainer_api.models.trainer_v1alpha1_cluster_training_runtime_list import TrainerV1alpha1ClusterTrainingRuntimeList
 from kubeflow_trainer_api.models.trainer_v1alpha1_container_override import TrainerV1alpha1ContainerOverride
@@ -379,3 +380,4 @@
 from kubeflow_trainer_api.models.trainer_v1alpha1_training_runtime import TrainerV1alpha1TrainingRuntime
 from kubeflow_trainer_api.models.trainer_v1alpha1_training_runtime_list import TrainerV1alpha1TrainingRuntimeList
 from kubeflow_trainer_api.models.trainer_v1alpha1_training_runtime_spec import TrainerV1alpha1TrainingRuntimeSpec
+from kubeflow_trainer_api.models.trainer_v1alpha1_volcano_pod_group_policy_source import TrainerV1alpha1VolcanoPodGroupPolicySource

api/python_api/kubeflow_trainer_api/models/scheduling_v1beta1_network_topology_spec.py

@@ -0,0 +1,89 @@
+# coding: utf-8
+
+"""
+    Kubeflow Trainer OpenAPI Spec
+
+    No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)
+
+    The version of the OpenAPI document: unversioned
+    Generated by OpenAPI Generator (https://openapi-generator.tech)
+
+    Do not edit the class manually.
+"""  # noqa: E501
+
+
+from __future__ import annotations
+import pprint
+import re  # noqa: F401
+import json
+
+from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr
+from typing import Any, ClassVar, Dict, List, Optional
+from typing import Optional, Set
+from typing_extensions import Self
+
+class SchedulingV1beta1NetworkTopologySpec(BaseModel):
+    """
+    SchedulingV1beta1NetworkTopologySpec
+    """ # noqa: E501
+    highest_tier_allowed: Optional[StrictInt] = Field(default=None, description="HighestTierAllowed specifies the highest tier that a job allowed to cross when scheduling.", alias="highestTierAllowed")
+    mode: Optional[StrictStr] = Field(default=None, description="Mode specifies the mode of the network topology constrain.")
+    __properties: ClassVar[List[str]] = ["highestTierAllowed", "mode"]
+
+    model_config = ConfigDict(
+        populate_by_name=True,
+        validate_assignment=True,
+        protected_namespaces=(),
+    )
+
+
+    def to_str(self) -> str:
+        """Returns the string representation of the model using alias"""
+        return pprint.pformat(self.model_dump(by_alias=True))
+
+    def to_json(self) -> str:
+        """Returns the JSON representation of the model using alias"""
+        # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
+        return json.dumps(self.to_dict())
+
+    @classmethod
+    def from_json(cls, json_str: str) -> Optional[Self]:
+        """Create an instance of SchedulingV1beta1NetworkTopologySpec from a JSON string"""
+        return cls.from_dict(json.loads(json_str))
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return the dictionary representation of the model using alias.
+
+        This has the following differences from calling pydantic's
+        `self.model_dump(by_alias=True)`:
+
+        * `None` is only added to the output dict for nullable fields that
+          were set at model initialization. Other fields with value `None`
+          are ignored.
+        """
+        excluded_fields: Set[str] = set([
+        ])
+
+        _dict = self.model_dump(
+            by_alias=True,
+            exclude=excluded_fields,
+            exclude_none=True,
+        )
+        return _dict
+
+    @classmethod
+    def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
+        """Create an instance of SchedulingV1beta1NetworkTopologySpec from a dict"""
+        if obj is None:
+            return None
+
+        if not isinstance(obj, dict):
+            return cls.model_validate(obj)
+
+        _obj = cls.model_validate({
+            "highestTierAllowed": obj.get("highestTierAllowed"),
+            "mode": obj.get("mode")
+        })
+        return _obj
+
+

api/python_api/kubeflow_trainer_api/models/trainer_v1alpha1_pod_group_policy.py

@@ -20,6 +20,7 @@
 from pydantic import BaseModel, ConfigDict, Field
 from typing import Any, ClassVar, Dict, List, Optional
 from kubeflow_trainer_api.models.trainer_v1alpha1_coscheduling_pod_group_policy_source import TrainerV1alpha1CoschedulingPodGroupPolicySource
+from kubeflow_trainer_api.models.trainer_v1alpha1_volcano_pod_group_policy_source import TrainerV1alpha1VolcanoPodGroupPolicySource
 from typing import Optional, Set
 from typing_extensions import Self
 
@@ -28,7 +29,8 @@ class TrainerV1alpha1PodGroupPolicy(BaseModel):
     PodGroupPolicy represents a PodGroup configuration for gang-scheduling.
     """ # noqa: E501
     coscheduling: Optional[TrainerV1alpha1CoschedulingPodGroupPolicySource] = Field(default=None, description="Coscheduling plugin from the Kubernetes scheduler-plugins for gang-scheduling.")
-    __properties: ClassVar[List[str]] = ["coscheduling"]
+    volcano: Optional[TrainerV1alpha1VolcanoPodGroupPolicySource] = Field(default=None, description="Volcano plugin for gang-scheduling.")
+    __properties: ClassVar[List[str]] = ["coscheduling", "volcano"]
 
     model_config = ConfigDict(
         populate_by_name=True,
@@ -72,6 +74,9 @@ def to_dict(self) -> Dict[str, Any]:
         # override the default output from pydantic by calling `to_dict()` of coscheduling
         if self.coscheduling:
             _dict['coscheduling'] = self.coscheduling.to_dict()
+        # override the default output from pydantic by calling `to_dict()` of volcano
+        if self.volcano:
+            _dict['volcano'] = self.volcano.to_dict()
         return _dict
 
     @classmethod
@@ -84,7 +89,8 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
             return cls.model_validate(obj)
 
         _obj = cls.model_validate({
-            "coscheduling": TrainerV1alpha1CoschedulingPodGroupPolicySource.from_dict(obj["coscheduling"]) if obj.get("coscheduling") is not None else None
+            "coscheduling": TrainerV1alpha1CoschedulingPodGroupPolicySource.from_dict(obj["coscheduling"]) if obj.get("coscheduling") is not None else None,
+            "volcano": TrainerV1alpha1VolcanoPodGroupPolicySource.from_dict(obj["volcano"]) if obj.get("volcano") is not None else None
         })
         return _obj
 

api/python_api/kubeflow_trainer_api/models/trainer_v1alpha1_pod_group_policy_source.py

@@ -20,6 +20,7 @@
 from pydantic import BaseModel, ConfigDict, Field
 from typing import Any, ClassVar, Dict, List, Optional
 from kubeflow_trainer_api.models.trainer_v1alpha1_coscheduling_pod_group_policy_source import TrainerV1alpha1CoschedulingPodGroupPolicySource
+from kubeflow_trainer_api.models.trainer_v1alpha1_volcano_pod_group_policy_source import TrainerV1alpha1VolcanoPodGroupPolicySource
 from typing import Optional, Set
 from typing_extensions import Self
 
@@ -28,7 +29,8 @@ class TrainerV1alpha1PodGroupPolicySource(BaseModel):
     PodGroupPolicySource represents supported plugins for gang-scheduling. Only one of its members may be specified.
     """ # noqa: E501
     coscheduling: Optional[TrainerV1alpha1CoschedulingPodGroupPolicySource] = Field(default=None, description="Coscheduling plugin from the Kubernetes scheduler-plugins for gang-scheduling.")
-    __properties: ClassVar[List[str]] = ["coscheduling"]
+    volcano: Optional[TrainerV1alpha1VolcanoPodGroupPolicySource] = Field(default=None, description="Volcano plugin for gang-scheduling.")
+    __properties: ClassVar[List[str]] = ["coscheduling", "volcano"]
 
     model_config = ConfigDict(
         populate_by_name=True,
@@ -72,6 +74,9 @@ def to_dict(self) -> Dict[str, Any]:
         # override the default output from pydantic by calling `to_dict()` of coscheduling
         if self.coscheduling:
             _dict['coscheduling'] = self.coscheduling.to_dict()
+        # override the default output from pydantic by calling `to_dict()` of volcano
+        if self.volcano:
+            _dict['volcano'] = self.volcano.to_dict()
         return _dict
 
     @classmethod
@@ -84,7 +89,8 @@ def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
             return cls.model_validate(obj)
 
         _obj = cls.model_validate({
-            "coscheduling": TrainerV1alpha1CoschedulingPodGroupPolicySource.from_dict(obj["coscheduling"]) if obj.get("coscheduling") is not None else None
+            "coscheduling": TrainerV1alpha1CoschedulingPodGroupPolicySource.from_dict(obj["coscheduling"]) if obj.get("coscheduling") is not None else None,
+            "volcano": TrainerV1alpha1VolcanoPodGroupPolicySource.from_dict(obj["volcano"]) if obj.get("volcano") is not None else None
         })
         return _obj
 

api/python_api/kubeflow_trainer_api/models/trainer_v1alpha1_volcano_pod_group_policy_source.py

@@ -0,0 +1,91 @@
+# coding: utf-8
+
+"""
+    Kubeflow Trainer OpenAPI Spec
+
+    No description provided (generated by Openapi Generator https://github.com/openapitools/openapi-generator)
+
+    The version of the OpenAPI document: unversioned
+    Generated by OpenAPI Generator (https://openapi-generator.tech)
+
+    Do not edit the class manually.
+"""  # noqa: E501
+
+
+from __future__ import annotations
+import pprint
+import re  # noqa: F401
+import json
+
+from pydantic import BaseModel, ConfigDict, Field
+from typing import Any, ClassVar, Dict, List, Optional
+from kubeflow_trainer_api.models.scheduling_v1beta1_network_topology_spec import SchedulingV1beta1NetworkTopologySpec
+from typing import Optional, Set
+from typing_extensions import Self
+
+class TrainerV1alpha1VolcanoPodGroupPolicySource(BaseModel):
+    """
+    VolcanoPodGroupPolicySource represents configuration for the Volcano gang-scheduler.
+    """ # noqa: E501
+    network_topology: Optional[SchedulingV1beta1NetworkTopologySpec] = Field(default=None, description="NetworkTopology defines the NetworkTopology config, this field works in conjunction with network topology feature and hyperNode CRD.", alias="networkTopology")
+    __properties: ClassVar[List[str]] = ["networkTopology"]
+
+    model_config = ConfigDict(
+        populate_by_name=True,
+        validate_assignment=True,
+        protected_namespaces=(),
+    )
+
+
+    def to_str(self) -> str:
+        """Returns the string representation of the model using alias"""
+        return pprint.pformat(self.model_dump(by_alias=True))
+
+    def to_json(self) -> str:
+        """Returns the JSON representation of the model using alias"""
+        # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead
+        return json.dumps(self.to_dict())
+
+    @classmethod
+    def from_json(cls, json_str: str) -> Optional[Self]:
+        """Create an instance of TrainerV1alpha1VolcanoPodGroupPolicySource from a JSON string"""
+        return cls.from_dict(json.loads(json_str))
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return the dictionary representation of the model using alias.
+
+        This has the following differences from calling pydantic's
+        `self.model_dump(by_alias=True)`:
+
+        * `None` is only added to the output dict for nullable fields that
+          were set at model initialization. Other fields with value `None`
+          are ignored.
+        """
+        excluded_fields: Set[str] = set([
+        ])
+
+        _dict = self.model_dump(
+            by_alias=True,
+            exclude=excluded_fields,
+            exclude_none=True,
+        )
+        # override the default output from pydantic by calling `to_dict()` of network_topology
+        if self.network_topology:
+            _dict['networkTopology'] = self.network_topology.to_dict()
+        return _dict
+
+    @classmethod
+    def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]:
+        """Create an instance of TrainerV1alpha1VolcanoPodGroupPolicySource from a dict"""
+        if obj is None:
+            return None
+
+        if not isinstance(obj, dict):
+            return cls.model_validate(obj)
+
+        _obj = cls.model_validate({
+            "networkTopology": SchedulingV1beta1NetworkTopologySpec.from_dict(obj["networkTopology"]) if obj.get("networkTopology") is not None else None
+        })
+        return _obj
+
+

charts/kubeflow-trainer/crds/trainer.kubeflow.org_clustertrainingruntimes.yaml

@@ -627,6 +627,29 @@ spec:
                         format: int32
                         type: integer
                     type: object
+                  volcano:
+                    description: Volcano plugin for gang-scheduling.
+                    properties:
+                      networkTopology:
+                        description: NetworkTopology defines the NetworkTopology config,
+                          this field works in conjunction with network topology feature
+                          and hyperNode CRD.
+                        properties:
+                          highestTierAllowed:
+                            default: 1
+                            description: HighestTierAllowed specifies the highest
+                              tier that a job allowed to cross when scheduling.
+                            type: integer
+                          mode:
+                            default: hard
+                            description: Mode specifies the mode of the network topology
+                              constrain.
+                            enum:
+                            - hard
+                            - soft
+                            type: string
+                        type: object
+                    type: object
                 type: object
               template:
                 description: JobSet template which will be used by TrainJob.