template.yml

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Parameters:
  DstS3Region:
    Type: String
  DstS3Bucket:
    Type: String
  DstS3Prefix:
    Type: String
    Default: ""
  AthenaTableName:
    Type: String

  InvokeSnsTopicName:
    Type: String
    Default: ""
  DlqSnsTopicName:
    Type: String
    Default: ""

  AthenaDatabaseName:
    Type: String
    Default: ""
  HasAthenaDatabase:
    Type: String
    Default: ""

  LambdaRoleArn:
    Type: String
    Default: ""
  SrcS3Bucket:
    Type: String
    Default: "-"

  ConcurrentExecution:
    Type: Number
    Default: 5

Conditions:
  LambdaRoleRequired:
    Fn::Equals: [ { Ref: LambdaRoleArn }, "" ]
  IsDefaultInvokeSnsTopicName:
    Fn::Equals: [ { Ref: InvokeSnsTopicName }, "" ]
  IsDefaultDlqSnsTopicName:
    Fn::Equals: [ { Ref: DlqSnsTopicName }, "" ]
  AthenaDatabaseRequired:
    Fn::Equals: [ { Ref: HasAthenaDatabase }, "" ]

Resources:
  # --------------------------------------------------------
  # Lambda functions
  Converter:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: build
      Handler: main
      Runtime: go1.x
      CodeUri: build
      Timeout: 300
      MemorySize: 2048
      ReservedConcurrentExecutions:
        Ref: ConcurrentExecution
      Role:
        Fn::If: [ LambdaRoleRequired, {"Fn::GetAtt": LambdaRole.Arn}, {Ref: LambdaRoleArn} ]
      Environment:
        Variables:
          S3_REGION:
            Ref: DstS3Region
          S3_BUCKET:
            Ref: DstS3Bucket
          S3_PREFIX:
            Ref: DstS3Prefix
          ATHENA_DATABASE_NAME:
            Ref: AthenaDatabaseName
          ATHENA_TABLE_NAME:
            Ref: AthenaTableName
      Events:
        ReportLine:
          Type: SNS
          Properties:
            Topic:
              Ref: InvokeSnsTopic
      DeadLetterQueue:
        Type: SNS
        TargetArn:
          Ref: DlqSnsTopic

  InvokeSnsTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName:
        Fn::If:
          - IsDefaultInvokeSnsTopicName
          - {"Fn::Sub": "VPCFlowLogsParquet-Invoke-${AWS::StackName}"}
          - {"Ref": InvokeSnsTopicName}
  DlqSnsTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName:
        Fn::If:
          - IsDefaultDlqSnsTopicName
          - {"Fn::Sub": "VPCFlowLogsParquet-Dlq-${AWS::StackName}"}
          - {"Ref": DlqSnsTopicName}

  FlowLogDB:
    Type: AWS::Glue::Database
    Condition: AthenaDatabaseRequired
    Properties:
      DatabaseInput:
        Description: "Log Database"
        Name:
          Ref: AthenaDatabaseName
      CatalogId:
        Ref: AWS::AccountId

  FlowLogTable:
    Type: AWS::Glue::Table
    Properties:
      TableInput:
        Description: "FlowLog table"
        TableType: EXTERNAL_TABLE
        PartitionKeys:
          - Name: dt
            Type: string
          - Name: account
            Type: string
        StorageDescriptor:
          InputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat
          OutputFormat: org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat
          Columns:
            - Name: version
              Type: int
            - Name: accountid
              Type: string
            - Name: interfaceid
              Type: string
            - Name: srcaddr
              Type: string
            - Name: dstaddr
              Type: string
            - Name: srcport
              Type: int
            - Name: dstport
              Type: int
            - Name: protocol
              Type: int
            - Name: packets
              Type: int
            - Name: bytes
              Type: int
            - Name: start_ts
              Type: bigint
            - Name: end_ts
              Type: int
            - Name: action
              Type: string
            - Name: logstatus
              Type: string
          SerdeInfo:
            SerializationLibrary: org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe
          Location:
            Fn::Sub: "s3://${DstS3Bucket}/${DstS3Prefix}/"
        Name:
          Ref: AthenaTableName
      DatabaseName:
        Ref: AthenaDatabaseName
      CatalogId:
        Ref: AWS::AccountId

  # --------------------------------------------------------
  # IAM Roles
  LambdaRole:
    Type: AWS::IAM::Role
    Condition: LambdaRoleRequired
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: "Allow"
            Principal:
              Service: ["lambda.amazonaws.com"]
            Action: ["sts:AssumeRole"]
      Path: "/"
      ManagedPolicyArns:
        - "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
      Policies:
        - PolicyName: "S3Writable"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action:
                  - s3:GetObject
                  - s3:GetObjectVersion
                  - s3:ListBucket
                  - s3:PutObject
                  - s3:GetBucketLocation
                  - s3:ListBucketMultipartUploads
                  - s3:ListMultipartUploadParts
                  - s3:AbortMultipartUpload
                  - s3:CreateBucket
                Resource:
                  - Fn::Sub: [ "arn:aws:s3:::${bucket}", { bucket: { "Ref": DstS3Bucket } } ]
                  - Fn::Sub: [ "arn:aws:s3:::${bucket}/${prefix}*", { bucket: { "Ref": DstS3Bucket }, prefix: {"Ref": DstS3Prefix} } ]
        - PolicyName: "S3Readable"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: "Allow"
                Action:
                  - s3:GetObject
                  - s3:GetObjectVersion
                  - s3:ListBucket
                Resource:
                  - Fn::Sub: [ "arn:aws:s3:::${bucket}", { bucket: { "Ref": SrcS3Bucket } } ]
                  - Fn::Sub: [ "arn:aws:s3:::${bucket}/*", { bucket: { "Ref": SrcS3Bucket } } ]
        - PolicyName: "QueryExecutable"
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - glue:GetDatabase
                  - glue:UpdateDatabase
                  - glue:GetTable
                  - glue:BatchCreatePartition
                Resource:
                  - Fn::Sub: "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/${AthenaDatabaseName}"
                  - Fn::Sub: "arn:aws:glue:${AWS::Region}:${AWS::AccountId}:table/${AthenaDatabaseName}/*"
              - Effect: Allow
                Action:
                  - glue:GetDatabase
                  - glue:GetTable
                  - glue:BatchCreatePartition
                Resource:
                  - Fn::Sub: arn:aws:glue:${AWS::Region}:${AWS::AccountId}:catalog
              - Effect: Allow
                Action:
                  - glue:GetDatabase
                Resource:
                  - Fn::Sub: arn:aws:glue:${AWS::Region}:${AWS::AccountId}:database/default
              - Effect: Allow
                Action:
                  - athena:StartQueryExecution
                  - athena:GetQueryExecution
                Resource:
                  - "*"