Add AWS::Redshift::Cluster support to LocalStack (localstack#9403)

localstack/services/cloudformation/resource_provider.py

@@ -75,6 +75,7 @@
     "AWS::ElasticBeanstalk::ConfigurationTemplate": "ResourceProvider",
     "AWS::CertificateManager::Certificate": "ResourceProvider",
     "AWS::EKS::Nodegroup": "ResourceProvider",
+    "AWS::Redshift::Cluster": "ResourceProvider",
 }
 
 

localstack/services/redshift/resource_providers/aws_redshift_cluster.py

@@ -0,0 +1,210 @@
+# LocalStack Resource Provider Scaffolding v2
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Optional, TypedDict
+
+import localstack.services.cloudformation.provider_utils as util
+from localstack.services.cloudformation.resource_provider import (
+    OperationStatus,
+    ProgressEvent,
+    ResourceProvider,
+    ResourceRequest,
+)
+
+
+class RedshiftClusterProperties(TypedDict):
+    ClusterType: Optional[str]
+    DBName: Optional[str]
+    MasterUserPassword: Optional[str]
+    MasterUsername: Optional[str]
+    NodeType: Optional[str]
+    AllowVersionUpgrade: Optional[bool]
+    AquaConfigurationStatus: Optional[str]
+    AutomatedSnapshotRetentionPeriod: Optional[int]
+    AvailabilityZone: Optional[str]
+    AvailabilityZoneRelocation: Optional[bool]
+    AvailabilityZoneRelocationStatus: Optional[str]
+    Classic: Optional[bool]
+    ClusterIdentifier: Optional[str]
+    ClusterParameterGroupName: Optional[str]
+    ClusterSecurityGroups: Optional[list[str]]
+    ClusterSubnetGroupName: Optional[str]
+    ClusterVersion: Optional[str]
+    DeferMaintenance: Optional[bool]
+    DeferMaintenanceDuration: Optional[int]
+    DeferMaintenanceEndTime: Optional[str]
+    DeferMaintenanceIdentifier: Optional[str]
+    DeferMaintenanceStartTime: Optional[str]
+    DestinationRegion: Optional[str]
+    ElasticIp: Optional[str]
+    Encrypted: Optional[bool]
+    Endpoint: Optional[Endpoint]
+    EnhancedVpcRouting: Optional[bool]
+    HsmClientCertificateIdentifier: Optional[str]
+    HsmConfigurationIdentifier: Optional[str]
+    IamRoles: Optional[list[str]]
+    Id: Optional[str]
+    KmsKeyId: Optional[str]
+    LoggingProperties: Optional[LoggingProperties]
+    MaintenanceTrackName: Optional[str]
+    ManualSnapshotRetentionPeriod: Optional[int]
+    NumberOfNodes: Optional[int]
+    OwnerAccount: Optional[str]
+    Port: Optional[int]
+    PreferredMaintenanceWindow: Optional[str]
+    PubliclyAccessible: Optional[bool]
+    ResourceAction: Optional[str]
+    RevisionTarget: Optional[str]
+    RotateEncryptionKey: Optional[bool]
+    SnapshotClusterIdentifier: Optional[str]
+    SnapshotCopyGrantName: Optional[str]
+    SnapshotCopyManual: Optional[bool]
+    SnapshotCopyRetentionPeriod: Optional[int]
+    SnapshotIdentifier: Optional[str]
+    Tags: Optional[list[Tag]]
+    VpcSecurityGroupIds: Optional[list[str]]
+
+
+class Tag(TypedDict):
+    Key: Optional[str]
+    Value: Optional[str]
+
+
+class LoggingProperties(TypedDict):
+    BucketName: Optional[str]
+    S3KeyPrefix: Optional[str]
+
+
+class Endpoint(TypedDict):
+    Address: Optional[str]
+    Port: Optional[str]
+
+
+REPEATED_INVOCATION = "repeated_invocation"
+
+
+class RedshiftClusterProvider(ResourceProvider[RedshiftClusterProperties]):
+
+    TYPE = "AWS::Redshift::Cluster"  # Autogenerated. Don't change
+    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change
+
+    def create(
+        self,
+        request: ResourceRequest[RedshiftClusterProperties],
+    ) -> ProgressEvent[RedshiftClusterProperties]:
+        """
+        Create a new resource.
+
+        Primary identifier fields:
+          - /properties/ClusterIdentifier
+
+        Required properties:
+          - MasterUserPassword
+          - NodeType
+          - MasterUsername
+          - DBName
+          - ClusterType
+
+        Create-only properties:
+          - /properties/ClusterIdentifier
+          - /properties/OwnerAccount
+          - /properties/SnapshotIdentifier
+          - /properties/DBName
+          - /properties/SnapshotClusterIdentifier
+          - /properties/ClusterSubnetGroupName
+          - /properties/MasterUsername
+
+        Read-only properties:
+          - /properties/Id
+          - /properties/DeferMaintenanceIdentifier
+          - /properties/Endpoint/Port
+          - /properties/Endpoint/Address
+
+        IAM permissions required:
+          - redshift:DescribeClusters
+          - redshift:CreateCluster
+          - redshift:RestoreFromClusterSnapshot
+          - redshift:EnableLogging
+
+        """
+        model = request.desired_state
+        redshift = request.aws_client_factory.redshift
+
+        if not model.get("ClusterIdentifier"):
+            model["ClusterIdentifier"] = util.generate_default_name(
+                stack_name=request.stack_name, logical_resource_id=request.logical_resource_id
+            )
+
+        redshift.create_cluster(**model)
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def read(
+        self,
+        request: ResourceRequest[RedshiftClusterProperties],
+    ) -> ProgressEvent[RedshiftClusterProperties]:
+        """
+        Fetch resource information
+
+        IAM permissions required:
+          - redshift:DescribeClusters
+          - redshift:DescribeLoggingStatus
+          - redshift:DescribeSnapshotCopyGrant
+          - redshift:DescribeClusterDbRevisions
+        """
+        raise NotImplementedError
+
+    def delete(
+        self,
+        request: ResourceRequest[RedshiftClusterProperties],
+    ) -> ProgressEvent[RedshiftClusterProperties]:
+        """
+        Delete a resource
+
+        IAM permissions required:
+          - redshift:DescribeClusters
+          - redshift:DeleteCluster
+        """
+        model = request.desired_state
+        redshift = request.aws_client_factory.redshift
+
+        redshift.delete_cluster(ClusterIdentifier=model["ClusterIdentifier"])
+        return ProgressEvent(
+            status=OperationStatus.SUCCESS,
+            resource_model=model,
+            custom_context=request.custom_context,
+        )
+
+    def update(
+        self,
+        request: ResourceRequest[RedshiftClusterProperties],
+    ) -> ProgressEvent[RedshiftClusterProperties]:
+        """
+        Update a resource
+
+        IAM permissions required:
+          - redshift:DescribeClusters
+          - redshift:ModifyCluster
+          - redshift:ModifyClusterIamRoles
+          - redshift:EnableLogging
+          - redshift:CreateTags
+          - redshift:DeleteTags
+          - redshift:DisableLogging
+          - redshift:RebootCluster
+          - redshift:EnableSnapshotCopy
+          - redshift:DisableSnapshotCopy
+          - redshift:ModifySnapshotCopyRetentionPeriod
+          - redshift:ModifyAquaConfiguration
+          - redshift:ResizeCluster
+          - redshift:ModifyClusterMaintenance
+          - redshift:DescribeClusterDbRevisions
+          - redshift:ModifyClusterDbRevisions
+          - redshift:PauseCluster
+          - redshift:ResumeCluster
+          - redshift:RotateEncryptionKey
+        """
+        raise NotImplementedError