Merge branch 'develop' into develop

Author: hgreebe

CHANGELOG.md

@@ -8,9 +8,14 @@ CHANGELOG
 - Add validator that warns against the downsides of disabling in-place updates on compute and login nodes through DevSettings.
 - Upgrade jmespath to ~=1.0 (from ~=0.10).
 - Upgrade tabulate to <=0.9.0 (from <=0.8.10).
+- Scope down storage security group ingress rules to specific ports for all node types.
 
 **BUG FIXES**
 - Add validation to block updates that change tag order. Blocking such change prevents update failures.
+- Fix LoginNodes NLB not being publicly accessible when using public subnet with implicit main route table association. 
+  See https://github.com/aws/aws-parallelcluster/issues/7173
+- Fix cluster creation failure without Internet access when GPU instances and DCV are used.
+- Fix intermittent cluster creation failure caused by eventual consistency issues when head, compute and login nodes have the same security group.
 
 3.14.1
 ------

cli/src/pcluster/cli/commands/ssh.py

@@ -79,15 +79,13 @@ class SshCommand(CliCommand):
         "Run ssh command with the cluster username and IP address pre-populated. "
         "Arbitrary arguments are appended to the end of the ssh command."
     )
-    epilog = textwrap.dedent(
-        """Example:
+    epilog = textwrap.dedent("""Example:
 
   pcluster ssh --cluster-name mycluster -i ~/.ssh/id_rsa
 
 Returns an ssh command with the cluster username and IP address pre-populated:
 
-  ssh ec2-user@1.1.1.1 -i ~/.ssh/id_rsa"""
-    )
+  ssh ec2-user@1.1.1.1 -i ~/.ssh/id_rsa""")
 
     def __init__(self, subparsers):
         super().__init__(

cli/src/pcluster/config/update_policy.py

@@ -20,6 +20,7 @@
     SLURM,
     STORAGE_TYPES_SUPPORTING_LIVE_UPDATES,
 )
+from pcluster.utils import get_dictionary_diff, parse_json_string
 
 
 class UpdatePolicy:
@@ -510,6 +511,38 @@ def get_pool_name_from_change_paths(change):
     return ""
 
 
+# ExtraChefAttributes update policy helpers
+# Define which paths within ExtraChefAttributes JSON are updatable (dot-notation)
+# IMPORTANT: We assume this set to contain the path to leaf fields.
+EXTRA_CHEF_ATTRIBUTES_UPDATABLE_PATHS: set[str] = {
+    "cluster.slurm.reconfigure_timeout",
+}
+
+
+def _get_non_updatable_changes(old_value: str, new_value: str) -> list[str]:
+    """Parse and compare two ExtraChefAttributes JSON strings, returning paths that are not updatable."""
+    old_attrs = parse_json_string(old_value, raise_on_error=False, default={}) if old_value else {}
+    new_attrs = parse_json_string(new_value, raise_on_error=False, default={}) if new_value else {}
+
+    return [p for p in get_dictionary_diff(old_attrs, new_attrs) if p not in EXTRA_CHEF_ATTRIBUTES_UPDATABLE_PATHS]
+
+
+def condition_checker_extra_chef_attributes(change, _) -> bool:
+    """
+    Check if ExtraChefAttributes changes are allowed.
+
+    Only changes to paths defined in EXTRA_CHEF_ATTRIBUTES_UPDATABLE_PATHS are allowed.
+    """
+    return len(_get_non_updatable_changes(change.old_value, change.new_value)) == 0
+
+
+def fail_reason_extra_chef_attributes(change, _) -> str:
+    """Generate fail reason for ExtraChefAttributes update."""
+    non_updatable_changes = _get_non_updatable_changes(change.old_value, change.new_value)
+    paths_str = ", ".join(sorted(non_updatable_changes))
+    return f"The following ExtraChefAttributes fields cannot be updated: {paths_str}"
+
+
 # Common fail_reason messages
 UpdatePolicy.FAIL_REASONS = {
     "ebs_volume_resize": "Updating the file system after a resize operation requires commands specific to your "
@@ -526,6 +559,7 @@ def get_pool_name_from_change_paths(change):
     "compute_or_login_nodes_running": lambda change, patch: (
         "The update is not supported when compute or login nodes are running"
     ),
+    "extra_chef_attributes_update": fail_reason_extra_chef_attributes,
 }
 
 # Common action_needed messages
@@ -548,6 +582,9 @@ def get_pool_name_from_change_paths(change):
         "Stop the login nodes by setting Count parameter to 0 "
         "and update the cluster with the pcluster update-cluster command"
     ),
+    "extra_chef_attributes_update": lambda change, patch: (
+        "Revert the non-updatable ExtraChefAttributes fields to their original values."
+    ),
 }
 
 # Base policies
@@ -567,6 +604,15 @@ def get_pool_name_from_change_paths(change):
     name="SUPPORTED", level=0, fail_reason="-", condition_checker=(lambda change, patch: True)
 )
 
+# Update policy for ExtraChefAttributes - allows updates to specific fields only
+UpdatePolicy.EXTRA_CHEF_ATTRIBUTES = UpdatePolicy(
+    name="EXTRA_CHEF_ATTRIBUTES",
+    level=5,
+    fail_reason=UpdatePolicy.FAIL_REASONS["extra_chef_attributes_update"],
+    action_needed=UpdatePolicy.ACTIONS_NEEDED["extra_chef_attributes_update"],
+    condition_checker=condition_checker_extra_chef_attributes,
+)
+
 # Checks resize of max_vcpus in Batch Compute Environment
 UpdatePolicy.AWSBATCH_CE_MAX_RESIZE = UpdatePolicy(
     name="AWSBATCH_CE_MAX_RESIZE",

cli/src/pcluster/constants.py

@@ -127,7 +127,8 @@
 
 FSX_PORTS = {
     # Lustre Security group: https://docs.aws.amazon.com/fsx/latest/LustreGuide/limit-access-security-groups.html
-    LUSTRE: {"tcp": [988]},
+    # Among the Lustre ports, only 988 is mandatory to provide basic features.
+    LUSTRE: {"tcp": [988, (1018, 1023)]},
     # OpenZFS Security group: https://docs.aws.amazon.com/fsx/latest/OpenZFSGuide/limit-access-security-groups.html
     OPENZFS: {"tcp": [111, 2049, 20001, 20002, 20003], "udp": [111, 2049, 20001, 20002, 20003]},
     # Ontap Security group: https://docs.aws.amazon.com/fsx/latest/ONTAPGuide/limit-access-security-groups.html

cli/src/pcluster/schemas/common_schema.py

@@ -216,13 +216,7 @@ class CookbookSchema(BaseSchema):
             )
         }
     )
-    extra_chef_attributes = fields.Str(
-        metadata={
-            "update_policy": UpdatePolicy(
-                UpdatePolicy.UNSUPPORTED, fail_reason=UpdatePolicy.FAIL_REASONS["cookbook_update"]
-            )
-        }
-    )
+    extra_chef_attributes = fields.Str(metadata={"update_policy": UpdatePolicy.EXTRA_CHEF_ATTRIBUTES})
 
     @post_load()
     def make_resource(self, data, **kwargs):
@@ -275,7 +269,7 @@ def make_resource(self, data, **kwargs):
 class BaseDevSettingsSchema(BaseSchema):
     """Represent the common schema of Dev Setting for ImageBuilder and Cluster."""
 
-    cookbook = fields.Nested(CookbookSchema, metadata={"update_policy": UpdatePolicy.UNSUPPORTED})
+    cookbook = fields.Nested(CookbookSchema, metadata={"update_policy": UpdatePolicy.IGNORED})
     node_package = fields.Str(metadata={"update_policy": UpdatePolicy.UNSUPPORTED})
     aws_batch_cli_package = fields.Str(metadata={"update_policy": UpdatePolicy.UNSUPPORTED})
 

cli/src/pcluster/templates/cluster_stack.py

@@ -795,27 +795,44 @@ def _add_storage_security_group(self, storage_cfn_id, storage):
             "Storage": [storage_security_group.ref],
         }
 
+        # Deduplicate SGs to avoid creating identical rules for the same security group.
+        # This prevents EC2 duplicate rule errors which are susceptible to eventual consistency issues.
+        seen_sgs = set()
         for sg_type, sg_refs in target_security_groups.items():
             for sg_ref_id, sg_ref in enumerate(sg_refs):
-                # TODO Scope down ingress rules to allow only traffic on the strictly necessary ports.
-                #      Currently scoped down only on Login nodes to limit blast radius.
-                ingress_protocol = "-1"
-                ingress_port = ALL_PORTS_RANGE
-                if sg_type == "Login":
-                    if storage_type == SharedStorageType.EFS:
-                        ingress_protocol = "tcp"
-                        ingress_port = EFS_PORT
-                    elif storage_type == SharedStorageType.FSX:
-                        ingress_protocol = "tcp"
-                        ingress_port = FSX_PORTS[LUSTRE]["tcp"][0]
-                ingress_rule = self._allow_all_ingress(
-                    description=f"{storage_cfn_id}SecurityGroup{sg_type}Ingress{sg_ref_id}",
-                    source_security_group_id=sg_ref,
-                    group_id=storage_security_group.ref,
-                    ip_protocol=ingress_protocol,
-                    port=ingress_port,
+                # This conversion is required because sg_ref could either be a string or a CDK token reference.
+                # A CDK token must be converted to a string to make it comparable within the set.
+                sg_key = str(sg_ref)
+                if sg_key in seen_sgs:
+                    continue
+                seen_sgs.add(sg_key)
+
+                # For Storage-to-Storage, allow all traffic.
+                # For Head/Compute/Login nodes, allow only the required ports.
+                storage_ports = {
+                    SharedStorageType.EFS: ("tcp", [EFS_PORT]),
+                    SharedStorageType.FSX: ("tcp", FSX_PORTS[LUSTRE]["tcp"]),
+                }
+                ingress_protocol, ingress_ports = (
+                    ("-1", [ALL_PORTS_RANGE])
+                    if sg_type == "Storage"
+                    else storage_ports.get(storage_type, ("-1", [ALL_PORTS_RANGE]))
                 )
-                rules.append(ingress_rule)
+
+                for rule_id, ingress_port in enumerate(ingress_ports):
+                    ingress_rule = self._allow_all_ingress(
+                        description=f"{storage_cfn_id}SecurityGroup{sg_type}Ingress{sg_ref_id}Rule{rule_id}",
+                        source_security_group_id=sg_ref,
+                        group_id=storage_security_group.ref,
+                        ip_protocol=ingress_protocol,
+                        port=ingress_port,
+                    )
+                    rules.append(ingress_rule)
+
+                    if sg_type == "Storage":
+                        ingress_rule.cfn_options.deletion_policy = ingress_rule.cfn_options.update_replace_policy = (
+                            storage_deletion_policy
+                        )
 
                 egress_rule = self._allow_all_egress(
                     description=f"{storage_cfn_id}SecurityGroup{sg_type}Egress{sg_ref_id}",
@@ -825,9 +842,6 @@ def _add_storage_security_group(self, storage_cfn_id, storage):
                 rules.append(egress_rule)
 
                 if sg_type == "Storage":
-                    ingress_rule.cfn_options.deletion_policy = ingress_rule.cfn_options.update_replace_policy = (
-                        storage_deletion_policy
-                    )
                     egress_rule.cfn_options.deletion_policy = egress_rule.cfn_options.update_replace_policy = (
                         storage_deletion_policy
                     )

cli/src/pcluster/utils.py

@@ -614,3 +614,79 @@ def get_needed_ultraserver_capacity_block_statuses(statuses, capacity_reservatio
         ):
             needed_capacity_block_statuses.append(status)
     return needed_capacity_block_statuses
+
+
+def parse_json_string(value: str, raise_on_error: bool = False, default: any = None) -> any:
+    """
+    Parse a JSON string into a Python object.
+
+    :param value: JSON string to parse
+    :param raise_on_error: If True, raises exception on parse failure; if False, returns default
+    :param default: Value to return on parse failure when raise_on_error is False
+    :return: Parsed JSON object or default value on failure
+    """
+    try:
+        return json.loads(value)
+    except (json.JSONDecodeError, TypeError) as e:
+        LOGGER.error("Failed to parse JSON string: %s. Error: %s", value, e)
+        if raise_on_error:
+            raise e
+        return default
+
+
+def _collect_leaves(d: dict, prefix: str) -> set[str]:
+    """Collect all leaf paths from a dict."""
+    leaves = set()
+    for k, v in d.items():
+        key_path = f"{prefix}.{k}" if prefix else str(k)
+        if isinstance(v, dict) and v:
+            leaves.update(_collect_leaves(v, key_path))
+        else:
+            leaves.add(key_path)
+    return leaves if leaves else {prefix} if prefix else set()
+
+
+def _collect_diff_for_key(d: dict, key: str, path: str) -> set[str]:
+    """Collect leaf paths for a key that exists only in one dictionary."""
+    key_path = f"{path}.{key}" if path else str(key)
+    if isinstance(d[key], dict) and d[key]:
+        return _collect_leaves(d[key], key_path)
+    return {key_path}
+
+
+def get_dictionary_diff(d1: dict, d2: dict, path: str = "") -> set[str]:
+    """
+    Recursively find all leaf paths where two dictionaries differ.
+
+    Returns a set of dot-notation paths (e.g., "cluster.settings.enabled") where
+    the dictionaries have different values, including added or removed keys.
+    Always returns leaf paths, never intermediate dict paths.
+
+    :param d1: First dictionary to compare (if None, treated as empty dict)
+    :param d2: Second dictionary to compare (if None, treated as empty dict)
+    :param path: Current path prefix (used for recursion)
+    :return: Set of dot-notation paths where dictionaries differ.
+             If no differences detected return empoty set.
+    """
+    d1 = d1 or {}
+    d2 = d2 or {}
+
+    diffs: set[str] = set()
+    keys1, keys2 = set(d1.keys()), set(d2.keys())
+
+    # Collect paths that are added or remove
+    for k in keys1 - keys2:
+        diffs.update(_collect_diff_for_key(d1, k, path))
+
+    for k in keys2 - keys1:
+        diffs.update(_collect_diff_for_key(d2, k, path))
+
+    # Compare paths
+    for k in keys1 & keys2:
+        key_path = f"{path}.{k}" if path else str(k)
+        if isinstance(d1[k], dict) and isinstance(d2[k], dict):
+            diffs.update(get_dictionary_diff(d1[k], d2[k], key_path))
+        elif d1[k] != d2[k]:
+            diffs.add(key_path)
+
+    return diffs

cli/src/pcluster/validators/cluster_validators.py

@@ -26,6 +26,7 @@
     DELETE_POLICY,
     EFS_PORT,
     FSX_PORTS,
+    LUSTRE,
     PCLUSTER_IMAGE_BUILD_STATUS_TAG,
     PCLUSTER_NAME_MAX_LENGTH,
     PCLUSTER_NAME_MAX_LENGTH_SLURM_ACCOUNTING,
@@ -660,11 +661,14 @@ def _check_file_storage(self, security_groups_by_nodes, file_storages, subnet_id
                 network_interfaces = [ni for ni in network_interface_responses if ni.get("VpcId") == vpc_id]
 
                 for protocol, ports in FSX_PORTS[file_storage.file_storage_type].items():
+                    # For Lustre, only validate the first port (988), which is the only mandatory one.
+                    # Other ports (1018-1023) are optional so we do not enforce them.
+                    ports_to_validate = [ports[0]] if file_storage.file_storage_type == LUSTRE else ports
                     missing_ports = self._get_missing_ports(
                         security_groups_by_nodes,
                         subnet_ids,
                         network_interfaces,
-                        ports,
+                        ports_to_validate,
                         protocol,
                         file_storage.file_storage_type,
                     )
@@ -676,7 +680,7 @@ def _check_file_storage(self, security_groups_by_nodes, file_storages, subnet_id
                         self._add_failure(
                             f"The current security group settings on file storage '{file_storage_id}' does not"
                             " satisfy mounting requirement. The file storage must be associated to a security group"
-                            f" that allows {direction} {protocol.upper()} traffic through ports {ports}. "
+                            f" that allows {direction} {protocol.upper()} traffic through ports {ports_to_validate}. "
                             f"Missing ports: {missing_ports}",
                             FailureLevel.ERROR,
                         )

cli/src/pcluster/validators/dev_settings_validators.py

@@ -15,6 +15,8 @@
 
 EXTRA_CHEF_ATTRIBUTES_PATH = "DevSettings/Cookbook/ExtraChefAttributes"
 ATTR_IN_PLACE_UPDATE_ON_FLEET_ENABLED = "in_place_update_on_fleet_enabled"
+ATTR_RECONFIGURE_TIMEOUT = "cluster.slurm.reconfigure_timeout"
+MIN_SLURM_RECONFIGURE_TIMEOUT = 300
 
 
 class ExtraChefAttributesValidator(Validator):
@@ -29,8 +31,10 @@ def _validate(self, extra_chef_attributes: str = None):
         """
         if not extra_chef_attributes:
             return
-        else:
-            self._validate_in_place_update_on_fleet_enabled(json.loads(extra_chef_attributes))
+
+        attrs = json.loads(extra_chef_attributes)
+        self._validate_in_place_update_on_fleet_enabled(attrs)
+        self._validate_slurm_reconfigure_timeout(attrs)
 
     def _validate_in_place_update_on_fleet_enabled(self, extra_chef_attributes: dict = None):
         """Validate attribute cluster.in_place_update_on_fleet_enabled.
@@ -60,3 +64,33 @@ def _validate_in_place_update_on_fleet_enabled(self, extra_chef_attributes: dict
                 "by replacing compute and login nodes according to the selected QueueUpdateStrategy.",
                 FailureLevel.WARNING,
             )
+
+    def _validate_slurm_reconfigure_timeout(self, extra_chef_attributes: dict = None):
+        """Validate attribute cluster.slurm.reconfigure-timeout.
+
+        Must be an integer greater than 300.
+
+        Args:
+            extra_chef_attributes: Dictionary of Chef attributes to validate.
+        """
+        reconfigure_timeout = dig(extra_chef_attributes, *ATTR_RECONFIGURE_TIMEOUT.split("."))
+
+        if reconfigure_timeout is None:
+            return
+
+        # Reject booleans explicitly (bool is subclass of int in Python)
+        if isinstance(reconfigure_timeout, bool) or not isinstance(reconfigure_timeout, int):
+            self._add_failure(
+                f"Invalid value in {EXTRA_CHEF_ATTRIBUTES_PATH}: "
+                f"attribute '{ATTR_RECONFIGURE_TIMEOUT}' must be an integer.",
+                FailureLevel.ERROR,
+            )
+            return
+
+        if reconfigure_timeout <= MIN_SLURM_RECONFIGURE_TIMEOUT:
+            self._add_failure(
+                f"Invalid value in {EXTRA_CHEF_ATTRIBUTES_PATH}: "
+                f"attribute '{ATTR_RECONFIGURE_TIMEOUT}' "
+                f"must be greater than {MIN_SLURM_RECONFIGURE_TIMEOUT}.",
+                FailureLevel.ERROR,
+            )