Add the Networking section to SlurmQueues.ComputeResources including only PlacementGroup (aws#4330)

This change allows for creating configs that include placement groups for
specific compute resources.  It changes the way Queue level placement groups
are configured as well.  This aligns better with AWS EC2 placement group
guidance to reduce insufficient capacity errors.  Parallel cluster
managed placement groups will no longer be created to include multiple
compute resources.

If enabled at the queue level and not assigned a name, every compute resource
will get a separate parallel cluster managed placement group.

If assigned a name at the queue level, there will be a single
placement group for all compute resources in the queue.

If enabled at the compute resource level and not assigned a name, this config
will override the queue level config and that resource will get its own
managed placement group.

If assigned a name at the compute resource level, this config will override the
queue level config and that resource will use the assigned placement group.

Signed-off-by: Ryan Anderson <[email protected]>

Author: dreambeyondorange

CHANGELOG.md

@@ -17,6 +17,7 @@ CHANGELOG
 - Disable Multithreading through script executed by cloud-init and not through CpuOptions set into Launch Template.
 - Add support for multiple instance types in the same Compute Resource.
 - Add support for a Name field in PlacementGroup as the preferred naming method.
+- Add support for Networking.PlacementGroup in the SlurmQueues.ComputeResources section
 
 **BUG FIXES**
 - Fix validation of parameter `SharedStorage/EfsSettings`: now validation fails when `FileSystemId` is specified

cli/src/pcluster/config/cluster_config.py

@@ -518,8 +518,8 @@ def __init__(self, http_proxy_address: str = None):
 class _BaseNetworking(Resource):
     """Represent the networking configuration shared by head node and compute node."""
 
-    def __init__(self, security_groups: List[str] = None, additional_security_groups: List[str] = None):
-        super().__init__()
+    def __init__(self, security_groups: List[str] = None, additional_security_groups: List[str] = None, **kwargs):
+        super().__init__(**kwargs)
         self.security_groups = Resource.init_param(security_groups)
         self.additional_security_groups = Resource.init_param(additional_security_groups)
 
@@ -548,17 +548,35 @@ def availability_zone(self):
 
 
 class PlacementGroup(Resource):
-    """Represent the placement group for the Queue networking."""
+    """Represent the placement group for networking."""
 
-    def __init__(self, enabled: bool = None, name: str = None, id: str = None):
-        super().__init__()
-        self.enabled = Resource.init_param(enabled, default=False)
+    def __init__(self, enabled: bool = None, name: str = None, id: str = None, **kwargs):
+        super().__init__(**kwargs)
+        self.enabled = Resource.init_param(enabled)
         self.name = Resource.init_param(name)
         self.id = Resource.init_param(id)  # Duplicate of name
 
     def _register_validators(self):
         self._register_validator(PlacementGroupNamingValidator, placement_group=self)
 
+    @property
+    def is_enabled_and_unassigned(self) -> bool:
+        """Check if the PlacementGroup is enabled without a name or id."""
+        return not (self.id or self.name) and self.enabled
+
+    @property
+    def assignment(self) -> str:
+        """Check if the placement group has a name or id and get it, preferring the name if it exists."""
+        return self.name or self.id
+
+
+class SlurmComputeResourceNetworking(Resource):
+    """Represent the networking configuration for the compute resource."""
+
+    def __init__(self, placement_group: PlacementGroup = None, **kwargs):
+        super().__init__(**kwargs)
+        self.placement_group = placement_group or PlacementGroup(implied=True)
+
 
 class _QueueNetworking(_BaseNetworking):
     """Represent the networking configuration for the Queue."""
@@ -574,7 +592,7 @@ class SlurmQueueNetworking(_QueueNetworking):
 
     def __init__(self, placement_group: PlacementGroup = None, proxy: Proxy = None, **kwargs):
         super().__init__(**kwargs)
-        self.placement_group = placement_group
+        self.placement_group = placement_group or PlacementGroup(implied=True)
         self.proxy = proxy
 
 
@@ -1623,6 +1641,7 @@ def __init__(
         disable_simultaneous_multithreading: bool = None,
         schedulable_memory: int = None,
         capacity_reservation_target: CapacityReservationTarget = None,
+        networking: SlurmComputeResourceNetworking = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -1637,6 +1656,7 @@ def __init__(
         self.capacity_reservation_target = capacity_reservation_target
         self._instance_types_with_instance_storage = []
         self._instance_type_info_map = {}
+        self.networking = networking or SlurmComputeResourceNetworking(implied=True)
 
     @staticmethod
     def fetch_instance_type_info(instance_type) -> InstanceTypeInfo:
@@ -1792,11 +1812,25 @@ def disable_simultaneous_multithreading_manually(self) -> bool:
         return self.disable_simultaneous_multithreading and self.instance_type_info.default_threads_per_core() > 1
 
 
+class SchedulerPluginComputeResource(SlurmComputeResource):
+    """Represent the Scheduler Plugin Compute Resource."""
+
+    def __init__(
+        self,
+        custom_settings: Dict = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.custom_settings = custom_settings
+
+
 class _CommonQueue(BaseQueue):
     """Represent the Common Queue resource between Slurm and Scheduler Plugin."""
 
     def __init__(
         self,
+        compute_resources: List[Union[_BaseSlurmComputeResource, SchedulerPluginComputeResource]],
+        networking: Union[SlurmQueueNetworking, SchedulerPluginQueueNetworking],
         compute_settings: ComputeSettings = None,
         custom_actions: CustomActions = None,
         iam: Iam = None,
@@ -1810,6 +1844,8 @@ def __init__(
         self.iam = iam or Iam(implied=True)
         self.image = image
         self.capacity_reservation_target = capacity_reservation_target
+        self.compute_resources = compute_resources
+        self.networking = networking
 
     @property
     def instance_role(self):
@@ -1829,6 +1865,43 @@ def queue_ami(self):
         else:
             return None
 
+    def get_managed_placement_group_keys(self) -> List[str]:
+        managed_placement_group_keys = []
+        for resource in self.compute_resources:
+            chosen_pg = (
+                resource.networking.placement_group
+                if not resource.networking.placement_group.implied
+                else self.networking.placement_group
+            )
+            if chosen_pg.is_enabled_and_unassigned:
+                managed_placement_group_keys.append(f"{self.name}-{resource.name}")
+        return managed_placement_group_keys
+
+    def get_placement_group_key_for_compute_resource(
+        self, compute_resource: Union[_BaseSlurmComputeResource, SchedulerPluginComputeResource]
+    ) -> (str, bool):
+        # prefer compute level groups over queue level groups
+        placement_group_key, managed = None, None
+        cr_pg = compute_resource.networking.placement_group
+        if cr_pg.assignment:
+            placement_group_key, managed = cr_pg.assignment, False
+        elif cr_pg.enabled:
+            placement_group_key, managed = f"{self.name}-{compute_resource.name}", True
+        elif cr_pg.enabled is False:
+            placement_group_key, managed = None, False
+        elif self.networking.placement_group.assignment:
+            placement_group_key, managed = self.networking.placement_group.assignment, False
+        elif self.networking.placement_group.enabled:
+            placement_group_key, managed = f"{self.name}-{compute_resource.name}", True
+        return placement_group_key, managed
+
+    def is_placement_group_disabled_for_compute_resource(self, compute_resource_pg_enabled: bool) -> bool:
+        return (
+            compute_resource_pg_enabled is False
+            or self.networking.placement_group.enabled is False
+            and compute_resource_pg_enabled is None
+        )
+
 
 class AllocationStrategy(Enum):
     """Define supported allocation strategies."""
@@ -1842,15 +1915,13 @@ class SlurmQueue(_CommonQueue):
 
     def __init__(
         self,
-        compute_resources: List[_BaseSlurmComputeResource],
-        networking: SlurmQueueNetworking,
         allocation_strategy: str = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
-        self.compute_resources = compute_resources
-        self.networking = networking
-        if any(isinstance(compute_resource, SlurmFlexibleComputeResource) for compute_resource in compute_resources):
+        if any(
+            isinstance(compute_resource, SlurmFlexibleComputeResource) for compute_resource in self.compute_resources
+        ):
             self.allocation_strategy = (
                 AllocationStrategy[to_snake_case(allocation_strategy).upper()]
                 if allocation_strategy
@@ -1896,7 +1967,10 @@ def _register_validators(self):
             self._register_validator(
                 EfaPlacementGroupValidator,
                 efa_enabled=compute_resource.efa.enabled,
-                placement_group=self.networking.placement_group,
+                placement_group_key=self.get_placement_group_key_for_compute_resource(compute_resource)[0],
+                placement_group_disabled=self.is_placement_group_disabled_for_compute_resource(
+                    compute_resource.networking.placement_group.enabled
+                ),
             )
             for instance_type in compute_resource.instance_types:
                 self._register_validator(
@@ -1967,31 +2041,15 @@ def _register_validators(self):
         )
 
 
-class SchedulerPluginComputeResource(SlurmComputeResource):
-    """Represent the Scheduler Plugin Compute Resource."""
-
-    def __init__(
-        self,
-        custom_settings: Dict = None,
-        **kwargs,
-    ):
-        super().__init__(**kwargs)
-        self.custom_settings = custom_settings
-
-
 class SchedulerPluginQueue(_CommonQueue):
     """Represent the Scheduler Plugin queue."""
 
     def __init__(
         self,
-        compute_resources: List[SchedulerPluginComputeResource],
-        networking: SchedulerPluginQueueNetworking,
         custom_settings: Dict = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
-        self.compute_resources = compute_resources
-        self.networking = networking
         self.custom_settings = custom_settings
 
     def _register_validators(self):
@@ -2014,7 +2072,10 @@ def _register_validators(self):
             self._register_validator(
                 EfaPlacementGroupValidator,
                 efa_enabled=compute_resource.efa.enabled,
-                placement_group=self.networking.placement_group,
+                placement_group_key=self.get_placement_group_key_for_compute_resource(compute_resource)[0],
+                placement_group_disabled=self.is_placement_group_disabled_for_compute_resource(
+                    compute_resource.networking.placement_group.enabled
+                ),
             )
 
     @property

cli/src/pcluster/schemas/cluster_schema.py

@@ -100,6 +100,7 @@
     SharedFsxLustre,
     SlurmClusterConfig,
     SlurmComputeResource,
+    SlurmComputeResourceNetworking,
     SlurmFlexibleComputeResource,
     SlurmQueue,
     SlurmQueueNetworking,
@@ -1128,6 +1129,19 @@ class _ComputeResourceSchema(BaseSchema):
     name = fields.Str(required=True, metadata={"update_policy": UpdatePolicy.UNSUPPORTED})
 
 
+class SlurmComputeResourceNetworkingSchema(BaseSchema):
+    """Represent the Networking schema of the Slurm ComputeResource."""
+
+    placement_group = fields.Nested(
+        PlacementGroupSchema, metadata={"update_policy": UpdatePolicy.QUEUE_UPDATE_STRATEGY}
+    )
+
+    @post_load
+    def make_resource(self, data, **kwargs):
+        """Generate resource."""
+        return SlurmComputeResourceNetworking(**data)
+
+
 class SlurmComputeResourceSchema(_ComputeResourceSchema):
     """Represent the schema of the Slurm ComputeResource."""
 
@@ -1148,6 +1162,9 @@ class SlurmComputeResourceSchema(_ComputeResourceSchema):
     capacity_reservation_target = fields.Nested(
         CapacityReservationTargetSchema, metadata={"update_policy": UpdatePolicy.QUEUE_UPDATE_STRATEGY}
     )
+    networking = fields.Nested(
+        SlurmComputeResourceNetworkingSchema, metadata={"update_policy": UpdatePolicy.QUEUE_UPDATE_STRATEGY}
+    )
 
     @validates_schema
     def no_coexist_instance_type_flexibility(self, data, **kwargs):

cli/src/pcluster/schemas/common_schema.py

@@ -64,7 +64,7 @@ class BaseSchema(Schema):
 
     def on_bind_field(self, field_name, field_obj):
         """
-        Bind PascalCase in the config with with snake_case in Python.
+        Bind PascalCase in the config with snake_case in Python.
 
         For example, subnet_id in the code is automatically bind with SubnetId in the config file.
         The bind can be overwritten by specifying data_key.
@@ -76,11 +76,11 @@ def on_bind_field(self, field_name, field_obj):
     @staticmethod
     def fields_coexist(data, field_list, one_required=False, **kwargs):
         """
-        Check if at least two fields in the filed lists co-exist in the schema.
+        Check if at least two fields in the field list co-exist in the schema.
 
         :param data: data to be checked
         :param field_list: list including the name of the fields to check
-        :param one_required: True if one of the field is required to be existed
+        :param one_required: True if one of the fields is required to exist
         :return: True if one and only one field is not None
         """
         if kwargs.get("partial"):

cli/src/pcluster/templates/cluster_stack.py

@@ -1364,47 +1364,39 @@ def _add_policies_to_cleanup_resources_lambda_role(self):
     def _add_placement_groups(self) -> Dict[str, ec2.CfnPlacementGroup]:
         managed_placement_groups = {}
         for queue in self._config.scheduling.queues:
-            if (
-                queue.networking.placement_group
-                and queue.networking.placement_group.enabled
-                and not (queue.networking.placement_group.id or queue.networking.placement_group.name)
-            ):
-                managed_placement_groups[queue.name] = ec2.CfnPlacementGroup(
-                    self, f"PlacementGroup{create_hash_suffix(queue.name)}", strategy="cluster"
+            for key in queue.get_managed_placement_group_keys():
+                managed_placement_groups[key] = ec2.CfnPlacementGroup(
+                    self,
+                    f"PlacementGroup{create_hash_suffix(key)}",
+                    strategy="cluster",
                 )
         return managed_placement_groups
 
+    @staticmethod
+    def _get_placement_group_for_compute_resource(queue, managed_placement_groups, compute_resource) -> str:
+        placement_group_key, managed = queue.get_placement_group_key_for_compute_resource(compute_resource)
+        return managed_placement_groups[placement_group_key].ref if managed else placement_group_key
+
     def _add_launch_templates(self, managed_placement_groups, instance_profiles):
         compute_launch_templates = {}
         for queue in self._config.scheduling.queues:
             compute_launch_templates[queue.name] = {}
             queue_lt_security_groups = get_queue_security_groups_full(self._compute_security_group, queue)
-
-            queue_placement_group = None
-            if queue.networking.placement_group:
-                if (queue.networking.placement_group.id or queue.networking.placement_group.name) and (
-                    queue.networking.placement_group.enabled or queue.networking.placement_group.is_implied("enabled")
-                ):
-                    queue_placement_group = queue.networking.placement_group.name or queue.networking.placement_group.id
-                elif queue.networking.placement_group.enabled:
-                    queue_placement_group = managed_placement_groups[queue.name].ref
-
             queue_pre_install_action, queue_post_install_action = (None, None)
             if queue.custom_actions:
                 queue_pre_install_action = queue.custom_actions.on_node_start
                 queue_post_install_action = queue.custom_actions.on_node_configured
 
-            for compute_resource in queue.compute_resources:
-                launch_template = self._add_compute_resource_launch_template(
+            for resource in queue.compute_resources:
+                compute_launch_templates[queue.name][resource.name] = self._add_compute_resource_launch_template(
                     queue,
-                    compute_resource,
+                    resource,
                     queue_pre_install_action,
                     queue_post_install_action,
                     queue_lt_security_groups,
-                    queue_placement_group,
+                    self._get_placement_group_for_compute_resource(queue, managed_placement_groups, resource),
                     instance_profiles,
                 )
-                compute_launch_templates[queue.name][compute_resource.name] = launch_template
         return compute_launch_templates
 
     def _add_compute_resource_launch_template(
@@ -1414,7 +1406,7 @@ def _add_compute_resource_launch_template(
         queue_pre_install_action,
         queue_post_install_action,
         queue_lt_security_groups,
-        queue_placement_group,
+        placement_group,
         instance_profiles,
     ):
         # LT network interfaces
@@ -1479,7 +1471,7 @@ def _add_compute_resource_launch_template(
                 ),
                 # key_name=,
                 network_interfaces=compute_lt_nw_interfaces,
-                placement=ec2.CfnLaunchTemplate.PlacementProperty(group_name=queue_placement_group),
+                placement=ec2.CfnLaunchTemplate.PlacementProperty(group_name=placement_group),
                 image_id=self._config.image_dict[queue.name],
                 iam_instance_profile=ec2.CfnLaunchTemplate.IamInstanceProfileProperty(
                     name=instance_profiles[queue.name]