---
AWSTemplateFormatVersion: 2010-09-09
Description: 'Master manifest for ADP Streaming usecase'
Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      - Label:
          default: General configuration
        Parameters:
          - ADPS3BucketName
          - ADPS3BucketRegion
          - BasePlatformVersion
          - UseCaseVersion
      - Label:
          default: Application configuration
        Parameters:
          - InputFilePath

Parameters:
  ADPS3BucketName:
    Type: String
    AllowedPattern: ^[0-9a-z]+([0-9a-z-]*[0-9a-z])*$
    ConstraintDescription:
      'Bucket name can include numbers, lowercase letters, and hyphens (-).
       It cannot start or end with a hyphen (-).'
    Description:
      'ADP bucket name with cloudformation scripts can include numbers,
      lowercase letters, and hyphens (-). It cannot start or end with a hyphen (-).'
    Default: 'griddynamics-analytical-data-platform-releases'
  ADPS3BucketRegion:
    Type: String
    Description: 'The AWS Region where the ADP S3 bucket (ADPS3BucketName) is hosted.'
    Default: 'us-east-1'
  BasePlatformVersion:
    Type: String
    Description: 'BasePlatform build version.'
    Default: '2023-02-16-marketplace'
  UseCaseVersion:
    Type: String
    Description: 'UseCase build version.'
    Default: '2023-02-16-marketplace'
  InputFilePath:
    Type: String
    Description: 'Input file for Batch pipline usecase.'
    Default: 's3://griddynamics-analytical-data-platform-releases/input/stream/part-of-transaction-events.csv'

Conditions:
  RunningInAnotherRegion: !Not [!Equals [!Ref ADPS3BucketRegion, !Ref "AWS::Region"]]

Resources:
  BasePlatformUtils:
    Type: AWS::CloudFormation::Stack
    Properties:
      Tags:
        - Key: 'Name'
          Value: 'BasePlatformUtils'
        - Key: 'Owner'
          Value: !ImportValue ADPPlatformOwner
        - Key: 'Project'
          Value: 'ADP'
      TemplateURL: !Sub 'https://${ADPS3BucketName}.s3.${ADPS3BucketRegion}.${AWS::URLSuffix}/platform/${BasePlatformVersion}/cloudformation/l0/base/utils.yaml'
      Parameters:
        ResourcePrefix: !ImportValue ADPResourcePrefix
        ADPS3BucketName: !Sub
          - '${BucketName}'
          - BucketName: !If [RunningInAnotherRegion, !ImportValue ADPS3BucketAnotherRegion, !Ref ADPS3BucketName]
        BasePlatformVersion: !Ref BasePlatformVersion

  BasePlatformVPCOutputs:
    DependsOn: BasePlatformUtils
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformOutputsLambdaArn
      StackName: "VPCStack"
      BpStateBucket: !Sub
        - "${ResourcePrefix}-platform-state-${AWS::AccountId}"
        - ResourcePrefix: !ImportValue ADPResourcePrefix

  BasePlatformLambdaFunctionsOutputs:
    DependsOn: BasePlatformVPCOutputs
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformOutputsLambdaArn
      StackName: "BasePlatformLambdaFunctions"
      BpStateBucket: !Sub
        - "${ResourcePrefix}-platform-state-${AWS::AccountId}"
        - ResourcePrefix: !ImportValue ADPResourcePrefix

  InitDeploymentBucketOutputs:
    Condition: RunningInAnotherRegion
    DependsOn: BasePlatformLambdaFunctionsOutputs
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformOutputsLambdaArn
      StackName: "InitDeploymentBucket"
      BpStateBucket: !Sub
        - "${ResourcePrefix}-platform-state-${AWS::AccountId}"
        - ResourcePrefix: !ImportValue ADPResourcePrefix

  BasePlatformCloudMapOutputs:
    DependsOn: BasePlatformLambdaFunctionsOutputs
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformOutputsLambdaArn
      StackName: "CloudMapStack"
      BpStateBucket: !Sub
        - "${ResourcePrefix}-platform-state-${AWS::AccountId}"
        - ResourcePrefix: !ImportValue ADPResourcePrefix

  CopyDeploymentFiles:
    Condition: RunningInAnotherRegion
    DependsOn: InitDeploymentBucketOutputs
    Type: Custom::CopyFiles
    Properties:
      ServiceToken: !GetAtt InitDeploymentBucketOutputs.CopyFilesFunctionArn
      DestBucket: !GetAtt InitDeploymentBucketOutputs.FilesBucket
      SourceBucket: !Ref ADPS3BucketName
      Prefix: !Sub 'applications/adp/${UseCaseVersion}/'

  BasePlatformKubernetes:
    DependsOn: BasePlatformCloudMapOutputs
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformLambdaOperateArn
      BpStateBucket: !GetAtt BasePlatformUtils.Outputs.BasePlatformStateBucketName
      BpGdComponent: "streaming"
      BpAwsComponent: "Kubernetes"
      BpReferenceComponentName: "KubernetesStack"

  BasePlatformKubernetesExtra:
    DependsOn: BasePlatformKubernetes
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformLambdaOperateArn
      BpStateBucket: !GetAtt BasePlatformUtils.Outputs.BasePlatformStateBucketName
      BpGdComponent: "streaming"
      BpAwsComponent: "KubernetesExtra"
      BpReferenceComponentName: "KubernetesExtraStack"

  BasePlatformAirflow:
    DependsOn: BasePlatformKubernetesExtra
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformLambdaOperateArn
      BpStateBucket: !GetAtt BasePlatformUtils.Outputs.BasePlatformStateBucketName
      BpGdComponent: "streaming"
      BpAwsComponent: "Airflow"
      BpReferenceComponentName: "AirflowStack"

  BasePlatformRedshift:
    DependsOn: BasePlatformAirflow
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformLambdaOperateArn
      BpStateBucket: !GetAtt BasePlatformUtils.Outputs.BasePlatformStateBucketName
      BpGdComponent: "streaming"
      BpAwsComponent: "Redshift"
      BpReferenceComponentName: "RedshiftStack"

  BasePlatformEMR:
    DependsOn: BasePlatformRedshift
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformLambdaOperateArn
      BpStateBucket: !GetAtt BasePlatformUtils.Outputs.BasePlatformStateBucketName
      BpGdComponent: "streaming"
      BpAwsComponent: "EMR"
      BpReferenceComponentName: "EMRStack"

  BasePlatformKinesis:
    DependsOn: BasePlatformEMR
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformUtils.Outputs.BasePlatformLambdaOperateArn
      BpStateBucket: !GetAtt BasePlatformUtils.Outputs.BasePlatformStateBucketName
      BpGdComponent: "streaming"
      BpAwsComponent: "Kinesis"
      BpReferenceComponentName: "KinesisStack"

  StreamingTargetBucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Delete
    Properties:
      BucketName: !Sub
        - '${ResourcePrefix}-streaming-${StackPrefix}'
        - { ResourcePrefix: !ImportValue ADPResourcePrefix,
            StackPrefix: !Select [ 2, !Split [ "/", !Ref "AWS::StackId" ] ] }
      AccessControl: "Private"
      Tags:
        - Key: 'Name'
          Value: 'StreamingTargetBucket'
        - Key: 'Owner'
          Value: !ImportValue ADPPlatformOwner
        - Key: 'Project'
          Value: 'ADP'

  EmptyBucketLambdaExec:
    Type: "Custom::EmptyBucketLambda"
    Properties:
      ServiceToken: !GetAtt BasePlatformLambdaFunctionsOutputs.EmptyBucketLambdaArn
      BucketToEmpty: !Ref StreamingTargetBucket

  DagsCopyExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: [lambda.amazonaws.com]
            Action: ['sts:AssumeRole']
      ManagedPolicyArns:
        - "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
      Policies:
        - PolicyName: root
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - 'ec2:DescribeNetworkInterfaces'
                  - 'ec2:DeleteNetworkInterface'
                  - 'ec2:CreateNetworkInterface'
                  - 'servicediscovery:DiscoverInstances'
                Resource:
                  - '*'
              - Effect: Allow
                Action: 'servicediscovery:*'
                Resource:
                  - !Sub
                    - 'arn:aws:servicediscovery:${AWS::Region}:${AWS::AccountId}:namespace/${CloudMapNamespaceId}'
                    - CloudMapNamespaceId: !GetAtt BasePlatformCloudMapOutputs.CloudMapNamespaceId
                  - !Sub 'arn:aws:servicediscovery:${AWS::Region}:${AWS::AccountId}:service/srv-*'
              - Effect: Allow
                Action: 's3:*'
                Resource:
                  - !Sub
                    - 'arn:${AWS::Partition}:s3:::${BucketName}'
                    - BucketName: !If [ RunningInAnotherRegion, !GetAtt InitDeploymentBucketOutputs.FilesBucket, !Ref ADPS3BucketName ]
                  - !Sub
                    - 'arn:${AWS::Partition}:s3:::${BucketName}/*'
                    - BucketName: !If [ RunningInAnotherRegion, !GetAtt InitDeploymentBucketOutputs.FilesBucket, !Ref ADPS3BucketName ]
                  - !Sub
                    - 'arn:${AWS::Partition}:s3:::${SyncerDagS3Bucket}'
                    - SyncerDagS3Bucket: !GetAtt BasePlatformAirflow.SyncerDagS3Bucket
                  - !Sub
                    - 'arn:${AWS::Partition}:s3:::${SyncerDagS3Bucket}/*'
                    - SyncerDagS3Bucket: !GetAtt BasePlatformAirflow.SyncerDagS3Bucket

  DagsCopyLambdaSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Dummy SG for dags initializer lambda function
      VpcId: !GetAtt BasePlatformVPCOutputs.VPCID
      SecurityGroupEgress:
        - CidrIp: 0.0.0.0/0
          FromPort: -1
          IpProtocol: '-1'
          ToPort: -1

  DagsCopyLambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      Code:
        S3Bucket: !Sub
          - '${BucketName}'
          - BucketName: !If [RunningInAnotherRegion, !ImportValue ADPS3BucketAnotherRegion, !Ref ADPS3BucketName]
        S3Key: !Sub 'platform/${BasePlatformVersion}/functions/packages/batchInit/lambda.zip'
      Handler: batch_s3_init.lambda_handler
      Runtime: python3.7
      Timeout: 120
      Role: !GetAtt DagsCopyExecutionRole.Arn
      VpcConfig:
        SecurityGroupIds: [!Ref DagsCopyLambdaSecurityGroup]
        SubnetIds:
          - !GetAtt BasePlatformVPCOutputs.PrivateSubnet1AID
          - !GetAtt BasePlatformVPCOutputs.PrivateSubnet2AID
          - !GetAtt BasePlatformVPCOutputs.PrivateSubnet3AID
      Tags:
        - Key: 'Name'
          Value: 'DagsCopyLambdaFunction'
        - Key: 'Owner'
          Value: !ImportValue ADPPlatformOwner
        - Key: 'Project'
          Value: 'ADP'

  GetSecretExecution:
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt BasePlatformLambdaFunctionsOutputs.BPGetSecretLambdaArn
      Name: !GetAtt BasePlatformAirflow.AirflowSecret

  DagsInitLambda:
    DependsOn: BasePlatformKinesis
    Type: AWS::CloudFormation::CustomResource
    Properties:
      ServiceToken: !GetAtt DagsCopyLambdaFunction.Arn
      AirflowServiceName: "airflow"
      AirflowUsername: !GetAtt GetSecretExecution.username
      AirflowPassword: !GetAtt GetSecretExecution.password
      SourceBucket: !Sub
        - '${BucketName}'
        - BucketName: !If [RunningInAnotherRegion, !ImportValue ADPS3BucketAnotherRegion, !Ref ADPS3BucketName]
      SourcePrefix: !Sub "applications/adp/${UseCaseVersion}/dags/"
      TargetBucket: !GetAtt BasePlatformAirflow.SyncerDagS3Bucket
      TargetPrefix: "dags"
      Namespace: !ImportValue ADPResourcePrefix
      AirflowProperties: {
        "streaming_discovery": {
          "aws_conn_id": "adp_aws_default",
          "class": 'com.griddynamics.platform.aws.streaming.Main',
          "namespace": !ImportValue ADPResourcePrefix,
          "region": !Ref 'AWS::Region'
        },
        "streaming_process": {
          "input_file_path": !Ref InputFilePath,
          "jar_file_path": !Sub "s3://${ADPS3BucketName}/applications/adp/${UseCaseVersion}/jars/stream-pipeline-app.jar",
          "kinesis_stream_name": !GetAtt BasePlatformKinesis.KinesisStream,
          "main_class": "com.griddynamics.platform.aws.streaming.Main",
          "packages": "org.apache.spark:spark-streaming-kinesis-asl_2.12:3.0.1,io.github.spark-redshift-community:spark-redshift_2.12:5.0.3,org.apache.spark:spark-avro_2.12:3.0.1,org.apache.hadoop:hadoop-aws:3.1.1",
          "target_bucket": !Ref StreamingTargetBucket
        }
      }
      AirflowConnections: {
        "adp_aws_default": {
          "extra": {
            "region_name": !Ref 'AWS::Region'
          }
        }
      }
      Objects:
        - streaming_dags.zip

Outputs:
  AirflowELB:
    Value: !GetAtt BasePlatformAirflow.AirflowELB
    Description: Airflow service ELB address (should be enabled in Base Platform first)

  AirflowFQDN:
    Value: !GetAtt BasePlatformAirflow.AirflowFQDN
    Description: Airflow service DNS address (private zone will not be accessible form outside)

  AirflowSecret:
    Value: !GetAtt BasePlatformAirflow.AirflowSecret
    Description: Airflow service secret

  StreamingTargetBucket:
    Description: Target bucket for Streaming usecase.
    Value: !Ref StreamingTargetBucket
