Config arguments not being passed with correct types (#159)

When trying to update an existing cluster the head not root volume size is
being flagged becuase it is str(200), not int(200).
It turns out that when passing the config to the CloudFormation CustomResource
for the cluster it isn't preserving the integer types and is passing their
values as strings.

Pass the configuration to the CustomResource as a json string and parse the json
string back into a Python structure so that int and string types are correctly
preserved.

Also combine the custom resources for the S3 config objects and the
ParallelCluster so that a change to the json config string will be
assured to also trigger an updates of the cluster.

Move the customer resource lambda to create_parallel_cluster_lambdas()
where it belongs.

Save the config as json and yaml in a fixed s3 key in the assets bucket.

Delete CreateParallelClusterConfig lambda.
Functionality is move the CreateParallelClusterConfig.

Updates weren't happening if the fleet was running.
Added checking for this condition and stop the fleet and retry the update
when necessary.

Resolves #158

Author: cartalla

source/cdk/cdk_slurm_stack.py

@@ -115,6 +115,8 @@ def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
 
         self.check_config()
 
+        self.cluster_region = self.config['Region']
+
         if self.use_parallel_cluster:
             self.create_vpc()
 
@@ -1096,6 +1098,58 @@ def create_lambdas(self):
     def create_parallel_cluster_lambdas(self):
         self.create_callSlurmRestApiLambda()
 
+        self.parallel_cluster_lambda_layer = aws_lambda.LayerVersion(self, "ParallelClusterLambdaLayer",
+            description = 'ParallelCluster Layer',
+            code = aws_lambda.Code.from_bucket(
+                s3.Bucket.from_bucket_name(self, 'ParallelClusterBucket', f"{self.cluster_region}-aws-parallelcluster"),
+                f"parallelcluster/{self.config['slurm']['ParallelClusterConfig']['Version']}/layers/aws-parallelcluster/lambda-layer.zip"
+            ),
+            compatible_architectures = [
+                aws_lambda.Architecture.ARM_64,
+                aws_lambda.Architecture.X86_64,
+            ],
+            compatible_runtimes = [
+                aws_lambda.Runtime.PYTHON_3_9,
+                aws_lambda.Runtime.PYTHON_3_10,
+                # aws_lambda.Runtime.PYTHON_3_11, # Doesn't work: No module named 'rpds.rpds'
+            ],
+        )
+
+        createParallelClusterLambdaAsset = s3_assets.Asset(self, "CreateParallelClusterAsset", path="resources/lambdas/CreateParallelCluster")
+        self.create_parallel_cluster_lambda = aws_lambda.Function(
+            self, "CreateParallelClusterLambda",
+            function_name=f"{self.stack_name}-CreateParallelCluster",
+            description="Create ParallelCluster from json string",
+            memory_size=2048,
+            runtime=aws_lambda.Runtime.PYTHON_3_9,
+            architecture=aws_lambda.Architecture.X86_64,
+            timeout=Duration.minutes(15),
+            log_retention=logs.RetentionDays.INFINITE,
+            handler="CreateParallelCluster.lambda_handler",
+            code=aws_lambda.Code.from_bucket(createParallelClusterLambdaAsset.bucket, createParallelClusterLambdaAsset.s3_object_key),
+            layers=[self.parallel_cluster_lambda_layer],
+            vpc = self.vpc,
+            allow_all_outbound = True
+        )
+        self.create_parallel_cluster_lambda.add_to_role_policy(
+            statement=iam.PolicyStatement(
+                effect=iam.Effect.ALLOW,
+                actions=[
+                    's3:*',
+                ],
+                resources=['*']
+                )
+            )
+        self.create_parallel_cluster_lambda.add_to_role_policy(
+            statement=iam.PolicyStatement(
+                effect=iam.Effect.ALLOW,
+                actions=[
+                    '*',
+                ],
+                resources=['*']
+                )
+            )
+
     def create_callSlurmRestApiLambda(self):
         callSlurmRestApiLambdaAsset = s3_assets.Asset(self, "CallSlurmRestApiLambdaAsset", path="resources/lambdas/CallSlurmRestApi")
         self.call_slurm_rest_api_lambda = aws_lambda.Function(
@@ -4143,102 +4197,29 @@ def create_parallel_cluster_config(self):
                 }
             self.parallel_cluster_config['SharedStorage'].append(parallel_cluster_storage_dict)
 
-        fh = NamedTemporaryFile()
-        yaml.dump(self.parallel_cluster_config, fh, encoding='utf-8', sort_keys=False)
-        self.parallelClusterConfigAsset = s3_assets.Asset(self, "ParallelClusterConfigAsset", path=fh.name)
-        self.parallelClusterConfigAsset.grant_read(self.parallel_cluster_asset_read_policy)
-
-        createParallelClusterConfigAsset = s3_assets.Asset(self, "CreateParallelClusterConfigAsset", path="resources/lambdas/CreateParallelClusterConfig")
-        self.create_parallel_cluster_config_lambda = aws_lambda.Function(
-            self, "CreateParallelClusterConfigLambda",
-            function_name=f"{self.stack_name}-CreateParallelClusterConfig",
-            description="Create ParallelCluster config file in s3",
-            memory_size=128,
-            runtime=aws_lambda.Runtime.PYTHON_3_8,
-            architecture=aws_lambda.Architecture.ARM_64,
-            timeout=Duration.minutes(3),
-            log_retention=logs.RetentionDays.INFINITE,
-            handler="CreateParallelClusterConfig.lambda_handler",
-            code=aws_lambda.Code.from_bucket(createParallelClusterConfigAsset.bucket, createParallelClusterConfigAsset.s3_object_key),
-            vpc = self.vpc,
-            allow_all_outbound = True
-        )
-        self.create_parallel_cluster_config_lambda.add_to_role_policy(
-            statement=iam.PolicyStatement(
-                effect=iam.Effect.ALLOW,
-                actions=[
-                    's3:*',
-                ],
-                resources=['*']
-                )
-            )
-        # This is created as a custom resource so that resources get resolved in the config string before it gets passed to the lambda.
-        self.parallel_cluster_config_s3_url = CustomResource(
-            self, "ParallelClusterConfigS3Object",
-            service_token = self.create_parallel_cluster_config_lambda.function_arn,
-            properties = {
-                'ParallelClusterConfigYaml': yaml.dump(self.parallel_cluster_config, sort_keys=False),
-                'S3Bucket': self.parallelClusterConfigAsset.s3_bucket_name,
-                'S3Key': self.parallelClusterConfigAsset.s3_object_key,
-                'S3ObjectUrl': self.parallelClusterConfigAsset.s3_object_url
-            }
-        ).get_att_string('S3ObjectUrl')
+        self.parallel_cluster_config_json_s3_key = f"{self.assets_base_key}/ParallelClusterConfig.json"
+        self.parallel_cluster_config_yaml_s3_key = f"{self.assets_base_key}/ParallelClusterConfig.yml"
 
-        self.parallel_cluster_lambda_layer = aws_lambda.LayerVersion(self, "ParallelClusterLambdaLayer",
-            description = 'ParallelCluster Layer',
-            code = aws_lambda.Code.from_bucket(
-                s3.Bucket.from_bucket_name(self, 'ParallelClusterBucket', f"{self.config['Region']}-aws-parallelcluster"),
-                f"parallelcluster/{self.config['slurm']['ParallelClusterConfig']['Version']}/layers/aws-parallelcluster/lambda-layer.zip"
-            ),
-            compatible_architectures = [
-                aws_lambda.Architecture.ARM_64,
-                aws_lambda.Architecture.X86_64,
-            ],
-            compatible_runtimes = [
-                aws_lambda.Runtime.PYTHON_3_8,
-                aws_lambda.Runtime.PYTHON_3_9,
-                # aws_lambda.Runtime.PYTHON_3_10,
-                # aws_lambda.Runtime.PYTHON_3_11,
-            ],
-        )
-        createParallelClusterAsset = s3_assets.Asset(self, "CreateParallelClusterAsset", path="resources/lambdas/CreateParallelCluster")
-        self.create_parallel_cluster_lambda = aws_lambda.Function(
-            self, "CreateParallelClusterLambda",
-            function_name=f"{self.stack_name}-CreateParallelCluster",
-            description="Create ParallelCluster from config file in s3",
-            memory_size=2048,
-            runtime=aws_lambda.Runtime.PYTHON_3_9,
-            architecture=aws_lambda.Architecture.X86_64,
-            timeout=Duration.minutes(15),
-            log_retention=logs.RetentionDays.INFINITE,
-            handler="CreateParallelCluster.lambda_handler",
-            code=aws_lambda.Code.from_bucket(createParallelClusterAsset.bucket, createParallelClusterAsset.s3_object_key),
-            layers=[self.parallel_cluster_lambda_layer],
-            vpc = self.vpc,
-            allow_all_outbound = True
-        )
-        self.create_parallel_cluster_lambda.add_to_role_policy(
-            statement=iam.PolicyStatement(
-                effect=iam.Effect.ALLOW,
-                actions=[
-                    '*',
-                ],
-                resources=['*']
-                )
-            )
         self.parallel_cluster = CustomResource(
-            self, "CreatedParallelCluster",
+            self, "ParallelCluster",
             service_token = self.create_parallel_cluster_lambda.function_arn,
             properties = {
+                'ParallelClusterConfigJson': json.dumps(self.parallel_cluster_config, sort_keys=False),
+                'ParallelClusterConfigS3Bucket': self.assets_bucket,
+                'ParallelClusterConfigJsonS3Key': self.parallel_cluster_config_json_s3_key,
+                'ParallelClusterConfigYamlS3Key': self.parallel_cluster_config_yaml_s3_key,
                 'Region': self.config['Region'],
                 'ClusterName': self.config['slurm']['ClusterName'],
-                'ParallelClusterConfigJson': self.parallel_cluster_config,
-                'ParallelClusterConfigS3Url': self.parallel_cluster_config_s3_url,
             }
         )
+        self.parallel_cluster_config_json_s3_url = self.parallel_cluster.get_att_string('ConfigJsonS3Url')
+        self.parallel_cluster_config_yaml_s3_url = self.parallel_cluster.get_att_string('ConfigYamlS3Url')
 
-        CfnOutput(self, "ParallelClusterConfigS3Url",
-            value = self.parallelClusterConfigAsset.s3_object_url
+        CfnOutput(self, "ParallelClusterConfigJsonS3Url",
+            value = self.parallel_cluster_config_json_s3_url
+        )
+        CfnOutput(self, "ParallelClusterConfigYamlS3Url",
+            value = self.parallel_cluster_config_yaml_s3_url
         )
         CfnOutput(self, "MungeParameterName",
             value = self.munge_key_ssm_parameter.parameter_name

source/resources/lambdas/CreateParallelCluster/CreateParallelCluster.py

@@ -17,13 +17,17 @@
 """
 
 '''
-Create/update/delete ParallelCluster cluster.
+Create/update/delete ParallelCluster cluster and save config to S3 as json and yaml.
 '''
+import boto3
 import cfnresponse
 import json
 import logging
 from os import environ as environ
 import pcluster.lib as pc
+from pcluster.api.errors import BadRequestException, UpdateClusterBadRequestException
+from time import sleep
+import yaml
 
 logger=logging.getLogger(__file__)
 logger_formatter = logging.Formatter('%(levelname)s: %(message)s')
@@ -40,10 +44,12 @@ def lambda_handler(event, context):
         requestType = event['RequestType']
         properties = event['ResourceProperties']
         required_properties = [
-            'Region',
-            'ClusterName',
             'ParallelClusterConfigJson',
-            'ParallelClusterConfigS3Url'
+            'ParallelClusterConfigS3Bucket',
+            'ParallelClusterConfigJsonS3Key',
+            'ParallelClusterConfigYamlS3Key',
+            'Region',
+            'ClusterName'
             ]
         error_message = ""
         for property in required_properties:
@@ -59,6 +65,42 @@ def lambda_handler(event, context):
             else:
                 raise KeyError(error_message)
 
+        s3_resource = boto3.resource('s3')
+
+        json_key = f"{properties['ParallelClusterConfigJsonS3Key']}"
+        json_s3_url = f"s3://{properties['ParallelClusterConfigS3Bucket']}/{json_key}"
+        json_config_object = s3_resource.Object(
+            bucket_name = properties['ParallelClusterConfigS3Bucket'],
+            key = json_key
+        )
+
+        if not properties['ParallelClusterConfigJson']:
+            # Read the config from S3 for debug so don't need a complex test event
+            logger.info(f"Reading ParallelClusterConfigJson from {json_s3_url}")
+            properties['ParallelClusterConfigJson'] = json_config_object.get()['Body'].read().decode('utf-8')
+
+        parallel_cluster_config = json.loads(properties['ParallelClusterConfigJson'])
+
+        if requestType == 'Delete':
+            logging.info(f"Deleting Parallel Cluster json config in {json_s3_url}")
+            json_config_object.delete()
+        else:
+            logging.info(f"Saving Parallel Cluster json config in {json_s3_url}")
+            json_config_object.put(Body=json.dumps(parallel_cluster_config, indent=4, sort_keys=False))
+
+        yaml_key = f"{properties['ParallelClusterConfigYamlS3Key']}"
+        yaml_s3_url = f"s3://{properties['ParallelClusterConfigS3Bucket']}/{yaml_key}"
+        yaml_config_object = s3_resource.Object(
+            bucket_name = properties['ParallelClusterConfigS3Bucket'],
+            key = yaml_key
+        )
+        if requestType == 'Delete':
+            logging.info(f"Deleting Parallel Cluster yaml config in {yaml_s3_url}")
+            yaml_config_object.delete()
+        else:
+            logging.info(f"Saving Parallel Cluster yaml config in {yaml_s3_url}")
+            yaml_config_object.put(Body=yaml.dump(parallel_cluster_config, sort_keys=False))
+
         cluster_name = properties['ClusterName']
         cluster_region = properties['Region']
         logger.info(f"{requestType} request for {cluster_name} in {cluster_region}")
@@ -94,24 +136,84 @@ def lambda_handler(event, context):
             try:
                 pc.create_cluster(
                     cluster_name = properties['ClusterName'],
-                    cluster_configuration = properties['ParallelClusterConfigJson'],
+                    cluster_configuration = parallel_cluster_config,
                     region = properties['Region'],
                     rollback_on_failure = False,
                 )
                 logger.info("Create call succeeded.")
             except:
                 logger.exception("ParallelCluster create failed. Ignoring exception")
         elif requestType == "Update":
+            logger.info("Checking compute fleet status.")
+            compute_fleet_status = pc.describe_compute_fleet(
+                cluster_name = properties['ClusterName'],
+                region = properties['Region'])['status']
+            logger.info(f"compute fleet status: {compute_fleet_status}")
+
             logger.info(f"Updating {properties['ClusterName']}")
+            stop_and_retry = False
             try:
                 pc.update_cluster(
                     cluster_name = properties['ClusterName'],
-                    cluster_configuration = properties['ParallelClusterConfigJson'],
+                    cluster_configuration = parallel_cluster_config,
                     region = properties['Region']
                 )
                 logger.info("Update call succeeded")
-            except:
-                logger.exception("ParallelCluster Update failed. Ignoring exception.")
+            except BadRequestException as e:
+                message = e.content.message
+                if 'No changes found in your cluster configuration' in message:
+                    logger.info('No changes found in your cluster configuration.')
+                else:
+                    logger.error(message)
+
+            except UpdateClusterBadRequestException as e:
+                message = e.content.message
+                logger.info(message)
+                logger.info(f"{e.content.__dict__}")
+                if 'All compute nodes must be stopped' in str(e.content.__dict__):
+                    stop_and_retry = True
+                else:
+                    logger.error(f"{message}")
+
+            if stop_and_retry:
+                logger.info(f"Stopping the cluster and retrying the update.")
+                try:
+                    pc.update_compute_fleet(
+                        cluster_name = properties['ClusterName'],
+                        status = 'STOP_REQUESTED',
+                        region = properties['Region']
+                    )
+                except BadRequestException as e:
+                    message = e.content.message
+                    logger.error(e.content)
+                    logger.error(e.content.message)
+                    raise
+                except Exception as e:
+                    logger.exception("update_compute_fleet failed")
+                    logger.error(f"{type(e)} {e.__dict__}")
+                    raise
+                logger.info(f"Stop requested. Waiting for cluster to be STOPPED.")
+                while compute_fleet_status != 'STOPPED':
+                    compute_fleet_status = pc.describe_compute_fleet(
+                        cluster_name = properties['ClusterName'],
+                        region = properties['Region'])['status']
+                    logger.info(f"compute fleet status: {compute_fleet_status}")
+                    sleep(1)
+                logger.info("Compute fleet is stopped. Retrying update.")
+                try:
+                    pc.update_cluster(
+                        cluster_name = properties['ClusterName'],
+                        cluster_configuration = parallel_cluster_config,
+                        region = properties['Region']
+                    )
+                    logger.info("Update call succeeded")
+                except (BadRequestException, UpdateClusterBadRequestException) as e:
+                    message = e.content.message
+                    logger.error(message)
+                    logger.error(f"{e.content.__dict__}")
+                except Exception as e:
+                    logger.exception("ParallelCluster Update failed.")
+
         elif requestType == 'Delete':
             logger.info(f"Deleting {properties['ClusterName']}")
             try:
@@ -130,4 +232,4 @@ def lambda_handler(event, context):
         cfnresponse.send(event, context, cfnresponse.FAILED, {'error': str(e)}, physicalResourceId=cluster_name)
         raise
 
-    cfnresponse.send(event, context, cfnresponse.SUCCESS, {}, physicalResourceId=cluster_name)
+    cfnresponse.send(event, context, cfnresponse.SUCCESS, {'ConfigJsonS3Url': json_s3_url, 'ConfigYamlS3Url': yaml_s3_url}, physicalResourceId=cluster_name)