Support CloudWatch integration for AWS::Redshift::Cluster LoggingProperties

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));