AWS Developer Tools Blog
Serverless Service Discovery: Part 2: Lookup
In this, the second part of our serverless service discovery series, we will use Amazon DynamoDB to store information about the services in our service discovery service and update our AWS Lambda function to read information from the DynamoDB table.
Updating Amazon API Gateway and AWS Lambda
As part of adding new functionality to what we are implementing, we need to update the Service Discovery API and the code in API Gateway and AWS Lambda. First, we update the code in AWS Lambda:
def get_function_arn(function_name):
"""Return ARN given the Function Name.
:param function_name: The name of the Lambda function.
:return: The ARN for the Lambda function.
"""
client = boto3.client('lambda')
response = client.get_function(
FunctionName=function_name
)
return response['Configuration']['FunctionArn']
def update_lambda_function(package_name, function_name):
"""Update a Lambda function from zip-file.
:param package_name: The name of the package. Full or relative path.
:param function_name: The name of the Lambda function.
:return: The ARN for the Lambda function.
"""
with open(package_name, "rb") as package_file:
package_data = package_file.read()
# connect to Lambda API
client = boto3.client('lambda')
# update the function code
client.update_function_code(
FunctionName=function_name,
ZipFile=package_data,
Publish=True
)
# get function configuration to get top level ARN
return get_function_arn(function_name)
The get_function_arn returns the top level ARN for the Lambda function. When we update the code, we get the ARN for the version that was uploaded, but we need to use the top level ARN with Swagger.
The update_lambda_function updates the code of the Lambda function only. There are other functions in the Lambda API to update other configurations for Lambda functions.
Next we update the API with Swagger:
def update_api(swagger_file_name):
"""Update an API defined in Swagger.
:param swagger_file_name: The name of the swagger file.
Full or relative path.
:return: The id of the REST API.
"""
# get the API Gateway ID of the existing API
rest_api_name = get_rest_api_name(swagger_file_name)
client = boto3.client('apigateway')
paginator = client.get_paginator('get_rest_apis')
rest_api_id = ""
for response in paginator.paginate():
for item in response["items"]:
if (rest_api_name == item["name"]):
rest_api_id = item["id"]
with open(swagger_file_name, "r") as swagger_file:
swagger_data = swagger_file.read()
response = client.put_rest_api(restApiId=rest_api_id,
body=swagger_data)
return response['id']
We first query API Gateway for the REST API identifier based on the name in the Swagger file. We need to submit this identifier for the update to work.
We can use the same deployment method we used in the first blog post to deploy the update to the API Gateway stage. If the stage already exists, it will just be updated.
Creating a DynamoDB Table
We need to be able to persistently store the information about a service, and we need to be able to create, update, and query that information. DynamoDB is a fully managed serverless service that works well with AWS Lambda. We will start by creating a table for our discovery service:
dynamodb = boto3.resource('dynamodb', region_name = 'us-east-1')
# create the table
table = dynamodb.create_table(
TableName='Services',
KeySchema=[ { 'AttributeName': 'name',
'KeyType': 'HASH' },
{ 'AttributeName': 'version',
'KeyType': 'RANGE' } ],
AttributeDefinitions=[ { 'AttributeName': 'name',
'AttributeType': 'S' },
{ 'AttributeName': 'version',
'AttributeType': 'S' }, ],
ProvisionedThroughput={ 'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10 } )
# wait for the table to be ready
# this will block until the table is ACTIVE
table = boto3.resource('dynamodb').Table('Services')
table.wait_until_exists()
# insert some test data
with table.batch_writer() as batch:
batch.put_item(Item={
'name': 'testservice1',
'version': '1.0',
'endpoint_url': 'notarealurl1',
'ttl': 300,
'status': 'healthy' })
batch.put_item(Item={
'name': 'testservice2',
'version': '1.0',
'endpoint_url': 'notarealurl2',
'ttl': 600,
'status': 'healthy' })
The only attributes that we need to define are the keys. We will store additional attributes, but we can add them as we store items in the table.
We are using the name and version as the key because that matches our access pattern.
We are provisioning 10 read and write capacity units for the table. The number can be adjusted, depending on the amount of traffic the service receives and the effectiveness of the client caching.
After the table is created and active, we then prepare for the testing of our lookup service by inserting a couple of test records.
Looking Up Service Information from a DynamoDB Table
We are now ready to update our Lambda function so we can start using our DynamoDB table:
def lambda_handler(api_parameters, context):
"""Lambda hander for service lookup."""
logger.info("lambda_handler - service_name: %s"
" service_version: %s"
% (api_parameters["service_name"],api_parameters["service_version"]))
table = boto3.resource('dynamodb',region_name='us-east-1').Table('Services')
dynamodb_response = table.get_item(
Key={
'name': str(api_parameters["service_name"]),
'version': str(api_parameters["service_version"])
}
)
if ('Item' in dynamodb_response):
logger.info("found service with: %s" %
(dynamodb_response['Item']['endpoint_url'],))
return {
"endpoint_url": dynamodb_response['Item']['endpoint_url'],
"ttl": dynamodb_response['Item']['ttl'],
"status": dynamodb_response['Item']['status']
}
else:
raise Exception('NotFound')
The function gets the item from the table, and then returns a JSON object with the information to the client.
Notice that we throw an exception if we don’t find a record in DynamoDB. We can use this exception in API Gateway to map to a 404 HTTP code. We update the two response sections in the Swagger file to make that happen:
{
"swagger": "2.0",
"info": {
"title": "catalog_service",
"version": "1.0.0"
},
"basePath": "/v1",
"schemes": ["https"],
"consumes": ["application/json"],
"produces": ["application/json"],
"paths": {
"/catalog/{serviceName}/{serviceVersion}": {
"parameters": [{
"name": "serviceName",
"in": "path",
"description": "The name of the service to look up.",
"required": true,
"type": "string"
},
{
"name": "serviceVersion",
"in": "path",
"description": "The version of the service to look up.",
"required": true,
"type": "string"
}],
"get": {
"responses": {
"200": {
"description": "version information"
},
"404": {
"description": "service not found"
}
},
"x-amazon-apigateway-integration": {
"type": "aws",
"uri": "arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/$catalog_serviceARN$/invocations",
"httpMethod": "POST",
"requestTemplates": {
"application/json": "{\"service_name\": \"$input.params('serviceName')\",\"service_version\": \"$input.params('serviceVersion')\"}"
},
"responses": {
"default": {
"statusCode": "200",
"schema": {
"$ref": "#/definitions/CatalogServiceModel"
}
},
".*NotFound.*": {
"statusCode": "404",
"responseTemplates" : {
"application/json": "{\"error_message\":\"Service Not Found\"}"
}
}
}
}
}
}
},
"definitions": {
"CatalogServiceModel": {
"type": "object",
"properties": {
"endpoint_url": {
"type": "string"
},
"ttl": {
"type": "integer"
},
"status": {
"type": "string"
}
},
"required": ["endpoint_url", "ttl", "status"]
}
}
}
We use a regular expression (.*NotFound.*) under x-amazon-apigateway-integration -> responses to catch our exception and map it to a static JSON message.
We can now put everything together and update the code and API:
create_deployment_package("/tmp/catalog_service.zip", ["catalog_service.py"])
function_arn = update_lambda_function(
"/tmp/catalog_service.zip",
"catalog_service")
replace_instances_in_file("swagger.json",
"/tmp/swagger_with_arn.json",
"$catalog_serviceARN$", function_arn)
api_id = update_api("/tmp/swagger_with_arn.json")
deploy_api(api_id, "/tmp/swagger_with_arn.json", "dev")
We can use the following to test our deployment:
request_url="https://yourrestapiid.execute-api.us-east-1.amazonaws.com/"\
"dev/catalog/testservice2/1.0"
response = requests.get(request_url)
json_response = json.loads(response.content)
logging.info("Endpoint URL: %s" % (json_response['endpoint_url'],))
logging.info("TTL: %i" % (json_response['ttl'],))
logging.info("Status: %s" % (json_response['status'],))
That should give us the following results:
INFO:root:Endpoint URL: notarealurl2
INFO:root:TTL: 600
INFO:root:Status: healthy