DEV Community: Stephen Bawks

Transit Gateway Flow Logs

Stephen Bawks — Thu, 12 Oct 2023 18:26:24 +0000

This one was not very well documented and I really struggled a little bit figuring this one out. Adding this to the collective internet intelligence in the hope that it will help someone out one day.... or just my future self (hello future self!).

Adding Flow Logs to the AWS Transit Gateway was honestly easy enough to do. I pointed them to S3 and they started showing up. Easy done, nope. Querying the logs was a different story. Took me more than a day to get this all working the way I expected.

Here is what I landed on


resource "aws_ec2_transit_gateway" "main" {
  description = "Super Awesome Transit Gateway"

  amazon_side_asn  = var.amazon_side_asn
  dns_support      = var.dns_support
  vpn_ecmp_support = var.vpn_ecmp_support

  default_route_table_association = var.default_route_table_association
  default_route_table_propagation = var.default_route_table_propagation

  auto_accept_shared_attachments = var.auto_accept_shared_attachments
}

resource "aws_s3_bucket" "flow_logs" {
  bucket = "tgw-flow-logs-super-awesome-tgw"
}

resource "aws_flow_log" "main" {
  log_destination          = aws_s3_bucket.flow_logs.arn
  log_destination_type     = "s3"
  traffic_type             = "ALL"
  transit_gateway_id       = aws_ec2_transit_gateway.main.id
  max_aggregation_interval = 60

  destination_options {
    file_format                = "parquet"
    hive_compatible_partitions = true
    per_hour_partition         = true
  }
}

You can see that I landed on Parquet being the output format and I ended up also turning on Hive Compatible partitions because it makes queries in Athena quite a bit more efficient.

Getting to the important parts, which was correctly setting up Athena and the queries for it. Below is a continuation of the same Terraform module adding in the important bits for Athena.

Created a bucket and then an Athena workgroup to save the queries and results.

resource "aws_s3_bucket" "athena_output" {
  bucket = "athena-${aws_ec2_transit_gateway.main.id}"
}

resource "aws_athena_workgroup" "tgw_workgroup" {
  name = "TransitGatewayFlowLogs"

  configuration {
    enforce_workgroup_configuration    = true
    publish_cloudwatch_metrics_enabled = true
    result_configuration {
      output_location = "s3://${aws_s3_bucket.athena_output.bucket}/output/"

      encryption_configuration {
        encryption_option = "SSE_S3"
      }
    }
  }
}

Now we get to the juicy parts. Once again, below is also the same module but adding in the Athena queries. I have gone and created three different queries.

Creating the Flow Logs table in Athena.
Update the Metadata in the Athena Table for the correct Hive Table partitions for the Parquet format.
Query the Flow Logs for a specific VPC ID to see what its talking to. This one you may want to tailor a bit to fit your needs so your mileage may vary.

resource "aws_athena_named_query" "create_flow_table" {
  name        = "Create Flow Logs Table"
  workgroup   = aws_athena_workgroup.tgw_workgroup.name
  description = "Create the flow logs table in Athena"
  database    = aws_athena_database.flow_logs.name
  query       = <<EOT
CREATE EXTERNAL TABLE IF NOT EXISTS tgwflowlogs (
  version int,
  resource_type string,
  account_id string,
  tgw_id string,
  tgw_attachment_id string,
  tgw_src_vpc_account_id string,
  tgw_dst_vpc_account_id string,
  tgw_src_vpc_id string,
  tgw_dst_vpc_id string,
  tgw_src_subnet_id string,
  tgw_dst_subnet_id string,
  tgw_src_eni string,
  tgw_dst_eni string,
  tgw_src_az_id string,
  tgw_dst_az_id string,
  tgw_pair_attachment_id string,
  srcaddr string,
  dstaddr string,
  srcpor int,
  dstport int,
  protocol bigint,
  packet bigint,
  bytes bigint,
  start bigint,
  `end` bigint,
  log_status string,
  type string,
  packets_lost_no_route bigint,
  packets_lost_blackhole bigint,
  packets_lost_mtu_exceeded bigint,
  packets_lost_ttl_expired bigint,
  tcp_flags int,
  region string,
  flow_direction string,
  pkt_src_aws_service string,
  pkt_dst_aws_service string
)
PARTITIONED BY (
  `aws-account-id` string,
  `aws-service` string,
  `aws-region` string,
  `year` string,
  `month` string,
  `day` string,
  `hour` string
)
ROW FORMAT SERDE
  'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat'
OUTPUTFORMAT
  'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat'
LOCATION
  's3://${aws_s3_bucket.flow_logs.id}/AWSLogs/'
TBLPROPERTIES (
  'EXTERNAL'='true',
  'skip.header.line.count'='1'
)
  EOT
}

resource "aws_athena_named_query" "update_metadata" {
  name        = "Update Metadata in Catalog"
  workgroup   = aws_athena_workgroup.tgw_workgroup.name
  description = "Update the metadata in the catalog after you add Hive compatible partitions"
  database    = aws_athena_database.flow_logs.name
  query       = "MSCK REPAIR TABLE tgwflowlogs"
}

resource "aws_athena_named_query" "vpc_query" {
  name        = "VPC ID Query"
  workgroup   = aws_athena_workgroup.tgw_workgroup.name
  description = "Select the VPC ID, source IP address, destination IP address, and total bytes transferred for each flow log record."
  database    = aws_athena_database.flow_logs.name
  query       = <<EOT
SELECT tgw_src_vpc_id AS "VPC ID",
  srcaddr AS "Source Address",
  dstaddr AS "Destination Address",
  dstport as "Destination Port",
  SUM(bytes) AS "Total Bytes"
FROM "${aws_athena_database.flow_logs.name}".tgwflowlogs
WHERE tgw_src_vpc_id = 'replace-vpc-id-here'
GROUP BY tgw_src_vpc_id,
  srcaddr,
  dstaddr,
  dstport
LIMIT 100
  EOT
}

Once you deploy this it will set up your TGW and attach the flow logs. Those logs will be outputted to the S3 bucket in Parquet format.

You should be able to go into Athena and in the Query Editor, you will see the Workgroup dropdown menu where you can change that to be the workgroup that was created via the Terraform above. It is named TransitGatewayFlowLogs (assuming you did not change it).

Once you select that workgroup, you will then be able to go to the Saved Queries tab and see the three queries that were also created from the Terraform.

You should then select The Create Flow Logs Table and run that query. That will create the Athena table for you.

Once you do that, go back to the Saved Queries and select the Update Metadata in Catalog and run that. That will make sure the table is set up correctly for the Parquet Hive partitions.

After you do that, you can then load up the last saved query VPC ID Query. One important note about that last query. In there, I have placed a placeholder replace-vpc-id-here bit where you will want to put in a real VPC ID.

Once you do that, you should be able to query the Transit Gateway Flow logs.

VPC Layouts .... Easy Way

Stephen Bawks — Thu, 20 Apr 2023 11:53:35 +0000

Choosing the "correct" AWS VPC layout can sometimes be pretty tricky. You cannot make many changes of any real value once you pick the size of your VPC and where you decide to break apart your subnets. On top of that, VPC layouts and subnetting can be difficult.

I created something similar to this but with logic all built into a Terraform module. However, trying to put any complex logic in Terraform can be a test of will power and often feels like a square-peg-round-hole situation. Recently I was trying to update it with some additional logic to make it a little more flexible. There were a few curse words used, then finally I took a step back to ask myself if there could be a better way to do this.

I present my solution and presenting it to you with the hope that it could help other people with their VPC design and layouts. I would love to get some feedback as well.

API Getting Started

POST: https://api.rusticators.dev/v1/vpc

Body

There are only three required parameters you need to pass in with your request body. There is a fourth optional value that defaults to false if you do not specify it but we will get to that below.

{
    "vpc_type": "a",
    "cidr_block": "10.144.0.0",
    "subnet_mask": 22
}

Important Notes

VPC Types available so far are a and b.
The VPC subnet size can be anything between a /16 all the way through a /24.

API Response

{
    "public": [
        "10.144.0.0/26",
        "10.144.0.64/26",
        "10.144.0.128/26"
    ],
    "private": [
        "10.144.1.0/24",
        "10.144.2.0/24",
        "10.144.3.0/24"
    ]
}

The way this works is that you can use any 10.x range to create a VPC and it will calculate the appropriate subnet layout depending on what VPC Type you select.

There are a few different layouts that I have found to work well. The catch with subnet math is that most of the time it does not play well with being able to use the full amount of IP space. Add in the fact that each subnet in AWS has an overhead of 5 IPs that are unusable as they are reserved for AWS networking reasons. There is a little trade-off or balance between trying to achieve the most amount of unusable space while also providing a VPC layout that offers up some protection for "blast radius" and general high availability best practices.

Typically I generally recommend a VPC with three availability zones. This would be a VPC Type A. However there are a few locations where three is not always possible. I have also included a VPC layout that is built around 2 availability zones. This would be VPC Type B

How Can I Use This?

So you may be asking, how does this relate back to Terraform? Here is where I propose how this can then be pulled back in Terraform and consumed as a resource.

data "http" "vpc_layout" {
  url    = "https://api.rusticators.dev/v1/vpc"
  method = "POST"

  request_body = jsonencode({
    "vpc_type"    = var.vpc_type,
    "cidr_block"  = var.cidr_block,
    "subnet_mask" = var.subnet_mask,
    "ephemeral"   = var.ephemeral_subnet
  })
}

resource "aws_vpc" "main" {
  cidr_block = "${var.cidr_block}/${var.subnet_mask}"
}


resource "aws_subnet" "public_subnets" {
  for_each   = toset(jsondecode(data.http.vpc_layout.response_body).public)
  vpc_id     = aws_vpc.main.id
  cidr_block = each.value
}

resource "aws_subnet" "private_subnets" {
  for_each   = toset(jsondecode(data.http.vpc_layout.response_body).private)
  vpc_id     = aws_vpc.main.id
  cidr_block = each.value
}

resource "aws_subnet" "private_subnets" {
  for_each   = toset(jsondecode(data.http.vpc_layout.response_body).private)
  vpc_id     = aws_vpc.main.id
  cidr_block = each.value
}

resource "aws_subnet" "ephemeral_subnet" {
  count      = var.ephemeral_subnet ? 1 : 0
  for_each   = toset(jsondecode(data.http.vpc_layout.response_body).ephemeral)
  vpc_id     = aws_vpc.main.id
  cidr_block = each. Value
}

The beauty of potentially doing it this way is that it just a simple for loop and no matter how many subnets you have it will create them all with the single resource.

Ephemeral Subnets

There is the extra boolean option for creating an ephemeral subnet. The subnet itself does not dissapear or go away. Named it that way because its intention is mainly for workloads or instances that may be for temporary operations or things you just need to spin up and spin down for testing. A classic example may be a jumpbox.

Secondary to this is that most of the time CIDR math does not always play nicely with the complete address space you have for your VPC. More often than not it leaves "extra" space that is tough to use in any meaningful format.

This is where the ephemeral subnets come from. The extra space that is left over but can still be utilized for something otherwise its going to be wasted.

With all the VPC layouts so far, there is only one ephemeral subnet that will be created. It is in only one availability zone so its highly recommended that you use it for temporary stuff since its not truly going to be "highly available".

They can be provisioned just by adding in the ephemeral attribute to the body of your request. It is optional of course and does not have to be specified and its false by default.

{
    "vpc_type": "a",
    "cidr_block": "10.144.0.0",
    "subnet_mask": 20,
    "ephemeral": true
}

Final Thoughts

I have tried quite a few different VPC layouts over the years and I have landed on these since its gives me the highest amount of IP's utilized inside the VPC with minimal amount of waste. With every subnet you create in AWS land there is always going to be 5 IP's that are used for networking purposes. The more subnets you create the more you lose just to overhead.

Going to be working on the formal documentation on this and hope to have that finished here soon so I will come back and update this when its out there.

The quest with this is a pretty simple one and I hope it can help someone out here in the future. I am curious if someone has thoughts so do not hesistate to message me and I am open to feedback.

Pulumi - Python - Docker

Stephen Bawks — Sun, 01 Jan 2023 14:59:57 +0000

I do love Pulumi and Python. Working on a few things lately and fought with the Docker build options, getting it working the way I wanted.

The Docker module for Pulumi does not have the greatest documentation and I felt the need to write this up real quick hopefully someone can find it useful.


import json

import pulumi
import pulumi_aws as aws
import pulumi_aws_native as aws_native
import pulumi_docker as docker

# ----------------------------------------------------------------
# Pull Stack Variables from Config File
# ----------------------------------------------------------------

CONFIG = pulumi.Config()
account_id = (aws.get_caller_identity()).account_id
region = (aws.get_region()).name



untagged_days = 14

life_cycle_policy = json.dumps({
    "rules": [
        {
            "rulePriority": 1,
            "description": "Expire images older than 30 days",
            "selection": {
                "tagStatus": "untagged",
                "countType": "sinceImagePushed",
                "countUnit": "days",
                "countNumber": untagged_days
            },
            "action": {
                "type": "expire"
            }
        }
    ]
})

# https://www.pulumi.com/registry/packages/aws-native/api-docs/ecr/repository/
ecr_repository = aws_native.ecr.Repository(
    "ecr-repository",
    image_scanning_configuration=aws_native.ecr.RepositoryImageScanningConfigurationArgs(
        scan_on_push=True
    ),
    lifecycle_policy=aws_native.ecr.RepositoryLifecyclePolicyArgs(
        lifecycle_policy_text=life_cycle_policy
    )
)

# https://www.pulumi.com/registry/packages/aws/api-docs/ecr/getauthorizationtoken/
ecr_token = aws.ecr.get_authorization_token()

# https://www.pulumi.com/registry/packages/docker/api-docs/image/
container = docker.Image(
    "build-container",
    image_name=ecr_repository.repository_uri,
    build=docker.DockerBuild(
        context="./src/api"
    ),
    registry = docker.ImageRegistry(
        server = ecr_repository.repository_uri.apply(lambda uri: uri.split("/")[0]),
        username = ecr_token.user_name,
        password = ecr_token.password,
    ),
    opts=pulumi.ResourceOptions(parent=ecr_repository)
)

Loops... where we are going we don't need loops

Stephen Bawks — Fri, 29 Jul 2022 00:46:42 +0000

Here is hopefully the first of many... on several different coding and IAC topics.

There is nothing against loops, however one of the main challenges with them is sometimes it can be especially difficult to iterate through when you may have several conditional if statements or different bits of logic happening in the loop.

Wanted to touch on map() in Python. As its one of my new favorite things to reduce some of the amount of code when it concerns needing a for loop.

The map() is defined as the following:

map(function, iterable[, iterable1, iterable2,..., iterableN])

With the map() function, you can see that there is two things going on here. There is a function and then an object that you are passing in. The object you are passing in can also be a list, tuple, etc.

In my example, I am taking in a list of scopes that is coming from a variables file that is being passed in to be consumed for an infrastructure as code module. I need to format the scopes in a certain fashion that will then be applied to an API Gateway.

In the example below I have a pretty simple amount of code that I have used to format, parse and structure a set of scopes that I needed to be applied to my API.

API_SCOPES = "read:stuff, write:stuff, delete:stuff"

Just to add in some extra protections where someone may have added some white space, aka spaces, into the string. I added an extra statement to remove any additional spacing and then I am spliting on commas.

['read:stuff', 'write:stuff', 'delete:stuff']

The result is a list that has all of the scopes from the original comma separated string.

API_SCOPES = "read:stuff, write:stuff, delete:stuff"
DOMAIN_NAME = "api.example.com"

def api_authorization_scopes(name: str) -> str:
    """
    Takes a string of scopes and returns a list of properly formatted scopes.

    Args:
        name (str): The scope string

    Returns:
        Optional[str]: Formatted scope string

    """
    return f"https://{DOMAIN_NAME}/{name}"

parsed_scopes = API_SCOPES.replace(' ', '').split(",")
allowed_oauth_scopes = map(api_authorization_scopes, parsed_scopes)

This results in a map object that I can output as a list and it looks like the following.

['https://api.example.com/read:stuff', 'https://api.example.com/write:stuff', 'https://api.example.com/delete:stuff']