feat(redshift): optionally reboot Clusters to apply parameter changes (#22063)

Closes #22009

Currently waiting on #22055  and #22059 for the assertions in the integration test to successfully run

----

### All Submissions:

* [X] Have you followed the guidelines in our [Contributing guide?](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md)

### Adding new Unconventional Dependencies:

* [ ] This PR adds new unconventional dependencies following the process described [here](https://github.com/aws/aws-cdk/blob/main/CONTRIBUTING.md/#adding-new-unconventional-dependencies)

### New Features

* [X] Have you added the new feature to an [integration test](https://github.com/aws/aws-cdk/blob/main/INTEGRATION_TESTS.md)?
	* [X] Did you use `yarn integ` to deploy the infrastructure and generate the snapshot (i.e. `yarn integ` without `--dry-run`)?

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: dontirun

Diff for: packages/@aws-cdk/aws-redshift/README.md

@@ -349,6 +349,25 @@ const cluster = new Cluster(this, 'Cluster', {
 cluster.addToParameterGroup('enable_user_activity_logging', 'true');
 ```
 
+## Rebooting for Parameter Updates
+
+In most cases, existing clusters [must be manually rebooted](https://docs.aws.amazon.com/redshift/latest/mgmt/working-with-parameter-groups.html) to apply parameter changes. You can automate parameter related reboots by setting the cluster's `rebootForParameterChanges` property to `true` , or by using `Cluster.enableRebootForParameterChanges()`.
+
+```ts
+declare const vpc: ec2.Vpc;
+
+const cluster = new Cluster(this, 'Cluster', {
+  masterUser: {
+    masterUsername: 'admin',
+    masterPassword: cdk.SecretValue.unsafePlainText('tooshort'),
+  },
+  vpc,
+});
+
+cluster.addToParameterGroup('enable_user_activity_logging', 'true');
+cluster.enableRebootForParameterChanges()
+```
+
 ## Elastic IP
 
 If you configure your cluster to be publicly accessible, you can optionally select an *elastic IP address* to use for the external IP address. An elastic IP address is a static IP address that is associated with your AWS account. You can use an elastic IP address to connect to your cluster from outside the VPC. An elastic IP address gives you the ability to change your underlying configuration without affecting the IP address that clients use to connect to your cluster. This approach can be helpful for situations such as recovery after a failure.

Diff for: packages/@aws-cdk/aws-redshift/lib/cluster-parameter-change-reboot-handler/index.ts

@@ -0,0 +1,53 @@
+// eslint-disable-next-line import/no-extraneous-dependencies
+import { Redshift } from 'aws-sdk';
+
+const redshift = new Redshift();
+
+export async function handler(event: AWSLambda.CloudFormationCustomResourceEvent): Promise<void> {
+  if (event.RequestType !== 'Delete') {
+    return rebootClusterIfRequired(event.ResourceProperties?.ClusterId, event.ResourceProperties?.ParameterGroupName);
+  } else {
+    return;
+  }
+}
+
+async function rebootClusterIfRequired(clusterId: string, parameterGroupName: string): Promise<void> {
+  return executeActionForStatus(await getApplyStatus());
+
+  // https://docs.aws.amazon.com/redshift/latest/APIReference/API_ClusterParameterStatus.html
+  async function executeActionForStatus(status: string, retryDurationMs?: number): Promise<void> {
+    await sleep(retryDurationMs ?? 0);
+    if (['pending-reboot', 'apply-deferred', 'apply-error'].includes(status)) {
+      try {
+        await redshift.rebootCluster({ ClusterIdentifier: clusterId }).promise();
+      } catch (err) {
+        if ((err as any).code === 'InvalidClusterState') {
+          return await executeActionForStatus(status, 30000);
+        } else {
+          throw err;
+        }
+      }
+      return;
+    } else if (['applying', 'retry'].includes(status)) {
+      return executeActionForStatus(await getApplyStatus(), 30000);
+    }
+    return;
+  }
+
+  async function getApplyStatus(): Promise<string> {
+    const clusterDetails = await redshift.describeClusters({ ClusterIdentifier: clusterId }).promise();
+    if (clusterDetails.Clusters?.[0].ClusterParameterGroups === undefined) {
+      throw new Error(`Unable to find any Parameter Groups associated with ClusterId "${clusterId}".`);
+    }
+    for (const group of clusterDetails.Clusters?.[0].ClusterParameterGroups) {
+      if (group.ParameterGroupName === parameterGroupName) {
+        return group.ParameterApplyStatus ?? 'retry';
+      }
+    }
+    throw new Error(`Unable to find Parameter Group named "${parameterGroupName}" associated with ClusterId "${clusterId}".`);
+  }
+}
+
+function sleep(ms: number) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}

Diff for: packages/@aws-cdk/aws-redshift/lib/cluster.ts

@@ -1,17 +1,18 @@
+import * as path from 'path';
 import * as ec2 from '@aws-cdk/aws-ec2';
 import * as iam from '@aws-cdk/aws-iam';
 import * as kms from '@aws-cdk/aws-kms';
+import * as lambda from '@aws-cdk/aws-lambda';
 import * as s3 from '@aws-cdk/aws-s3';
 import * as secretsmanager from '@aws-cdk/aws-secretsmanager';
-import { Duration, IResource, Lazy, RemovalPolicy, Resource, SecretValue, Token } from '@aws-cdk/core';
-import { AwsCustomResource, AwsCustomResourcePolicy, PhysicalResourceId } from '@aws-cdk/custom-resources';
+import { ArnFormat, CustomResource, Duration, IResource, Lazy, RemovalPolicy, Resource, SecretValue, Stack, Token } from '@aws-cdk/core';
+import { AwsCustomResource, AwsCustomResourcePolicy, PhysicalResourceId, Provider } from '@aws-cdk/custom-resources';
 import { Construct } from 'constructs';
 import { DatabaseSecret } from './database-secret';
 import { Endpoint } from './endpoint';
 import { ClusterParameterGroup, IClusterParameterGroup } from './parameter-group';
 import { CfnCluster } from './redshift.generated';
 import { ClusterSubnetGroup, IClusterSubnetGroup } from './subnet-group';
-
 /**
  * Possible Node Types to use in the cluster
  * used for defining `ClusterProps.nodeType`.
@@ -364,6 +365,12 @@ export interface ClusterProps {
    */
   readonly elasticIp?: string
 
+  /**
+   * If this flag is set, the cluster will be rebooted when changes to the cluster's parameter group that require a restart to apply.
+   * @default false
+   */
+  readonly rebootForParameterChanges?: boolean
+
   /**
    * If this flag is set, Amazon Redshift forces all COPY and UNLOAD traffic between your cluster and your data repositories through your virtual private cloud (VPC).
    *
@@ -592,7 +599,9 @@ export class Cluster extends ClusterBase {
 
     const defaultPort = ec2.Port.tcp(this.clusterEndpoint.port);
     this.connections = new ec2.Connections({ securityGroups, defaultPort });
-
+    if (props.rebootForParameterChanges) {
+      this.enableRebootForParameterChanges();
+    }
     // Add default role if specified and also available in the roles list
     if (props.defaultRole) {
       if (props.roles?.some(x => x === props.defaultRole)) {
@@ -689,6 +698,71 @@ export class Cluster extends ClusterBase {
     }
   }
 
+  /**
+   * Enables automatic cluster rebooting when changes to the cluster's parameter group require a restart to apply.
+   */
+  public enableRebootForParameterChanges(): void {
+    if (this.node.tryFindChild('RedshiftClusterRebooterCustomResource')) {
+      return;
+    }
+    const rebootFunction = new lambda.SingletonFunction(this, 'RedshiftClusterRebooterFunction', {
+      uuid: '511e207f-13df-4b8b-b632-c32b30b65ac2',
+      runtime: lambda.Runtime.NODEJS_16_X,
+      code: lambda.Code.fromAsset(path.join(__dirname, 'cluster-parameter-change-reboot-handler')),
+      handler: 'index.handler',
+      timeout: Duration.seconds(900),
+    });
+    rebootFunction.addToRolePolicy(new iam.PolicyStatement({
+      actions: ['redshift:DescribeClusters'],
+      resources: ['*'],
+    }));
+    rebootFunction.addToRolePolicy(new iam.PolicyStatement({
+      actions: ['redshift:RebootCluster'],
+      resources: [
+        Stack.of(this).formatArn({
+          service: 'redshift',
+          resource: 'cluster',
+          resourceName: this.clusterName,
+          arnFormat: ArnFormat.COLON_RESOURCE_NAME,
+        }),
+      ],
+    }));
+    const provider = new Provider(this, 'ResourceProvider', {
+      onEventHandler: rebootFunction,
+    });
+    const customResource = new CustomResource(this, 'RedshiftClusterRebooterCustomResource', {
+      resourceType: 'Custom::RedshiftClusterRebooter',
+      serviceToken: provider.serviceToken,
+      properties: {
+        ClusterId: this.clusterName,
+        ParameterGroupName: Lazy.string({
+          produce: () => {
+            if (!this.parameterGroup) {
+              throw new Error('Cannot enable reboot for parameter changes when there is no associated ClusterParameterGroup.');
+            }
+            return this.parameterGroup.clusterParameterGroupName;
+          },
+        }),
+        ParametersString: Lazy.string({
+          produce: () => {
+            if (!(this.parameterGroup instanceof ClusterParameterGroup)) {
+              throw new Error('Cannot enable reboot for parameter changes when using an imported parameter group.');
+            }
+            return JSON.stringify(this.parameterGroup.parameters);
+          },
+        }),
+      },
+    });
+    Lazy.any({
+      produce: () => {
+        if (!this.parameterGroup) {
+          throw new Error('Cannot enable reboot for parameter changes when there is no associated ClusterParameterGroup.');
+        }
+        customResource.node.addDependency(this, this.parameterGroup);
+      },
+    });
+  }
+
   /**
    * Adds default IAM role to cluster. The default IAM role must be already associated to the cluster to be added as the default role.
    *

Diff for: packages/@aws-cdk/aws-redshift/test/cluster-reboot.integ.snapshot/asset.1b88b7c3e3e0f8d3e27ded1bde51b7a80c75f3d8733872af7952c3a6d902147e/index.d.ts

@@ -0,0 +1 @@
+export declare function handler(event: AWSLambda.CloudFormationCustomResourceEvent): Promise<void>;

Diff for: packages/@aws-cdk/aws-redshift/test/cluster-reboot.integ.snapshot/asset.1b88b7c3e3e0f8d3e27ded1bde51b7a80c75f3d8733872af7952c3a6d902147e/index.js

@@ -0,0 +1,53 @@
+// eslint-disable-next-line import/no-extraneous-dependencies
+import { Redshift } from 'aws-sdk';
+
+const redshift = new Redshift();
+
+export async function handler(event: AWSLambda.CloudFormationCustomResourceEvent): Promise<void> {
+  if (event.RequestType !== 'Delete') {
+    return rebootClusterIfRequired(event.ResourceProperties?.ClusterId, event.ResourceProperties?.ParameterGroupName);
+  } else {
+    return;
+  }
+}
+
+async function rebootClusterIfRequired(clusterId: string, parameterGroupName: string): Promise<void> {
+  return executeActionForStatus(await getApplyStatus());
+
+  // https://docs.aws.amazon.com/redshift/latest/APIReference/API_ClusterParameterStatus.html
+  async function executeActionForStatus(status: string, retryDurationMs?: number): Promise<void> {
+    await sleep(retryDurationMs ?? 0);
+    if (['pending-reboot', 'apply-deferred', 'apply-error', 'unknown-error'].includes(status)) {
+      try {
+        await redshift.rebootCluster({ ClusterIdentifier: clusterId }).promise();
+      } catch (err) {
+        if ((<any>err).code === 'InvalidClusterState') {
+          return await executeActionForStatus(status, 30000);
+        } else {
+          throw err;
+        }
+      }
+      return;
+    } else if (['applying', 'retry'].includes(status)) {
+      return executeActionForStatus(await getApplyStatus(), 30000);
+    }
+    return;
+  }
+
+  async function getApplyStatus(): Promise<string> {
+    const clusterDetails = await redshift.describeClusters({ ClusterIdentifier: clusterId }).promise();
+    if (clusterDetails.Clusters?.[0].ClusterParameterGroups === undefined) {
+      throw new Error(`Unable to find any Parameter Groups associated with ClusterId "${clusterId}".`);
+    }
+    for (const group of clusterDetails.Clusters?.[0].ClusterParameterGroups) {
+      if (group.ParameterGroupName === parameterGroupName) {
+        return group.ParameterApplyStatus ?? 'retry';
+      }
+    }
+    throw new Error(`Unable to find Parameter Group named "${parameterGroupName}" associated with ClusterId "${clusterId}".`);
+  }
+}
+
+function sleep(ms: number) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+}