---
AWSTemplateFormatVersion: "2010-09-09"
Description: >
  Example for AWS Blog post "Collecting Apache Flink metrics in Amazon CloudWatch with Agents".

  WARNING This template creates IAM role, IAM instance profile, SSM Parameter, and EMR cluster. 
  The cluster will download a Flink library from Maven and start a Flink application. You will 
  be billed for the AWS resources used if you create a stack from this template.

Parameters:
  InstanceType:
    Type: String
    Default: m4.xlarge
    Description: Type of instances for all instance groups (https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-supported-instance-types.html)

  InstanceCountCore:
    Type: Number
    Default: 2
    MinValue: 1
    MaxValue: 5
    Description: Number of instances in core instance group

  EMRReleaseLabel:
    Type: String
    Default: emr-5.32.0
    Description: EMR release label (https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-release-components.html)

  BootstrapScriptPath:
    Type: String
    Description: S3 path of CloudWatch Agent installation bootstrap script
    AllowedPattern : "^s3://([^/]+)/(.*?([^/]+))$"
    ConstraintDescription: "BootstrapScriptPath must match pattern s3://bucket/folder/.../file"

  Subnet:
    Type: AWS::EC2::Subnet::Id
    Description: EC2 subnet to launch cluster into

  EC2KeyPairName:
    Type: String
    Description: Optional EC2 keypair for connecting to cluster nodes, as an alternative to Session Manager

Conditions:
  UseKeyPair:
    Fn::Not:
    - Fn::Equals:
      - {Ref: EC2KeyPairName}
      - ''

Resources:

  DemoFlinkClusterInstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      Roles: [{Ref: DemoFlinkClusterEMRRole}]

  DemoFlinkClusterEMRRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - 'ec2.amazonaws.com'
            Action:
              - 'sts:AssumeRole'
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role'
        - 'arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore'
        - 'arn:aws:iam::aws:policy/CloudWatchAgentServerPolicy'

  DemoFlinkCluster:
    Type: AWS::EMR::Cluster
    DependsOn:
      - CloudWatchAgentConfigSSMParam
    Properties:
      Name: 'Demo Flink Metrics'
      ReleaseLabel: 
        Ref: EMRReleaseLabel
      Applications:
        - Name: flink
        - Name: ganglia
      Configurations:
        - Classification: hadoop-env
          Configurations:
            # export JAVA_HOME
            - Classification: export
              ConfigurationProperties:
                JAVA_HOME: /usr/lib/jvm/java-11-amazon-corretto.x86_64
        - Classification: flink-conf
          ConfigurationProperties:
            # Configure servers
            jobmanager.web.address: 0.0.0.0
            rest.address: 0.0.0.0
            # Prevent error:
            #   org.apache.flink.configuration.IllegalConfigurationException: Either required fine-grained memory (jobmanager.memory.heap.size), or Total Flink Memory size (Key: 'jobmanager.memory.flink.size' , default: null (fallback keys: [])), or Total Process Memory size (Key: 'jobmanager.memory.process.size' , default: null (fallback keys: [])) need to be configured explicitly.
            jobmanager.heap.size: 1024m
            taskmanager.memory.process.size: 1728m
            # Configure Flink StatsD metric reporter
            metrics.reporters: stsd
            metrics.reporter.stsd.class: org.apache.flink.metrics.statsd.StatsDReporter
            metrics.reporter.stsd.host: localhost
            metrics.reporter.stsd.port: '8125'
            # Configure Flink metrics scope strings
            #   https://ci.apache.org/projects/flink/flink-docs-stable/ops/metrics.html#system-scope
            metrics.scope.jm: jobmanager
            metrics.scope.jm.job: jobmanager.<job_id>
            metrics.scope.tm: taskmanager.<tm_id>
            metrics.scope.tm.job: taskmanager.<tm_id>.<job_id>
            metrics.scope.task: taskmanager.<tm_id>.<job_name>.<task_name>.<subtask_index>
            metrics.scope.operator: taskmanager.<tm_id>.<job_name>.<operator_name>.<subtask_index>
          Configurations: []
      Instances:
        CoreInstanceGroup:
          InstanceCount:
            Ref: InstanceCountCore
          InstanceType: 
            Ref: InstanceType
          Name: 'Core Group'
        Ec2SubnetId:
          Ref: Subnet
        Ec2KeyName:
          Fn::If:
            - UseKeyPair
            - Ref: EC2KeyPairName
            - Ref: AWS::NoValue
        KeepJobFlowAliveWhenNoSteps: True
        MasterInstanceGroup:
          InstanceCount: 1
          InstanceType:
            Ref: InstanceType
          Name: "Management Group"
        TerminationProtected: False
      JobFlowRole: { Ref: DemoFlinkClusterInstanceProfile}
      ServiceRole: "EMR_DefaultRole"
      VisibleToAllUsers: True
      Tags: 
      - Key: "DemoFlinkMetrics"
        Value: ''
      BootstrapActions:
        - Name: "Install CloudWatch Agent"
          ScriptBootstrapAction:
            Path: { Ref: BootstrapScriptPath}

  AddLibrariesStep:
    # https://docs.aws.amazon.com/emr/latest/ReleaseGuide/flink-jobs.html
    Type: AWS::EMR::Step
    Properties:
      Name: AddLibraries
      ActionOnFailure: CONTINUE
      HadoopJarStep:
        Jar: "command-runner.jar"
        Args:
        - "sudo"
        - "wget"
        - "https://repo1.maven.org/maven2/org/apache/flink/flink-metrics-statsd/1.11.2/flink-metrics-statsd-1.11.2.jar"
        - "-P"
        - "/usr/lib/flink/lib/"
      JobFlowId: {Ref: DemoFlinkCluster}

  FlinkStartStep:
    # https://docs.aws.amazon.com/emr/latest/ReleaseGuide/flink-jobs.html
    DependsOn:
    - AddLibrariesStep
    Type: AWS::EMR::Step
    Properties:
      Name: FlinkStart
      ActionOnFailure: CONTINUE
      HadoopJarStep:
        Jar: "command-runner.jar"
        Args:
        - "bash"
        - "-c"
        - " flink-yarn-session -d -n 2"
      JobFlowId: {Ref: DemoFlinkCluster}

  CloudWatchAgentConfigSSMParam:
    # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-Configuration-File-Details.html
    # https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch-Agent-custom-metrics-statsd.html
    Type: AWS::SSM::Parameter
    Properties:
      Name: AmazonCloudWatch-Config.json
      Type: String
      Value: |
        { "agent": {
          "logfile": "/opt/aws/amazon-cloudwatch-agent/logs/amazon-cloudwatch-agent.log",
          "debug": false,
          "run_as_user": "cwagent"
          },
          "metrics":{
              "namespace": "FlinkSystem",
              "metrics_collected":{
                 "statsd":{
                    "service_address":":8125",
                    "metrics_collection_interval":60,
                    "metrics_aggregation_interval":300
                 }
              },
              "aggregation_dimensions" : [["InstanceId"],[]]
          }
        }

