# Writing my first lambda to trigger a step function

### I. What is a CDC?

CDC stands for change data capture. It is used to capture database updates for a kafka topic. To test out a topic, go to dev-ahkq.io and search for a kafka topic within that CDC-based cluster. Then click Produce to Topic and add sample message data in the value section. 

This example message structure comes from `stream.asset.mezzanine.audio.flac` which subscribes to database changes where an audio track is added or updated. The sample data was retrieved from Snowflake by querying this database.
 
 
 __Database__: ORCHARD_APP_REPORTING_V2 <br>
 __Schema__: QA_OWS_ASSETS_OWS_ASSETS


```
{
  "ASSET_FINAL_ID": 46564,
  "ASSET_UPLOAD_ID": 154263,
  "ASSET_TYPE": "FLAC",
  "ASSET_SUBTYPE": "none",
  "BUCKET": "qa-orcd-mezzanine-assets",
  "FILENAME": "ee8c275b_90f4_4a5c_bc7b_f7f8d6f94c7a.flac",
  "METADATA": null,
  "DURATION_MS": 0
}
```


### II. How do you test out your lambda with a CDC event?

First, you want to have your lambda present in the AWS dev console and associated with your latest docker image from the ECR repository. To do so, follow the instructions in [Section V](#v-create-your-lambda). Then you can produce sample data to the designated kafka topic in dev-akhq.io. Once that has been produced to the topic, head over to the dev console and find your lambda. It should have an MSK trigger associated with a CDC cluster, based on your lambda code and terraform.

Go to the Monitor tab and look at your recent invocations in the CloudWatch logs. Open one up, and in those log messages you can find the final event message that was used as input to your lambda. You can use it as a sample event to unit test your lambda code. 

```
{
    "message": null,
    "eventSource": "aws:kafka",
    "eventSourceArn": "arn:aws:kafka:us-east-1:103233932089:cluster/dev-managed-kafka-cdc-destination/194e93a4-1a4e-4e96-85bd-99b4ebce8df0-11",
    "bootstrapServers": "b-2.dev-managed-kafka-cdc.rk4es0.c11.kafka.us-east-1.amazonaws.com:9094,b-3.dev-managed-kafka-cdc.rk4es0.c11.kafka.us-east-1.amazonaws.com:9094,b-1.dev-managed-kafka-cdc.rk4es0.c11.kafka.us-east-1.amazonaws.com:9094",
    "records": {
        "stream.asset.mezzanine.audio.flac-0": [
            {
                "topic": "stream.asset.mezzanine.audio.flac",
                "partition": 0,
                "offset": 65,
                "timestamp": 1674677853258,
                "timestampType": "CREATE_TIME",
                "key": "",
                "value": "ewogICJBU1NFVF9GSU5BTF9JRCI6IDQ2NTY0LAogICJBU1NFVF9VUExPQURfSUQiOiAxNTQyNjMsCiAgIkFTU0VUX1RZUEUiOiAiRkxBQyIsCiAgIkFTU0VUX1NVQlRZUEUiOiAibm9uZSIsCiAgIkJVQ0tFVCI6ICJxYS1vcmNkLW1lenphbmluZS1hc3NldHMiLAogICJGSUxFTkFNRSI6ICJlZThjMjc1Yl85MGY0XzRhNWNfYmM3Yl9mN2Y4ZDZmOTRjN2EuZmxhYyIsCiAgIk1FVEFEQVRBIjogbnVsbCwKICAiRFVSQVRJT05fTVMiOiAwCn0=",
                "headers": []
            }
        ]
    }
}
```


### III. How do you initiate a state machine in your lambda code?

You use the boto3 API to [execute a step function](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions.html?highlight=start_execution#SFN.Client.start_execution) with desired input. (Boto3 is an AWS SDK for Python that is used to manage other AWS services.) Pass in your step function ARN, input data, and a name. Using a unique name allows for idempotency, so that your code does not execute the same message data more than once. It prevents duplicate execution of a message. 



### IV. How do you create a custom message type and logger?

This python library (A) has a BaseEventMessage that you can use to format your deserialized event message data. You can build a subclass of this to customize a desired message structure for your data. Look a this example here: [kafka_utils/consumer/message/product.py](https://github.com/theorchard/python-kafka-utils/blob/f43f111f9b97b8871d57688b1c80b1cfc5f71b44/kafka_utils/consumer/message/product.py)

This library (B) has a ContentLambdaLogger to log info and errors for your app. You can build a subclass of it to log data for your customized event message. Look at this example: [content_utils/logging/product_logging.py](https://github.com/theorchard/python-content-utils/blob/2be17418e6cb916e251ca8e5a1bd7745d1e0800d/content_utils/logging/product_logging.py)



#### Useful libraries for lambda code

A) Python-content-utils  <br>
To use ContentLambdaLogger, Elasticsearch and GraphQL connectors

B) Python-kafka-utils <br>
To help with parsing events and message data in a lambda



### V. Create your lambda

To build and push a docker image for your lambda function, you will need to login to the AWS CLI with your credentials.To do so, follow the instructions here: https://www.notion.so/Docker-Lambda-da3f86a36e294c24bbef2f2c0ced0b4a#5a1053c49fd345099dcf018bcdf8396f

Then look up your private ECR repository in the aws console, which shares the same name as your lambda. Click on View Push Commands. You can run those commands one by one from your lambda base repo. The docker image should be tagged as latest by default. You want to have Docker open locally and check that your lambda function exists before you do the push command. 

To create the lambda, the final stage is to terraform it. You want to have dev terraform configurations in order to build and test out your lambda in the dev AWS console. A lambda is created when you run `atlantis apply` on an approved terraform PR.