Airflow: Glue Operator looses its region name

 The other day, I tried to run an AWS Glue script from our Airflow instance. Nothing fancy, it would just convert a parquet file to CSV between two S3 buckets. Looking for an operator to use, I found that there is indeed a Glue Operator. It looks pretty easy to configure, so I tried it out:

parquet2csv_glue = AwsGlueJobOperator(

    task_id='parquet2csv_glue',

    dag=parquet2csv_dag,

    job_name='glue-parquet2csv',

    region_name='eu-west-3',

    aws_conn_id=None,

    script_args={

        "--SOURCE_PATH": input_path,

        "--TARGET_PATH": output_path,

    }

)

After triggering the DAG, it would fail, telling me that I need to set the region name. Well, I thought I did!

It probably comes from the fact that recent boto3 versions require region to be set, while it was optional in the past. As airflow is open source, I decided to have a look at the code. As it turns out, the bug was not too hard to spot. The Glue Operator creates a Glue Hook, which declares this constructor:

class AwsGlueJobHook(AwsBaseHook):

    def __init__(

        self,

        s3_bucket: Optional[str] = None,

        job_name: Optional[str] = None,

        desc: Optional[str] = None,

        concurrent_run_limit: int = 1,

        script_location: Optional[str] = None,

        retry_limit: int = 0,

        num_of_dpus: int = 10,

        region_name: Optional[str] = None,

        iam_role_name: Optional[str] = None,

        *args,

        **kwargs,

    ):

        self.job_name = job_name

        self.desc = desc

        self.concurrent_run_limit = concurrent_run_limit

        self.script_location = script_location

        self.retry_limit = retry_limit

        self.num_of_dpus = num_of_dpus

        self.region_name = region_name

        self.s3_bucket = s3_bucket

        self.role_name = iam_role_name

        self.s3_glue_logs = 'logs/glue-logs/'

        kwargs['client_type'] = 'glue'

        super().__init__(*args, **kwargs)

The Hook derives from a base class, AwsBaseHook, that handles the common connection part for all AWS Hooks. The call to the constructor of the super class does not forward the region name. It should probably be called like this:

        super().__init__(region_name=region_name, *args, **kwargs)

I opened a bug report. But in the meanwhile, I still needed my code to work. So I found quite an ugly patch. There is probably better, but since I could see that the boto3 session was created in the AwsBaseHook class, and since our Airflow instance is running from an EC2 and I can just inherit its profile, I made this simple workaround:

import boto3

from airflow.providers.amazon.aws.hooks.base_aws import AwsBaseHook

# ugly patch to resolve the bug in the Glue Hook

def _get_credentials(self, region_name):

    return boto3.session.Session(region_name="eu-west-3"), None

AwsBaseHook._get_credentials = _get_credentials

Hopefully, it won't stay here long...

Group files by folders in Python

 I sometimes need to display a list of files coming in this format:

folder1/file1

folder2/file1

folder1/file2 ...

 And I want to display it in this format:

folder1

file1

file2

folder2

 file1

 Here is my code. It uses the groupby function from the itertools library:

from itertools import groupby

def format_files_by_folder(folder,filenames):

    return folder + "\n  " + "\n  ".join([f[1] for f in filenames])

def file_by_folder(file_list):

    files_and_folders = [(f.split('/')[0], '/'.join(f.split('/')[1:])) 

        for f in file_list]

    # Group by folder

    files_and_folders.sort(key=lambda f: f[0])

    files_by_folder = groupby(files_and_folders, lambda f: f[0])

    return "\n\n".join(

        [format_files_by_folder(folder, filenames) 

            for folder, filenames in files_by_folder])

Snake Case in Terraform

 How do you convert from camel case to snake case in Terraform? How do you go from "MyProjectName" to "my-project-name"? Here is a simple solution:

locals {

# convert to snake case

snake_case_name = lower(replace(var.camel_case_name, 

"/(\\w)([A-Z])/", "$${1}_$2"))

}

Basically, you add an underscore before each capital letters. The strange syntax with the double dollars is a workaround to a strange bug in Terraform, which considers '$1_' as meaning something. 

This line works well if you have simple cases like "MyDatabaseName". However, I had to handle some more complex cases, like "ABCProject", or "ProjectABC". In that case, you have to work a bit more.

My solution was to implement two replace functions:

• One for a series of capital letters anywhere in my word. I insert an underscore only if my capital letter is followed by a lower case.
• One for a series of capital letters at the end of my word. I insert an underscore before the first letter of the series.

This gives this more complex command:

locals {

  # convert to snake case

  snake_case_name = lower(replace(replace(var.camel_case_name,

      # add underscore before capital letter followed by lowcase

      "/(\\w)([A-Z][a-z])/", "$${1}_$2"),

      # add underscore before capital letters at the end of the word

      "/([A-Z]+)$/", "_$1"))

}

Work around EC2 Termination Protection in Jenkins Pipeline

Recently, we enabled Termination Protection on our EC2 instances on our AWS cloud. That means that it is not possible to accidentally switch off an instance, you have to switch off the PRotection flag first. In our Terraform files, it is easy to put in place:

resource "aws_instance" "myinstance" {
  ...
  disable_api_termination = true
}

The problem is, when we run our Terraform scripts, and a new instance has to be created instead of an old one (like when we modify user data), it won't let us. You'd have to manually remove the flag. Since we decided that running a Terraform script for deployment means that we really want to be able to terminate our instance, we decide to allow the Jenkins pipeline to remove the flag before running the deployment scripts.

In order to do this, we wrote a small Groovy script at the beginning of our pipeline:

def removeTerminationProtection(instanceName) {
    echo "Looking for '${instanceName}'"
    def instanceId = sh (script: "aws ec2 describe-instances --region ${AWS_REGION} --filters Name=tag:Name,Values=${instanceName} --query 'Reservations[0].Instances[0].InstanceId' | xargs echo -n", returnStdout: true)
    if ('null'.equals(instanceId)) {
        echo "Instance ${instanceName} not found."
    }
    else {
        echo "Removing termination protection for '${instanceId}'"
        sh (script: "aws ec2 modify-instance-attribute --instance-id ${instanceId} --region ${AWS_REGION} --disable-api-termination Value=false || exit 1")
    }
}

It uses AWS CLI to first find our instance by its name, and then disable termination protection. The way to call it from a pipeline stage is the following:

pipeline {
    ...
    stages {
        ...
        stage('Remove Termination Protection') {
            environment {
                AWS_PROFILE = "${PROFILE_NAME}"
            }

            steps {
                script {
                    removeTerminationProtection("myinstance")
                }
            }
        }
        ...
    }
}

Since we run the AWS CLI, we must have AWS credentials, so we do it using an AWS Profile. We run our Terraform script in a later stage.

S3 Multipart Upload of Memory Mapped File in Java

In AWS,when uploading files bigger than 5 GB, you have no choice bu to use a Multipart Upload. In the first versions of the AWS SDK in Java, you had a TransferManager class to handle all the low level bits for you. Unfortunately, I did not find any trace of it in the latest versions.
The low level operations of the Multipart Upload expect you to divide your file into parts, each time providing a buffer to the part you want to upload. The best way to go through a big file without reading it all in memory is to use a Memory Mapped File. Here is my code:

private void uploadMultiPartFile(String bucketName, Path path, String key) throws IOException {
    S3Client s3 = S3Client.builder().region(REGION).build();
 
    CreateMultipartUploadRequest createMultipartUploadRequest = CreateMultipartUploadRequest.builder()
       .bucket(bucketName)
       .key(key)
       .build();

    CreateMultipartUploadResponse response = s3.createMultipartUpload(createMultipartUploadRequest);
    String uploadId = response.uploadId();

    long position = 0;
    long fileSize = Files.size(path);
    int part = 1;
    List<CompletedPart> completedParts = new ArrayList<>();

    try (FileChannel channel = FileChannel.open(path, StandardOpenOption.READ)) {
        while (position < fileSize) {
            long remaining = fileSize - position;
            int toRead = (int) Math.min(BUFFER_SIZE, remaining);
            MappedByteBuffer map = channel.map(MapMode.READ_ONLY, position, toRead);

            UploadPartRequest uploadPartRequest = UploadPartRequest.builder()
                .bucket(bucketName)
                .key(key)
                .uploadId(uploadId)
                .partNumber(part)
                .build();
            String etag = s3.uploadPart(uploadPartRequest, RequestBody.fromByteBuffer(map)).eTag();
            CompletedPart completed = CompletedPart.builder().partNumber(part).eTag(etag).build();
            completedParts.add(completed);
   
            position += BUFFER_SIZE;
            part++;
        }
    }  
 
    CompletedMultipartUpload completedMultipartUpload = CompletedMultipartUpload.builder().parts(completedParts).build();
    CompleteMultipartUploadRequest completeMultipartUploadRequest = CompleteMultipartUploadRequest.builder()
        .bucket(bucketName)
        .key(key)
        .uploadId(uploadId)
        .multipartUpload(completedMultipartUpload)
        .build();
    s3.completeMultipartUpload(completeMultipartUploadRequest);
}

For my case, I set the BUFFER_SIZE to 100MB. The AWS API states that buffers can be up to 5GB, but I had the bad surprise of noticing that the buffer size parameter in the Java SDK is an int instead of a long, so you cannot use anything above 2GB. When I tried to use 2GB, the process would just hang, so I guess there must be other limitations in the OS.
However, 100MB is OK for me because it allows me to upload files up to 1TB (10000 parts is the maximum allowed). But you can play with this parameter since bigger numbers allow you to upload files faster. Another optimization would be to use a parallel stream to run several uploads in parallel, but I didn't try it.
Another piece of advice if you start playing with Multipart Uploads: if your upload fails at some point, the parts already uploaded are stored in your S3, but you cannot see them, also you pay for them. So do not forget to attach a lifecycle policy for aborted Multipart Uploads.

Use Terraform to transform a CSV file to JSON documents

I used this trick to ingest items into DynamoDB. Terraform is maybe not the best tool for that, but we already created the DynamoDB table with the tool, and needed to import a couple of items representing metadata at creation. So here it is.
I have a semi-colon separated CSV file, where each line contains data for a JSON document. Here is the Terraform code:

locals {
    content = file("myfile.csv")
    lines = split("\n", local.content)
    json_docs = [for item in local.lines: format(<<EOT
{
    "key1": {
      "S": "%s"
    },
    "key2": {
      "S": "%s"
    },
    "key3": {
      "S": "%s"
    }
}
EOT
    , split(";", item)...)]
}

resource "aws_dynamodb_table_item" "items" {
    count = length(local.json_docs)

    table_name = aws_dynamodb_table.mytable.name
    hash_key   = aws_dynamodb_table.mytable.hash_key

    item = local.json_docs[count.index]
}

I first copy the content of the CSV file into a String. I then split it along newline characters. I assume that it is a Unix style file. Then I have a loop, transforming each line into a String containing the JSON document. For this, I use the format function, with the template of the JSON document and my line split along the semi-colons as parameters. Notice that to expand my split line, which is a list, into arguments to the format function, I have to use the three periods (...) symbol.

Finally, I can import all those documents into my DynamoDB table.

Implement a Whitelist in Terraform

This happens sometimes that you need to implement a variable in Terraform, that can only take an acceptable list of values. In my case, it was a list of DNS names that needed to be accepted by a security team, and stored in a file on an S3.
The difficulty is to make the Terraform fail if you decide to use a bad value. There are several ways to do that, here is mine:

data "aws_s3_bucket_object" "white_list" {
  bucket = "my-bucket"
  key    = "my_white_list"
}

locals {
  value_to_check = "SomeValue"

  white_list = split(
    " ",
    replace(data.aws_s3_bucket_object.white_list.body, "/\\s+/", " "),
  )
  allowed = zipmap(local.white_list, local.white_list)[local.value_to_check]
}

The data part is fetching my file from an S3, but you can imagine using a simple file command, or even a hardcoded list.
Then I am setting the value to check, which is hardcoded here for the example, but it will typically be calculated or retrieved from some other place. I then create a Terraform list from the file, by removing any extra space and splitting the lines.
Finally, here is my way of making Terraform fail. I create a map from the white list, using the zipmap function, and get the value from it. If the value is not in the map, Terraform will just stop with an error.