The CloudFormation Registry is the central hub for discovering and using custom AWS resources.
Here’s how we can get our custom resource types published and discoverable through the Registry, using a simple example of a hypothetical MyCompany::Service::Database resource.
Let’s imagine we’ve developed a new database service that we want to expose as a CloudFormation resource.
First, we need to package our resource. This involves creating a .zip file containing our resource’s handler code (e.g., Python, Go, Java) and a schema.json file.
The schema.json is crucial. It defines the resource’s properties, including required inputs, optional inputs, and outputs.
{
"typeName": "MyCompany::Service::Database",
"description": "Manages a database instance in MyCompany's service.",
"sourceUrl": "https://github.com/mycompany/cfn-resources",
"version": "1.0.0",
"properties": {
"Engine": {
"type": "string",
"description": "The database engine type (e.g., 'mysql', 'postgres').",
"required": true
},
"InstanceClass": {
"type": "string",
"description": "The compute and memory capacity of the database instance.",
"required": true
},
"AllocatedStorage": {
"type": "integer",
"description": "The amount of storage in gibibytes.",
"required": false
}
},
"required": [
"Engine",
"InstanceClass"
],
"readOnlyProperties": [
"/properties/DatabaseName",
"/properties/Endpoint"
],
"createOnlyProperties": [
"/properties/Engine"
]
}
The sourceUrl points to the repository where the handler code lives, allowing users to inspect the implementation. version follows semantic versioning. readOnlyProperties are values that CloudFormation will output but not attempt to modify. createOnlyProperties are properties that, if changed, require the resource to be replaced.
Next, we need to build our handler code. For a Python handler, this would look something like this:
import boto3
import logging
import json
import uuid
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
def create(request):
props = request['desiredResourceState']
client = boto3.client('service-quotas') # Replace with your actual service client
try:
# Simulate creating a database instance
db_instance_id = f"db-{uuid.uuid4()}"
logger.info(f"Creating database instance: {db_instance_id}")
# In a real scenario, you'd call your actual service's API here
# For example: response = my_company_service.create_database(
# engine=props['Engine'],
# instance_class=props['InstanceClass'],
# storage=props.get('AllocatedStorage')
# )
# Simulate successful creation and return attributes
response_data = {
"DatabaseName": db_instance_id,
"Endpoint": f"{db_instance_id}.mycompany.com"
}
return {
"status": "SUCCESS",
"callbackContext": {},
"callbackDelaySeconds": 1,
"resourceId": db_instance_id,
"output": response_data
}
except Exception as e:
logger.error(f"Error creating database: {e}")
return {
"status": "FAILED",
"message": f"Failed to create database: {str(e)}"
}
def read(request):
resource_id = request['physicalResourceId']
logger.info(f"Reading database instance: {resource_id}")
# Simulate reading the database instance
# In a real scenario, you'd call your actual service's API here
# For example: db_details = my_company_service.get_database(resource_id)
# Simulate finding the resource
response_data = {
"DatabaseName": resource_id,
"Endpoint": f"{resource_id}.mycompany.com",
"Engine": "mysql", # Example, would fetch from actual service
"InstanceClass": "db.t3.medium" # Example
}
return {
"status": "SUCCESS",
"output": response_data
}
def update(request):
resource_id = request['physicalResourceId']
props = request['desiredResourceState']
old_props = request['previousResourceState']
logger.info(f"Updating database instance: {resource_id}")
# Check for createOnlyProperties changes
if props.get('Engine') != old_props.get('Engine'):
return {
"status": "FAILED",
"message": "Engine cannot be updated. Resource must be replaced."
}
try:
# Simulate updating the database instance properties
# For example: my_company_service.update_database(
# resource_id,
# instance_class=props['InstanceClass'],
# storage=props.get('AllocatedStorage')
# )
# Simulate successful update and return attributes
response_data = {
"DatabaseName": resource_id,
"Endpoint": f"{resource_id}.mycompany.com",
"Engine": props.get('Engine', 'mysql'),
"InstanceClass": props.get('InstanceClass', 'db.t3.medium')
}
return {
"status": "SUCCESS",
"output": response_data
}
except Exception as e:
logger.error(f"Error updating database: {e}")
return {
"status": "FAILED",
"message": f"Failed to update database: {str(e)}"
}
def delete(request):
resource_id = request['physicalResourceId']
logger.info(f"Deleting database instance: {resource_id}")
try:
# Simulate deleting the database instance
# For example: my_company_service.delete_database(resource_id)
pass # Deletion successful
return {
"status": "SUCCESS"
}
except Exception as e:
logger.error(f"Error deleting database: {e}")
return {
"status": "FAILED",
"message": f"Failed to delete database: {str(e)}"
}
The handler code needs to implement create, read, update, and delete functions. These functions receive a request dictionary and must return a dictionary indicating SUCCESS or FAILURE, along with any relevant output or resourceId.
Once the handler code and schema.json are ready, zip them together. For example, if your handler is handler.py, you’d run:
zip MyCompany-Service-Database.zip handler.py schema.json
Now, you can use the AWS CLI to register this as a private extension. You’ll need to specify the region, the .zip file, and the type name.
aws cloudformation register-type \
--region us-east-1 \
--type RESOURCE \
--type-name MyCompany::Service::Database \
--schema-upload-body file://schema.json \
--logging-config '{"logGroupName":"/aws/cloudformation/MyCompany-Service-Database","logRoleArn":"arn:aws:iam::123456789012:role/CloudFormationRegistryRole"}' \
--auto-publish-type \
--handler-package fileb://MyCompany-Service-Database.zip
Here’s a breakdown of the important parameters:
--type RESOURCE: Specifies that this is a resource type.--type-name MyCompany::Service::Database: Matches thetypeNamein yourschema.json.--schema-upload-body file://schema.json: Uploads your schema.--logging-config: Essential for debugging. It specifies a CloudWatch Log Group and an IAM role that CloudFormation will assume to write logs. The role needsCreateLogStreamandPutLogEventspermissions for the specified log group.--auto-publish-type: This is key. It automatically publishes the type to the CloudFormation Registry once it’s successfully registered and activated.--handler-package fileb://MyCompany-Service-Database.zip: Uploads your handler code.
The register-type command initiates a process. CloudFormation will extract your handler, upload it to an S3 bucket, and then invoke your handler’s create function (or delete for rollback) to validate it. If successful, and --auto-publish-type is used, it will then be available in the CloudFormation Registry.
After registration and publishing, you can use your custom resource in a CloudFormation template:
Resources:
MyDatabase:
Type: MyCompany::Service::Database
Properties:
Engine: mysql
InstanceClass: db.t3.medium
AllocatedStorage: 20
When CloudFormation processes this template, it will call your handler’s create function to provision the database. The output from your handler’s create function becomes available as attributes of the MyDatabase resource.
One aspect often overlooked is the IAM role specified in --logging-config. This role is not just for logging; it’s also the identity CloudFormation uses to execute your handler code. Therefore, this role must have permissions to interact with your actual backend service (e.g., to create, read, update, and delete databases in MyCompany’s service). If your handler code uses AWS SDKs to interact with other AWS services, this role needs those permissions too.
The next step is to version your custom resource by updating the version in schema.json and repeating the register-type process. CloudFormation will then create a new version of your resource type.