AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  MSKSourceKafkaVersion:
    Type: String
    Default: 2.6.1
    Description: The Apache Kafka version for the source Amazon MSK cluster.
    AllowedValues:
      - 1.1.1
      - 2.2.1
      - 2.3.1
      - 2.4.1.1
      - 2.5.1
      - 2.6.0
      - 2.6.1
      - 2.7.0
  MSKDestinationKafkaVersion:
    Type: String
    Default: 2.7.0
    Description: The Apache Kafka version for the destination Amazon MSK cluster.
    AllowedValues:
      - 1.1.1
      - 2.2.1
      - 2.3.1
      - 2.4.1.1
      - 2.5.1
      - 2.6.0
      - 2.6.1
      - 2.7.0
  TLSMutualAuthenticationSourceMSKCluster:
      Type: String
      Default: false
      Description: Whether TLS Mutual Auth should be enabled for the Amazon MSK cluster1.
      AllowedValues:
        - true
        - false
  TLSMutualAuthenticationDestinationMSKCluster:
      Type: String
      Default: false
      Description: Whether TLS Mutual Auth should be enabled for the Amazon MSK cluster2.
      AllowedValues:
        - true
        - false
  PCAARNSourceMSKCluster:
      Type: String
      AllowedPattern: 'arn:aws:acm-pca:[us\-east\-1|us\-east\-2|eu\-west\-1]{9}:\d{12}:certificate-authority\/[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}|^$'
      ConstraintDescription: Not a Valid ACM PCA ARN
      Description: Provide the ARN for an ACM PCA in your account for the source MSK Cluster
  PCAARNDestinationMSKCluster:
      Type: String
      AllowedPattern: 'arn:aws:acm-pca:[us\-east\-1|us\-east\-2|eu\-west\-1]{9}:\d{12}:certificate-authority\/[\da-f]{8}-[\da-f]{4}-[\da-f]{4}-[\da-f]{4}-[\da-f]{12}|^$'
      ConstraintDescription: Not a Valid ACM PCA ARN
      Description: Provide the ARN for an ACM PCA in your account for the destination MSK Cluster

Conditions:
  MTLSMSKCluster1: !Equals [ !Ref TLSMutualAuthenticationSourceMSKCluster, true ]
  noMTLSMSKCluster1: !Equals [ !Ref TLSMutualAuthenticationSourceMSKCluster, false ]
  MTLSMSKCluster2: !Equals [ !Ref TLSMutualAuthenticationDestinationMSKCluster, true ]
  noMTLSMSKCluster2: !Equals [ !Ref TLSMutualAuthenticationDestinationMSKCluster, false ]

Resources:

  MSKVPCStack:
    Type: AWS::CloudFormation::Stack
    Properties:
      TemplateURL: "https://aws-streaming-artifacts.s3.amazonaws.com/msk-lab-resources/MSKLabs/MSKPrivateVPCOnly.yml"

  KafkaClientInstanceSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Enable SSH access via port 22 from BastionHostSecurityGroup
      VpcId: !GetAtt MSKVPCStack.Outputs.VPCId
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 22
        ToPort: 22
        CidrIp: 10.0.0.0/24
      - IpProtocol: tcp
        FromPort: 3500
        ToPort: 3500
        CidrIp: 10.0.0.0/24
      - IpProtocol: tcp
        FromPort: 3600
        ToPort: 3600
        CidrIp: 10.0.0.0/24
      - IpProtocol: tcp
        FromPort: 3800
        ToPort: 3800
        CidrIp: 10.0.0.0/24
      - IpProtocol: tcp
        FromPort: 3900
        ToPort: 3900
        CidrIp: 10.0.0.0/24

  KafkaClientInstanceSecurityGroup8081:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn: KafkaClientInstanceSecurityGroup
    Properties:
      Description: Enable access to Schema Registry inside the KafkaClientInstanceSecurityGroup
      GroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      IpProtocol: tcp
      FromPort: 8081
      ToPort: 8081
      SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId

  KafkaClientInstanceSecurityGroup8083:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn: KafkaClientInstanceSecurityGroup
    Properties:
      Description: Enable access to Kafka Connect inside the KafkaClientInstanceSecurityGroup
      GroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      IpProtocol: tcp
      FromPort: 8083
      ToPort: 8083
      SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId

  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK Security Group
      VpcId: !GetAtt MSKVPCStack.Outputs.VPCId
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 2181
        ToPort: 2181
        SourceSecurityGroupId: !Ref KafkaClientInstanceSecurityGroup
      - IpProtocol: tcp
        FromPort: 9094
        ToPort: 9094
        SourceSecurityGroupId: !Ref KafkaClientInstanceSecurityGroup
      - IpProtocol: tcp
        FromPort: 9092
        ToPort: 9092
        SourceSecurityGroupId: !Ref KafkaClientInstanceSecurityGroup

  MSKSourceClusterMTLS:
    Type: AWS::MSK::Cluster
    Condition: MTLSMSKCluster1
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKOne
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKTwo
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKThree
        InstanceType: kafka.m5.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 100
      ClusterName: !Join
                    - '-'
                    - - 'MSKCluster1'
                      - !Ref 'AWS::StackName'
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS_PLAINTEXT
          InCluster: true
      ClientAuthentication:
        Tls:
            CertificateAuthorityArnList:
              - !Ref PCAARNSourceMSKCluster
      EnhancedMonitoring: DEFAULT
      KafkaVersion: !Ref MSKSourceKafkaVersion
      NumberOfBrokerNodes: 3

  MSKSourceClusterNoMTLS:
    Type: AWS::MSK::Cluster
    Condition: noMTLSMSKCluster1
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKOne
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKTwo
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKThree
        InstanceType: kafka.m5.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 100
      ClusterName: !Join
                    - '-'
                    - - 'MSKCluster1'
                      - !Ref 'AWS::StackName'
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS_PLAINTEXT
          InCluster: true
      EnhancedMonitoring: DEFAULT
      KafkaVersion: !Ref MSKSourceKafkaVersion
      NumberOfBrokerNodes: 3

  MSKDestinationClusterMTLS:
    Type: AWS::MSK::Cluster
    Condition: MTLSMSKCluster2
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKOne
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKTwo
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKThree
        InstanceType: kafka.m5.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 100
      ClusterName: !Join
                    - '-'
                    - - 'MSKCluster2'
                      - !Ref 'AWS::StackName'
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS_PLAINTEXT
          InCluster: true
      ClientAuthentication:
        Tls:
            CertificateAuthorityArnList:
              - !Ref PCAARNDestinationMSKCluster
      EnhancedMonitoring: DEFAULT
      KafkaVersion: !Ref MSKDestinationKafkaVersion
      NumberOfBrokerNodes: 3

  MSKDestinationClusterNoMTLS:
    Type: AWS::MSK::Cluster
    Condition: noMTLSMSKCluster2
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKOne
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKTwo
          - !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKThree
        InstanceType: kafka.m5.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
        StorageInfo:
          EBSStorageInfo:
            VolumeSize: 100
      ClusterName: !Join
                    - '-'
                    - - 'MSKCluster2'
                      - !Ref 'AWS::StackName'
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS_PLAINTEXT
          InCluster: true
      EnhancedMonitoring: DEFAULT
      KafkaVersion: !Ref MSKDestinationKafkaVersion
      NumberOfBrokerNodes: 3

Outputs:
  MSKSourceClusterArn:
    Description: The Arn for the Source MSK cluster
    Value: !If [MTLSMSKCluster1, !Ref 'MSKSourceClusterMTLS', !Ref 'MSKSourceClusterNoMTLS']
  MSKDestinationClusterArn:
    Description: The Arn for the Destination MSK cluster
    Value: !If [MTLSMSKCluster2, !Ref 'MSKDestinationClusterMTLS', !Ref 'MSKDestinationClusterNoMTLS']
  KafkaClientEC2InstanceSecurityGroupId:
    Description: The security group id for the EC2 instance
    Value: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
    Export:
      Name: !Sub "${AWS::StackName}-KafkaClientEC2InstanceSecurityGroupId"
  MSKClustersSecurityGroupId:
    Description: The security group id for the Amazon MSK clusters
    Value: !GetAtt MSKSecurityGroup.GroupId
  VPCStackName:
    Description: The name of the VPC Stack
    Value: !GetAtt MSKVPCStack.Outputs.VPCStackName
