def _check_file_storage()

in cli/src/pcluster/validators/cluster_validators.py [0:0]

• 43 lines of code
• 13 McCabe index (conditional complexity)

    def _check_file_storage(self, security_groups_by_nodes, file_storages, subnet_ids):
        vpc_id = AWSApi.instance().ec2.get_subnet_vpc(subnet_ids[0])
        network_interfaces_data = self._describe_network_interfaces(file_storages)
        for file_storage in file_storages:
            # Check to see if fs is in the same VPC as the stack
            file_storage_id = file_storage.file_system_id if file_storage.file_system_id else file_storage.file_cache_id
            if file_storage.vpc_id != vpc_id:
                self._add_failure(
                    "Currently only support using FSx file storage that is in the same VPC as the cluster. "
                    f"The file system {file_storage_id} is in {file_storage.vpc_id}.",
                    FailureLevel.ERROR,
                )

            # If there is an existing mt in the az, check the inbound and outbound rules of the security groups
            network_interface_ids = file_storage.network_interface_ids
            if not network_interface_ids:
                self._add_failure(
                    f"Unable to validate FSx security groups. The given FSx file storage '{file_storage_id}'"
                    " doesn't have Elastic Network Interfaces attached to it.",
                    FailureLevel.ERROR,
                )
            else:
                network_interface_responses = []
                for network_interface_id in network_interface_ids:
                    network_interface_responses.append(network_interfaces_data[network_interface_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():
                    missing_ports = self._get_missing_ports(
                        security_groups_by_nodes,
                        subnet_ids,
                        network_interfaces,
                        ports,
                        protocol,
                        file_storage.file_storage_type,
                    )

                    if missing_ports:
                        direction = "inbound and outbound"
                        if file_storage.file_storage_type == "OPENZFS":
                            direction = "inbound"
                        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"Missing ports: {missing_ports}",
                            FailureLevel.ERROR,
                        )