AWSTemplateFormatVersion: '2010-09-09'

Parameters:
  DatabaseName:
    Description: The name of the first database to be created for provisioned Speed cluster
    Type: String
    Default: dev
    AllowedPattern: "([a-z]|[0-9])+"
  NumberOfNodes:
    Description: The number of compute nodes in the cluster. For multi-node clusters,
      the NumberOfNodes parameter must be greater than 1
    Type: Number
    Default: '2'
  NodeType:
    Description: The type of node to be provisioned
    Type: String
    Default: ra3.xlplus
    AllowedValues:
      - ra3.4xlarge
      - ra3.16xlarge
      - ra3.xlplus
  ClusterType:
    Description: The type of cluster to be provisioned
    Type: String
    Default: multi-node
    AllowedValues:
      - multi-node
      - single-node
  MasterUsername:
    Description: The user name that is associated with the master user account for
      the cluster that is being created
    Type: String
    Default: awsuser
    AllowedPattern: "([a-z])([a-z]|[0-9])*"
  MasterUserPassword:
    Description: The password that is associated with the master user account for
      the cluster that is being created.
    Type: String
    Default: 'Speed321'
    NoEcho: 'true'
  InboundTraffic:
    Description: Allow inbound traffic to the cluster from this CIDR range.
    Type: String
    MinLength: '9'
    MaxLength: '18'
    Default: 0.0.0.0/0
    AllowedPattern: "(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})/(\\d{1,2})"
    ConstraintDescription: must be a valid CIDR range of the form x.x.x.x/x.
  PortNumber:
    Description: The port number on which the cluster accepts incoming connections.
    Type: Number
    Default: '5439'
  SQLforData:
    Description: The Source query for data
    Type: String
    Default: SELECT * FROM 'connectedcar/telemetry'
Resources:
  LambdaRole:
    Type: AWS::IAM::Role
    Properties :
      AssumeRolePolicyDocument:
        Version : 2012-10-17
        Statement :
          -
            Effect : Allow
            Principal :
              Service :
                - lambda.amazonaws.com
            Action :
              - sts:AssumeRole
      Path : /
      Policies:
        -
          PolicyName: LambdaCloudFormationPolicy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - s3:*
                Resource:
                  - !Sub "arn:aws:s3:::cloudformation-custom-resource-response-${AWS::Region}"
                  - !Sub "arn:aws:s3:::cloudformation-waitcondition-${AWS::Region}"
                  - !Sub "arn:aws:s3:::cloudformation-custom-resource-response-${AWS::Region}/*"
                  - !Sub "arn:aws:s3:::cloudformation-waitcondition-${AWS::Region}/*"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
        - arn:aws:iam::aws:policy/AmazonS3FullAccess
##        - arn:aws:iam::aws:policy/CloudWatchLogsFullAccess
##        - arn:aws:iam::aws:policy/AmazonRDSDataFullAccess
##        - arn:aws:iam::aws:policy/IAMFullAccess
        - arn:aws:iam::aws:policy/AmazonRedshiftFullAccess
  LambdaFunctionDefaultRole:
    Type: AWS::Lambda::Function
    Properties:
        Timeout: 300
        Code:
          ZipFile: |
            import sys
            import os
            import json
            import cfnresponse
            import logging
            from pip._internal import main

            main(['install', 'boto3', '--target', '/tmp/'])
            sys.path.insert(0,'/tmp/')

            import boto3
            from botocore.exceptions import ClientError

            def lambda_handler(event, context):

                print(boto3.__version__)

                if event['RequestType'] == 'Delete':
                    cfnresponse.send(event, context, cfnresponse.SUCCESS, {'Data': 'Delete complete'})
                else:
                    try:
                        client = boto3.client('redshift')
                        response = client.modify_cluster_iam_roles(
                            ClusterIdentifier=os.environ['RedshiftClusterIdentifier'],
                            DefaultIamRoleArn=os.environ['RedshiftClusterRole']
                        )
                        print(response)
                    except Exception as e:
                        logger.error(e)
                        cfnresponse.send(event, context, cfnresponse.FAILED, {'Data': 'Create failed'})

                cfnresponse.send(event, context, cfnresponse.SUCCESS, {'Data': 'Create complete'})

                return {
                    'statusCode': 200,
                    'body': json.dumps('Deployed Default Role')
                }
          # S3Bucket:
          #   Fn::FindInMap:
          #   - RegionMap
          #   - !Ref AWS::Region
          #   - S3BucketDirectory
          # S3Key: my-deployment-package.zip
        Environment:
          Variables:
            RedshiftClusterIdentifier:
              Ref: SpeedRedshiftCluster
            RedshiftClusterRole:
              Fn::GetAtt: [RedshiftRole, Arn]
        Handler: index.lambda_handler
        Role:
          Fn::GetAtt: [LambdaRole, Arn]
        Runtime: python3.9
    DependsOn:
      - LambdaRole
      - SpeedRedshiftCluster
      - RedshiftRole
  PrimerInvokeDefaultRole:
    Type: AWS::CloudFormation::CustomResource
    DependsOn:
      - LambdaFunctionDefaultRole
    Version: "1.0"
    Properties:
      ServiceToken: !GetAtt 'LambdaFunctionDefaultRole.Arn'
  RedshiftRole:
    Type: AWS::IAM::Role
    Properties :
      RoleName: RedshiftLambdaArchitectureRole
      AssumeRolePolicyDocument:
        Version : 2012-10-17
        Statement :
          -
            Effect : Allow
            Principal :
              Service :
                - redshift.amazonaws.com
            Action :
              - sts:AssumeRole
          -
            Effect : Allow
            Principal :
              Service :
                - events.amazonaws.com
            Action :
              - sts:AssumeRole
          -
            Effect : Allow
            Principal :
              Service :
                - scheduler.redshift.amazonaws.com
            Action :
              - sts:AssumeRole
      Path : /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonS3FullAccess
        - arn:aws:iam::aws:policy/AWSGlueConsoleFullAccess
        - arn:aws:iam::aws:policy/AmazonRedshiftFullAccess
        - arn:aws:iam::aws:policy/AmazonKinesisFullAccess

  IOTdatabucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Join
        - '-'
        - - !Ref 'AWS::AccountId'
          - !Ref 'AWS::Region'
          - 'lambda-demo-iot-databucket'


  IOTKDStream:
    Type: AWS::Kinesis::Stream
    Properties:
      Name: 'vehicleiotstream'
      RetentionPeriodHours: 168
      ShardCount: 3
  IOTRule:
    Type: AWS::IoT::TopicRule
    Properties:
      TopicRulePayload:
        RuleDisabled: false
        Sql:
          Ref: SQLforData
        Actions:
          - Kinesis:
              StreamName:
                Ref: IOTKDStream
              RoleArn: !GetAtt IOTKinesisRole.Arn
  IOTKinesisRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: iot.amazonaws.com
            Action: 'sts:AssumeRole'
      Policies:
        - PolicyName: IOT_delivery_policy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              Effect: Allow
              Action: kinesis:PutRecord
              Resource: !GetAtt IOTKDStream.Arn
    # Kinesis read role for Firehose to read kinesis stream
  FirehoseKinesisReadRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - firehose.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: "/"
      Policies:
        - PolicyName: FirehoseKinesisRead
          PolicyDocument:
            Statement:
              - Effect: Allow
                Action:
                  - kinesis:DescribeStream
                  - kinesis:GetShardIterator
                  - kinesis:GetRecords
                  - kinesis:ListShards
                Resource:
                  - !GetAtt IOTKDStream.Arn

  GlueRole:
        Type: "AWS::IAM::Role"
        Properties:
            Path: "/"
            RoleName: "AWSGlueServiceRoleDefaultdemo"
            AssumeRolePolicyDocument: "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Service\":\"glue.amazonaws.com\"},\"Action\":\"sts:AssumeRole\"}]}"
            MaxSessionDuration: 3600
            Policies:
              - PolicyName: GlueS3AccessPolicy
                PolicyDocument:
                  Version: "2012-10-17"
                  Statement:
                    - Effect: Allow
                      Action: ['s3:*','s3-object-lambda:*']
                      Resource:
                        - !Join
                          - ''
                          - - 'arn:aws:s3:::'
                            - !Ref IOTdatabucket
                            - /*
            ManagedPolicyArns:
              - "arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole"
              - "arn:aws:iam::aws:policy/AmazonKinesisReadOnlyAccess"
            Description: "Allows Glue to call AWS services on your behalf. "

  GlueDatabase:
    Type: AWS::Glue::Database
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseInput:
        Description: "The Database for raw telematics data."
        Name: 'vehicle_telematics_database'

  GlueTable:
    Type: AWS::Glue::Table
    Properties:
      CatalogId: !Ref AWS::AccountId
      DatabaseName: !Ref GlueDatabase
      TableInput:
        Owner: owner
        Name: 'vehicle_telematics_raw'
        Retention: 0
        Parameters:
          classification: "parquet"
          CrawlerSchemaDeserializerVersion: "1.0"
          CrawlerSchemaSerializerVersion: "1.0"
          UPDATED_BY_CRAWLER: "vehicle_iot_crawler"
          compressionType: "none"
          typeOfData: "file"
        StorageDescriptor:
          Columns:
          - Name: VIN
            Type: string
          - Name: trip_id
            Type: string
          - Name: brake
            Type: float
          - Name: steeringWheelAngle
            Type: float
          - Name: torqueAtTransmission
            Type: float
          - Name: engineSpeed
            Type: float
          - Name: vehicleSpeed
            Type: float
          - Name: acceleration
            Type: float
          - Name: acceleratorPedalPos
            Type: float
          - Name: parkingBrakeStatus
            Type: boolean
          - Name: brakePedalStatus
            Type: boolean
          - Name: transmissionGearPosition
            Type: string
          - Name: gearLeverPosition
            Type: string
          - Name: odometer
            Type: float
          - Name: ignitionStatus
            Type: string
          - Name: fuelLevel
            Type: float
          - Name: fuelConsumedSinceRestart
            Type: float
          - Name: oilTemp
            Type: float
          - Name: location
            Type: string
          Location: !Sub "s3://${IOTdatabucket}/vehicle_telematics_raw/"
          InputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat
          OutputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat
          Compressed: false
          NumberOfBuckets: -1
          SerdeInfo:
            SerializationLibrary: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
            Parameters:
              serialization.format: '1'
          BucketColumns: []
          SortColumns: []
          Parameters:
              CrawlerSchemaDeserializerVersion: "1.0"
              CrawlerSchemaSerializerVersion: "1.0"
              UPDATED_BY_CRAWLER: "vehicle_iot_crawler"
              classification: "parquet"
              compressionType: "none"
              typeOfData: "file"
          StoredAsSubDirectories: false
        PartitionKeys:
        - Name: year
          Type: string
        - Name: month
          Type: string
        - Name: day
          Type: string
        TableType: EXTERNAL_TABLE

  GlueCrawler:
    Type: AWS::Glue::Crawler
    DependsOn:
      - GlueDatabase
      - GlueTable
    Properties:
      Name: 'vehicle_iot_crawler'
      Role: !GetAtt GlueRole.Arn
      #Classifiers: none, use the default classifier
      Description: AWS Glue crawler to crawl vehicle iot data
      #Schedule: none, use default run-on-demand
      DatabaseName: !Ref GlueDatabase
      Targets:
        CatalogTargets:
          - DatabaseName: !Ref GlueDatabase
            Tables:
              - !Ref GlueTable
      SchemaChangePolicy:
        UpdateBehavior: "UPDATE_IN_DATABASE"
        DeleteBehavior: "LOG"
      Configuration: "{\"Version\":1.0,\"CrawlerOutput\":{\"Partitions\":{\"AddOrUpdateBehavior\":\"InheritFromTable\"},\"Tables\":{\"AddOrUpdateBehavior\":\"MergeNewColumns\"}}}"



  DeliveryStream:
    Type: AWS::KinesisFirehose::DeliveryStream
    DependsOn:
      - GlueDatabase
      - GlueTable
    Properties:
      DeliveryStreamName: 'vehicleiotdeliverystream'
      DeliveryStreamType: KinesisStreamAsSource
      KinesisStreamSourceConfiguration:
        KinesisStreamARN: !GetAtt IOTKDStream.Arn
        RoleARN: !GetAtt FirehoseKinesisReadRole.Arn
      ExtendedS3DestinationConfiguration:
        RoleARN: !GetAtt DeliveryRole.Arn
        BucketARN: !Join
          - ''
          - - 'arn:aws:s3:::'
            - !Ref IOTdatabucket
        Prefix: !Join
          - ''
          - - !Ref GlueTable
            -  '/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/'
        ErrorOutputPrefix: !Join
          - ''
          - - !Ref GlueTable
            -  'error/!{firehose:error-output-type}/year=!{timestamp:YYYY}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/'
        BufferingHints:
          SizeInMBs: 128
          IntervalInSeconds: 60
        CompressionFormat: UNCOMPRESSED
        EncryptionConfiguration:
          NoEncryptionConfig: NoEncryption
        CloudWatchLoggingOptions:
          Enabled: true
          LogGroupName: !Join
            - ''
            - - 'KDF-'
              - !Ref GlueTable
          LogStreamName: S3Delivery
        S3BackupMode: Disabled
        DataFormatConversionConfiguration:
          SchemaConfiguration:
            CatalogId: !Ref AWS::AccountId
            RoleARN: !GetAtt DeliveryRole.Arn
            DatabaseName: !Ref GlueDatabase
            TableName: !Ref GlueTable
            Region: !Ref AWS::Region
            VersionId: LATEST
          InputFormatConfiguration:
            Deserializer:
              OpenXJsonSerDe: {}
          OutputFormatConfiguration:
            Serializer:
              ParquetSerDe: {}
          Enabled: True

  DeliveryRole:
    Type: AWS::IAM::Role
    DependsOn:
      - GlueDatabase
      - GlueTable
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: firehose.amazonaws.com
            Action: 'sts:AssumeRole'
            Condition:
              StringEquals:
                'sts:ExternalId': !Ref 'AWS::AccountId'
      Path: "/"
      Policies:
        - PolicyName: firehose_delivery_policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 's3:AbortMultipartUpload'
                  - 's3:GetBucketLocation'
                  - 's3:GetObject'
                  - 's3:ListBucket'
                  - 's3:ListBucketMultipartUploads'
                  - 's3:PutObject'
                Resource:
                  - !Join
                    - ''
                    - - 'arn:aws:s3:::'
                      - !Ref IOTdatabucket
                  - !Join
                    - ''
                    - - 'arn:aws:s3:::'
                      - !Ref IOTdatabucket
                      - '/*'
              - Effect: Allow
                Action: 'glue:GetTableVersions'
                Resource: '*'
              - Effect: Allow
                Action: 'logs:PutLogEvents'
                Resource:
                - !Join
                    - ''
                    - - 'arn:aws:logs:'
                      - !Ref 'AWS::Region'
                      - ':'
                      - !Ref 'AWS::AccountId'
                      - 'log-group:/aws/kinesisfirehose/KDF-vehicle_telematics_raw'
                      - ':log-stream:*'



  SpeedRedshiftCluster:
    Type: AWS::Redshift::Cluster
    DependsOn: AttachGateway
    Properties:
      ClusterIdentifier: "speed-cluster-iot"
      NumberOfNodes: 2
      NodeType:
        Ref: NodeType
      ClusterType:
        Ref: ClusterType
      DBName:
        Ref: DatabaseName
      MasterUsername:
        Ref: MasterUsername
      MasterUserPassword:
        Ref: MasterUserPassword
      ClusterParameterGroupName:
        Ref: RedshiftClusterParameterGroup
      VpcSecurityGroupIds:
      - Ref: SecurityGroup
      ClusterSubnetGroupName:
        Ref: RedshiftClusterSubnetGroup
      PubliclyAccessible: 'true'
      Port:
        Ref: PortNumber
      IamRoles:
        - Fn::GetAtt: [ RedshiftRole, Arn ]
  RedshiftClusterParameterGroup:
    Type: AWS::Redshift::ClusterParameterGroup
    Properties:
      Description: Cluster parameter group
      ParameterGroupFamily: redshift-1.0
      Parameters:
      - ParameterName: enable_user_activity_logging
        ParameterValue: 'true'
  RedshiftClusterSubnetGroup:
    Type: AWS::Redshift::ClusterSubnetGroup
    Properties:
      Description: Cluster subnet group
      SubnetIds:
      - Ref: PublicSubnet
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.0.0.0/16
  PublicSubnet:
    Type: AWS::EC2::Subnet
    Properties:
      CidrBlock: 10.0.0.0/24
      VpcId:
        Ref: VPC
  SecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group
      SecurityGroupIngress:
      - CidrIp:
          Ref: InboundTraffic
        FromPort:
          Ref: PortNumber
        ToPort:
          Ref: PortNumber
        IpProtocol: tcp
      VpcId:
        Ref: VPC
  MyInternetGateway:
    Type: AWS::EC2::InternetGateway
  AttachGateway:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId:
        Ref: VPC
      InternetGatewayId:
        Ref: MyInternetGateway
  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId:
        Ref: VPC
  PublicRoute:
    Type: AWS::EC2::Route
    DependsOn: AttachGateway
    Properties:
      RouteTableId:
        Ref: PublicRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId:
        Ref: MyInternetGateway
  PublicSubnetRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId:
        Ref: PublicSubnet
      RouteTableId:
        Ref: PublicRouteTable
Outputs:
  DeliveryStreamARN:
    Description: The ARN of the firehose delivery stream
    Value: !GetAtt DeliveryStream.Arn

  SpeedClusterName:
    Description: Name of cluster
    Value:
      Ref: SpeedRedshiftCluster

  ParameterGroupName:
    Description: Name of parameter group
    Value:
      Ref: RedshiftClusterParameterGroup
  RedshiftClusterSubnetGroupName:
    Description: Name of cluster subnet group
    Value:
      Ref: RedshiftClusterSubnetGroup
  RedshiftClusterSecurityGroupName:
    Description: Name of cluster security group
    Value:
      Ref: SecurityGroup
  IOTdatabucket:
    Description: Name of IOT Data Bucket
    Value:
      Ref: IOTdatabucket
  IOTKDStream:
    Description: Name of IOT Kinesis Data Stream
    Value:
      Ref: IOTKDStream
  IOTRule:
    Description: Name of IOT Rule to load Kinesis Data Stream
    Value:
      Ref: IOTRule
