Support Cloudwatch integration for AWS::Redshift::Cluster LoggingProp…

…erties (#180) (#180)

This change provides CFN support for enabling redshift audit logging with CloudWatch. We are adding 2 new parameters "LogDestinationType" and "LogExports" to the existing schema, which are used to setup CloudWatch logging for audit logging.

Manual testing and integration testing have been performed on cloudformation with all new changes.

Co-authored-by: Jack Fei <[email protected]>

aws-redshift-cluster/aws-redshift-cluster.json

@@ -30,6 +30,17 @@
             "type": "object",
             "additionalProperties": false,
             "properties": {
+                "LogDestinationType": {
+                    "type": "string"
+                },
+                "LogExports": {
+                    "type": "array",
+                    "insertionOrder": false,
+                    "maxItems": 3,
+                    "items": {
+                        "type": "string"
+                    }
+                },
                 "BucketName": {
                     "type": "string",
                     "relationshipRef": {
@@ -143,17 +154,20 @@
         "KmsKeyId": {
             "description": "The AWS Key Management Service (KMS) key ID of the encryption key that you want to use to encrypt data in the cluster.",
             "type": "string",
-            "anyOf": [{
-                "relationshipRef": {
-                    "typeName": "AWS::KMS::Key",
-                    "propertyPath": "/properties/Arn"
-                }
-            }, {
-                "relationshipRef": {
-                    "typeName": "AWS::KMS::Key",
-                    "propertyPath": "/properties/KeyId"
+            "anyOf": [
+                {
+                    "relationshipRef": {
+                        "typeName": "AWS::KMS::Key",
+                        "propertyPath": "/properties/Arn"
+                    }
+                },
+                {
+                    "relationshipRef": {
+                        "typeName": "AWS::KMS::Key",
+                        "propertyPath": "/properties/KeyId"
+                    }
                 }
-            }]
+            ]
         },
         "NumberOfNodes": {
             "description": "The number of compute nodes in the cluster. This parameter is required when the ClusterType parameter is specified as multi-node.",
@@ -178,17 +192,20 @@
             "uniqueItems": false,
             "items": {
                 "type": "string",
-                "anyOf": [{
-                    "relationshipRef": {
-                        "typeName": "AWS::EC2::SecurityGroup",
-                        "propertyPath": "/properties/Id"
-                    }
-                }, {
-                    "relationshipRef": {
-                        "typeName": "AWS::Redshift::ClusterSecurityGroup",
-                        "propertyPath": "/properties/Id"
+                "anyOf": [
+                    {
+                        "relationshipRef": {
+                            "typeName": "AWS::EC2::SecurityGroup",
+                            "propertyPath": "/properties/Id"
+                        }
+                    },
+                    {
+                        "relationshipRef": {
+                            "typeName": "AWS::Redshift::ClusterSecurityGroup",
+                            "propertyPath": "/properties/Id"
+                        }
                     }
-                }]
+                ]
             }
         },
         "IamRoles": {
@@ -335,17 +352,20 @@
         "MasterPasswordSecretKmsKeyId": {
             "description": "The ID of the Key Management Service (KMS) key used to encrypt and store the cluster's admin user credentials secret.",
             "type": "string",
-            "anyOf": [{
-                "relationshipRef": {
-                    "typeName": "AWS::KMS::Key",
-                    "propertyPath": "/properties/Arn"
-                }
-            }, {
-                "relationshipRef": {
-                    "typeName": "AWS::KMS::Key",
-                    "propertyPath": "/properties/KeyId"
+            "anyOf": [
+                {
+                    "relationshipRef": {
+                        "typeName": "AWS::KMS::Key",
+                        "propertyPath": "/properties/Arn"
+                    }
+                },
+                {
+                    "relationshipRef": {
+                        "typeName": "AWS::KMS::Key",
+                        "propertyPath": "/properties/KeyId"
+                    }
                 }
-            }]
+            ]
         },
         "MasterPasswordSecretArn": {
             "description": "The Amazon Resource Name (ARN) for the cluster's admin user credentials secret.",

aws-redshift-cluster/docs/loggingproperties.md

@@ -8,6 +8,8 @@ To declare this entity in your AWS CloudFormation template, use the following sy
 
 <pre>
 {
+    "<a href="#logdestinationtype" title="LogDestinationType">LogDestinationType</a>" : <i>String</i>,
+    "<a href="#logexports" title="LogExports">LogExports</a>" : <i>[ String, ... ]</i>,
     "<a href="#bucketname" title="BucketName">BucketName</a>" : <i>String</i>,
     "<a href="#s3keyprefix" title="S3KeyPrefix">S3KeyPrefix</a>" : <i>String</i>
 }
@@ -16,12 +18,31 @@ To declare this entity in your AWS CloudFormation template, use the following sy
 ### YAML
 
 <pre>
+<a href="#logdestinationtype" title="LogDestinationType">LogDestinationType</a>: <i>String</i>
+<a href="#logexports" title="LogExports">LogExports</a>: <i>
+      - String</i>
 <a href="#bucketname" title="BucketName">BucketName</a>: <i>String</i>
 <a href="#s3keyprefix" title="S3KeyPrefix">S3KeyPrefix</a>: <i>String</i>
 </pre>
 
 ## Properties
 
+#### LogDestinationType
+
+_Required_: No
+
+_Type_: String
+
+_Update requires_: [No interruption](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/using-cfn-updating-stacks-update-behaviors.html#update-no-interrupt)
+
+#### LogExports
+
+_Required_: No
+
+_Type_: List of String
+
+_Update requires_: [No interruption](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/using-cfn-updating-stacks-update-behaviors.html#update-no-interrupt)
+
 #### BucketName
 
 _Required_: No

aws-redshift-cluster/src/main/java/software/amazon/redshift/cluster/ReadHandler.java

@@ -74,6 +74,8 @@ protected ProgressEvent<ResourceModel, CallbackContext> handleRequest(
                             .makeServiceCall(this::describeLoggingStatus)
                             .done(enableLoggingResponse -> {
                                 LoggingProperties loggingProperties = LoggingProperties.builder()
+                                        .logDestinationType(enableLoggingResponse.logDestinationTypeAsString())
+                                        .logExports(enableLoggingResponse.logExports())
                                         .bucketName(enableLoggingResponse.bucketName())
                                         .s3KeyPrefix(enableLoggingResponse.s3KeyPrefix())
                                         .build();

aws-redshift-cluster/src/main/java/software/amazon/redshift/cluster/Translator.java

@@ -128,11 +128,14 @@ static CreateClusterRequest translateToCreateRequest(final ResourceModel model,
    * @return awsRequest the aws service request to create a resource
    */
   static EnableLoggingRequest translateToEnableLoggingRequest(final ResourceModel model) {
-    String s3KeyPrefix = model.getLoggingProperties().getS3KeyPrefix().lastIndexOf("/")
-            == model.getLoggingProperties().getS3KeyPrefix().length() - 1 ? model.getLoggingProperties().getS3KeyPrefix()
-            : model.getLoggingProperties().getS3KeyPrefix() + "/";
+    String s3KeyPrefix = model.getLoggingProperties().getS3KeyPrefix();
+    if (s3KeyPrefix != null) { // S3 key prefix can be empty if it is CW logging
+      s3KeyPrefix = s3KeyPrefix.lastIndexOf("/") == s3KeyPrefix.length() - 1 ? s3KeyPrefix : s3KeyPrefix + "/";
+    }
     return EnableLoggingRequest.builder()
             .clusterIdentifier(model.getClusterIdentifier())
+            .logDestinationType(model.getLoggingProperties().getLogDestinationType())
+            .logExports(model.getLoggingProperties().getLogExports())
             .bucketName(model.getLoggingProperties().getBucketName())
             .s3KeyPrefix(s3KeyPrefix)
             .build();
@@ -293,6 +296,8 @@ static DescribeLoggingStatusRequest translateToDescribeStatusLoggingRequest(fina
    */
   static ResourceModel translateFromDescribeLoggingResponse(final DescribeLoggingStatusResponse awsResponse) {
       LoggingProperties loggingProperties = LoggingProperties.builder()
+              .logDestinationType(awsResponse.logDestinationTypeAsString())
+              .logExports(awsResponse.logExports())
               .bucketName(awsResponse.bucketName())
               .s3KeyPrefix(awsResponse.s3KeyPrefix())
               .build();

aws-redshift-cluster/src/test/java/software/amazon/redshift/cluster/AbstractTestBase.java

@@ -1,8 +1,11 @@
 package software.amazon.redshift.cluster;
 
+import com.google.common.collect.ImmutableList;
 import java.lang.UnsupportedOperationException;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import software.amazon.awssdk.awscore.AwsRequest;
@@ -35,15 +38,18 @@ public class AbstractTestBase {
   protected static final String CLUSTER_NAMESPACE_ARN;
   protected static final String NAMESPACE_POLICY;
   protected static final String NAMESPACE_POLICY_EMPTY;
+  protected static final String LOG_DESTINATION_TYPE_CW;
   protected static final ResourcePolicy RESOURCE_POLICY;
   protected static final ResourcePolicy RESOURCE_POLICY_EMPTY;
-  protected static final LoggingProperties LOGGING_PROPERTIES;
+  protected static final LoggingProperties LOGGING_PROPERTIES_S3;
+  protected static final LoggingProperties LOGGING_PROPERTIES_CW;
   protected static final LoggingProperties LOGGING_PROPERTIES_DISABLED;
   protected static final software.amazon.awssdk.services.redshift.model.Tag TAG;
-  protected  static final Integer DEFER_MAINTENANCE_DURATION;
-  protected  static final String DEFER_MAINTENANCE_IDENTIFIER;
-  protected  static final String DEFER_MAINTENANCE_START_TIME;
-  protected  static final String DEFER_MAINTENANCE_END_TIME;
+  protected static final Integer DEFER_MAINTENANCE_DURATION;
+  protected static final String DEFER_MAINTENANCE_IDENTIFIER;
+  protected static final String DEFER_MAINTENANCE_START_TIME;
+  protected static final String DEFER_MAINTENANCE_END_TIME;
+  protected static final List<String> LOG_EXPORTS_TYPES;
 
   static {
     MOCK_CREDENTIALS = new Credentials("accessKey", "secretKey", "token");
@@ -67,6 +73,8 @@ public class AbstractTestBase {
     DEFER_MAINTENANCE_START_TIME = "2023-12-10T00:00:00Z";
     DEFER_MAINTENANCE_END_TIME = "2024-01-19T00:00:00Z";
     SNAPSHOT_IDENTIFIER = "redshift-cluster-1-snapshot";
+    LOG_DESTINATION_TYPE_CW = "cloudwatch";
+    LOG_EXPORTS_TYPES = ImmutableList.of("connectionlog", "useractivitylog", "userlog");
 
     RESOURCE_POLICY = ResourcePolicy.builder()
             .resourceArn(CLUSTER_NAMESPACE_ARN)
@@ -79,12 +87,19 @@ public class AbstractTestBase {
             .build();
 
 
-    LOGGING_PROPERTIES = LoggingProperties.builder()
+    LOGGING_PROPERTIES_S3 = LoggingProperties.builder()
             .bucketName(BUCKET_NAME)
             .s3KeyPrefix("test")
             .build();
 
+    LOGGING_PROPERTIES_CW = LoggingProperties.builder()
+            .logDestinationType(LOG_DESTINATION_TYPE_CW)
+            .logExports(LOG_EXPORTS_TYPES)
+            .build();
+
     LOGGING_PROPERTIES_DISABLED = LoggingProperties.builder()
+            .logDestinationType(null)
+            .logExports(new ArrayList<>())
             .bucketName(null)
             .s3KeyPrefix(null)
             .build();
@@ -216,6 +231,23 @@ public static CreateClusterResponse createClusterResponseSdk() {
             .build();
   }
 
+  public static EnableLoggingResponse createS3EnableLoggingResponseSdk() {
+    return EnableLoggingResponse.builder()
+            .bucketName(BUCKET_NAME)
+            .loggingEnabled(true)
+            .lastSuccessfulDeliveryTime(Instant.now())
+            .build();
+  }
+
+  public static EnableLoggingResponse createCWEnableLoggingResponseSdk() {
+    return EnableLoggingResponse.builder()
+            .logDestinationType(LOG_DESTINATION_TYPE_CW)
+            .logExports(LOG_EXPORTS_TYPES)
+            .loggingEnabled(true)
+            .lastSuccessfulDeliveryTime(Instant.now())
+            .build();
+  }
+
   public static DescribeClustersResponse describeClustersResponseSdk() {
     return DescribeClustersResponse.builder()
             .clusters(responseCluster())

aws-redshift-cluster/src/test/java/software/amazon/redshift/cluster/CreateHandlerTest.java

@@ -144,9 +144,10 @@ public void handleRequest_SimpleSuccess() {
     }
 
     @Test
-    public void testCreateClusterAndEnableLogging() {
+    public void testCreateClusterAndEnableS3Logging() {
+        // Arrange
         ResourceModel model = createClusterRequestModel();
-        model.setLoggingProperties(LOGGING_PROPERTIES);
+        model.setLoggingProperties(LOGGING_PROPERTIES_S3);
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .desiredResourceState(model)
@@ -158,22 +159,56 @@ public void testCreateClusterAndEnableLogging() {
         when(proxyClient.client().createCluster(any(CreateClusterRequest.class))).thenReturn(createClusterResponseSdk());
 
         when(proxyClient.client().enableLogging(any(EnableLoggingRequest.class)))
-                .thenReturn(EnableLoggingResponse.builder()
+            .thenReturn(createS3EnableLoggingResponseSdk());
+
+        when(proxyClient.client().describeClusters(any(DescribeClustersRequest.class))).thenReturn(describeClustersResponseSdk());
+
+        when(proxyClient.client().describeLoggingStatus(any(DescribeLoggingStatusRequest.class)))
+                .thenReturn(DescribeLoggingStatusResponse.builder()
                         .bucketName(BUCKET_NAME)
                         .loggingEnabled(true)
                         .lastSuccessfulDeliveryTime(Instant.now())
                         .build());
+        when(proxyClient.client().getResourcePolicy(any(GetResourcePolicyRequest.class))).thenReturn(getEmptyResourcePolicyResponseSdk());
+
+        // Act, Assert
+        handleRequestAndVerifyEnableLogging(request);
+    }
+
+    @Test
+    public void testCreateClusterAndEnableCWLogging() {
+        // Arrange
+        ResourceModel model = createClusterRequestModel();
+        model.setLoggingProperties(LOGGING_PROPERTIES_CW);
+
+        final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
+            .desiredResourceState(model)
+            .region(AWS_REGION)
+            .logicalResourceIdentifier("logicalId")
+            .clientRequestToken("token")
+            .build();
+
+        when(proxyClient.client().createCluster(any(CreateClusterRequest.class))).thenReturn(createClusterResponseSdk());
+
+        when(proxyClient.client().enableLogging(any(EnableLoggingRequest.class)))
+            .thenReturn(createCWEnableLoggingResponseSdk());
 
         when(proxyClient.client().describeClusters(any(DescribeClustersRequest.class))).thenReturn(describeClustersResponseSdk());
 
         when(proxyClient.client().describeLoggingStatus(any(DescribeLoggingStatusRequest.class)))
                 .thenReturn(DescribeLoggingStatusResponse.builder()
-                        .bucketName(BUCKET_NAME)
+                        .logDestinationType(LOG_DESTINATION_TYPE_CW)
+                        .logExports(LOG_EXPORTS_TYPES)
                         .loggingEnabled(true)
                         .lastSuccessfulDeliveryTime(Instant.now())
                         .build());
         when(proxyClient.client().getResourcePolicy(any(GetResourcePolicyRequest.class))).thenReturn(getEmptyResourcePolicyResponseSdk());
 
+        // Act, Assert
+        handleRequestAndVerifyEnableLogging(request);
+    }
+
+    private void handleRequestAndVerifyEnableLogging(ResourceHandlerRequest<ResourceModel> request) {
         ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, new CallbackContext(), proxyClient, logger);
 
         assertThat(response).isNotNull();
@@ -192,6 +227,8 @@ public void testCreateClusterAndEnableLogging() {
                 isEqualTo(request.getDesiredResourceState().getClusterNamespaceArn());
         assertThat(response.getResourceModel().getClusterIdentifier()).
                 isEqualTo(request.getDesiredResourceState().getClusterIdentifier());
+        assertThat(response.getResourceModel().getLoggingProperties()).
+                isEqualTo(request.getDesiredResourceState().getLoggingProperties());
         verify(proxyClient.client()).createCluster(any(CreateClusterRequest.class));
         verify(proxyClient.client(), times(4))
                 .describeClusters(any(DescribeClustersRequest.class));