Menu
In aws api gateway there is a section called API Link and I can manually set that.
The problem is I cannot find any section in cloudformation documentation on how I can create vpc link via cloud formation on api gateway. Is it sth that cloudformation does not support or am I missing it?
848
To declare this entity in your AWS CloudFormation template, use the following syntax: { "Type" : "AWS::ApiGateway::VpcLink" , "Properties" : { "Description" : String , "Name" : String , "TargetArns" : [ String, ... ] } } A description of the VPC link. A name for the VPC link.
Requests from the API Gateway accounts are automatically approved in the VPC link creation process. This is because the AWS accounts that serve API Gateway for each Region are allow-listed in the VPC endpoint service.
A VPC link is a resource in Amazon API Gateway that allows for connecting API routes to private resources inside a VPC. A VPC link acts like any other integration endpoint for an API and is an abstraction layer on top of other networking resources. This helps simplify configuring private integrations.
Requests from the API Gateway accounts are automatically approved in the VPC link creation process. This is because the AWS accounts that serve API Gateway for each Region are allow-listed in the VPC endpoint service. When a Network Load Balancer is associated with an endpoint service, the traffic to the targets is sourced from the NLB.
You can use swagger to define an API Gateway using VPC Link. This is a complete CloudFormation template you can deploy to test it out...
{
"AWSTemplateFormatVersion": "2010-09-09",
"Description": "Test backend access via API Gateway. This template provisions a Regional API Gateway proxing requests to a backend via VPC Link and Direct Connect to on-premises resources using private ip addresses.",
"Parameters": {
"VPCId": {
"Description": "VPC Id for API Gateway VPC Link",
"Type": "AWS::EC2::VPC::Id"
},
"NLBSubnetList": {
"Type": "List<AWS::EC2::Subnet::Id>",
"Description": "Subnet Ids for provisioning load balancer supporting the VPC Link"
},
"BackendBaseEndpoint": {
"Description": "The backend service base url including protocol. e.g.: https://<url>",
"Type": "String",
"Default": "https://mybackend.dev.mycompany.com"
},
"TargetIpAddresses": {
"Type": "CommaDelimitedList",
"Description": "Comma separated list of NLB target ip addresses. Specify two entries.",
"Default": "10.78.80.1, 10.79.80.1"
}
},
"Resources": {
"API": {
"Type": "AWS::ApiGateway::RestApi",
"Properties": {
"Name": "Test Api",
"Description": "Test Api using VPC_LINK and AWS_IAM authorisation",
"Body": {
"swagger": "2.0",
"info": {
"title": "Test Api"
},
"schemes": [
"https"
],
"paths": {
"/{proxy+}": {
"x-amazon-apigateway-any-method": {
"parameters": [
{
"name": "proxy",
"in": "path",
"required": true,
"type": "string"
}
],
"responses": {},
"security": [
{
"sigv4": []
}
],
"x-amazon-apigateway-integration": {
"responses": {
"default": {
"statusCode": "200"
}
},
"requestParameters": {
"integration.request.path.proxy": "method.request.path.proxy"
},
"uri": {
"Fn::Join": [
"",
[
{
"Ref": "BackendBaseEndpoint"
},
"/{proxy}"
]
]
},
"passthroughBehavior": "when_no_match",
"connectionType": "VPC_LINK",
"connectionId": "${stageVariables.vpcLinkId}",
"httpMethod": "GET",
"type": "http_proxy"
}
}
}
},
"securityDefinitions": {
"sigv4": {
"type": "apiKey",
"name": "Authorization",
"in": "header",
"x-amazon-apigateway-authtype": "awsSigv4"
}
}
},
"EndpointConfiguration": {
"Types": [
"REGIONAL"
]
}
},
"DependsOn": "VPCLink"
},
"APIStage": {
"Type": "AWS::ApiGateway::Stage",
"Properties": {
"StageName": "dev",
"Description": "dev Stage",
"RestApiId": {
"Ref": "API"
},
"DeploymentId": {
"Ref": "APIDeployment"
},
"MethodSettings": [
{
"ResourcePath": "/*",
"HttpMethod": "GET",
"MetricsEnabled": "true",
"DataTraceEnabled": "true",
"LoggingLevel": "ERROR"
}
],
"Variables": {
"vpcLinkId": {
"Ref": "VPCLink"
}
}
}
},
"APIDeployment": {
"Type": "AWS::ApiGateway::Deployment",
"Properties": {
"RestApiId": {
"Ref": "API"
},
"Description": "Test Deployment"
}
},
"VPCLink": {
"Type": "AWS::ApiGateway::VpcLink",
"Properties": {
"Description": "Vpc link to GIS platform",
"Name": "VPCLink",
"TargetArns": [
{
"Ref": "NLB"
}
]
}
},
"NLBTargetGroup": {
"Type": "AWS::ElasticLoadBalancingV2::TargetGroup",
"Properties": {
"Name": "NLBTargetGroup",
"Port": 443,
"Protocol": "TCP",
"TargetGroupAttributes": [
{
"Key": "deregistration_delay.timeout_seconds",
"Value": "20"
}
],
"TargetType": "ip",
"Targets": [
{
"Id": { "Fn::Select" : [ "0", {"Ref": "TargetIpAddresses"} ] },
"Port": 443,
"AvailabilityZone": "all"
},
{
"Id": { "Fn::Select" : [ "1", {"Ref": "TargetIpAddresses"} ] },
"Port": 443,
"AvailabilityZone": "all"
}
],
"VpcId": {
"Ref": "VPCId"
},
"Tags": [
{
"Key": "Project",
"Value": "API and VPC Link Test"
}
]
}
},
"NLB": {
"Type": "AWS::ElasticLoadBalancingV2::LoadBalancer",
"Properties": {
"Type": "network",
"Scheme": "internal",
"Subnets": {
"Ref": "NLBSubnetList"
}
}
},
"NLBListener": {
"Type": "AWS::ElasticLoadBalancingV2::Listener",
"Properties": {
"DefaultActions": [
{
"Type": "forward",
"TargetGroupArn": {
"Ref": "NLBTargetGroup"
}
}
],
"LoadBalancerArn": {
"Ref": "NLB"
},
"Port": "443",
"Protocol": "TCP"
}
}
},
"Outputs": {
"NetworkLoadBalancerArn": {
"Value": {
"Ref": "NLB"
},
"Description": "The network elastic load balancer Amazon resource name"
}
}
}
Unfortunately, CloudFormation does not support API Gateway's VPC Links at this time.
You can create a Lambda-backed custom resource to manage the VPC Link using CloudFormation.
Here is a Lambda function (using python3.6) for a CloudFormation custom resource I use to manage VPC links:
import copy
import json
import re
import time
import boto3
from botocore.vendored import requests
SUCCESS = "SUCCESS"
FAILED = "FAILED"
FAILED_PHYSICAL_RESOURCE_ID = "FAILED_PHYSICAL_RESOURCE_ID"
class AddOrUpdateTargetArnsError(Exception):
def __init__(self):
self.message = 'Target arns are not allowed to be changed/added.'
super().__init__(self.message)
class FailedVpcLinkError(Exception):
def __init__(self, status_message):
self.message = f'statusMessages: {status_message}'
super().__init__(self.message)
def lambda_handler(event, context):
try:
_lambda_handler(event, context)
except Exception as e:
send(
event,
context,
response_status=FAILED,
# Do not fail on delete to avoid rollback failure
response_data=None,
physical_resource_id=event.get('PhysicalResourceId', FAILED_PHYSICAL_RESOURCE_ID),
reason=e
)
# Must raise, otherwise the Lambda will be marked as successful, and the exception
# will not be logged to CloudWatch logs.
raise
def _lambda_handler(event, context):
print("Received event: ")
print(event)
resource_type = event['ResourceType']
if resource_type != "Custom::ApiGatewayVpcLink":
raise ValueError(f'Unexpected resource_type: {resource_type}')
request_type = event['RequestType']
wait_for = event.get('WaitFor', None)
resource_properties = event['ResourceProperties']
physical_resource_id = event.get('PhysicalResourceId', None)
apigateway = boto3.client('apigateway')
if wait_for:
handle_self_invocation(
wait_for=wait_for,
physical_resource_id=physical_resource_id,
event=event,
context=context,
)
else:
if request_type == 'Create':
kwargs = dict(
name=resource_properties['Name'],
targetArns=resource_properties['TargetArns'],
description=resource_properties.get('Description', None)
)
response = apigateway.create_vpc_link(**kwargs)
event_copy = copy.deepcopy(event)
event_copy['WaitFor'] = 'CreateComplete'
event_copy['PhysicalResourceId'] = response['id']
print('Reinvoking function because VPC link creation is asynchronous')
relaunch_lambda(event=event_copy, context=context)
return
elif request_type == 'Update':
old_resource_properties = event['OldResourceProperties']
current_target_arns = apigateway.get_vpc_link(
vpcLinkId=physical_resource_id,
)['targetArns']
# must compare current_target_arns to resource_properties['TargetArns'], to protect against
# UPDATE created by UPDATE_FAILED. In that particular case, current_target_arns will be the same as
# resource_properties['TargetArns'] but different than old_resource_properties['TargetArns']
if set(current_target_arns) != set(resource_properties['TargetArns']) and \
set(resource_properties['TargetArns']) != set(old_resource_properties['TargetArns']):
raise AddOrUpdateTargetArnsError()
patch_operations = []
if resource_properties['Name'] != old_resource_properties['Name']:
patch_operations.append(dict(
op='replace',
path='/name',
value=resource_properties['Name'],
))
if 'Description' in resource_properties and 'Description' in old_resource_properties:
if resource_properties['Description'] != old_resource_properties['Description']:
patch_operations.append(dict(
op='replace',
path='/description',
value=resource_properties['Description'],
))
elif 'Description' in resource_properties and 'Description' not in old_resource_properties:
patch_operations.append(dict(
op='replace',
path='/description',
value=resource_properties['Description'],
))
elif 'Description' not in resource_properties and 'Description' in old_resource_properties:
patch_operations.append(dict(
op='replace',
path='/description',
value=None,
))
apigateway.update_vpc_link(
vpcLinkId=physical_resource_id,
patchOperations=patch_operations,
)
elif request_type == 'Delete':
delete = True
if physical_resource_id == FAILED_PHYSICAL_RESOURCE_ID:
delete = False
print('Custom resource was never properly created, skipping deletion.')
stack_name = re.match("arn:aws:cloudformation:.+:stack/(?P<stack_name>.+)/.+", event['StackId']).group('stack_name')
if stack_name in physical_resource_id:
delete = False
print(f'Skipping deletion, because VPC link was not created properly. Heuristic: stack name ({stack_name}) found in physical resource ID ({physical_resource_id})')
logical_resource_id = event['LogicalResourceId']
if logical_resource_id in physical_resource_id:
delete = False
print(f'Skipping deletion, because VPC link was not created properly. Heuristic: logical resource ID ({logical_resource_id}) found in physical resource ID ({physical_resource_id})')
if delete:
apigateway.delete_vpc_link(
vpcLinkId=physical_resource_id
)
event_copy = copy.deepcopy(event)
event_copy['WaitFor'] = 'DeleteComplete'
print('Reinvoking function because VPC link deletion is asynchronous')
relaunch_lambda(event=event_copy, context=context)
return
else:
print(f'Request type is {request_type}, doing nothing.')
send(
event,
context,
response_status=SUCCESS,
response_data=None,
physical_resource_id=physical_resource_id,
)
def handle_self_invocation(wait_for, physical_resource_id, event, context):
apigateway = boto3.client('apigateway')
if wait_for == 'CreateComplete':
print('Waiting for creation of VPC link: {vpc_link_id}'.format(vpc_link_id=physical_resource_id))
response = apigateway.get_vpc_link(
vpcLinkId=physical_resource_id,
)
status = response['status']
print('Status of VPC link {vpc_link_id} is {status}'.format(vpc_link_id=physical_resource_id, status=status))
if status == 'AVAILABLE':
send(
event,
context,
response_status=SUCCESS,
response_data=None,
physical_resource_id=physical_resource_id,
)
elif status == 'FAILED':
raise FailedVpcLinkError(status_message=response['statusMessage'])
elif status == 'PENDING':
# Sleeping here to avoid polluting CloudWatch Logs by reinvoking the Lambda too quickly
time.sleep(30)
relaunch_lambda(event, context)
else:
print('Unexpected status, doing nothing')
elif wait_for == 'DeleteComplete':
print('Waiting for deletion of VPC link: {vpc_link_id}'.format(vpc_link_id=physical_resource_id))
try:
response = apigateway.get_vpc_link(
vpcLinkId=physical_resource_id,
)
except apigateway.exceptions.NotFoundException:
print('VPC link {vpc_link_id} deleted successfully'.format(vpc_link_id=physical_resource_id))
send(
event,
context,
response_status=SUCCESS,
response_data=None,
physical_resource_id=physical_resource_id,
)
else:
status = response['status']
assert status == 'DELETING', f'status is {status}'
# Sleeping here to avoid polluting CloudWatch Logs by reinvoking the Lambda too quickly
time.sleep(10)
relaunch_lambda(event, context)
else:
raise ValueError(f'Unexpected WaitFor: {wait_for}')
def relaunch_lambda(event, context):
boto3.client("lambda").invoke(
FunctionName=context.function_name,
InvocationType='Event',
Payload=json.dumps(event),
)
def send(event, context, response_status, response_data, physical_resource_id, reason=None):
response_url = event['ResponseURL']
response_body = {
'Status': response_status,
'Reason': str(reason) if reason else 'See the details in CloudWatch Log Stream: ' + context.log_stream_name,
'PhysicalResourceId': physical_resource_id,
'StackId': event['StackId'],
'RequestId': event['RequestId'],
'LogicalResourceId': event['LogicalResourceId'],
'Data': response_data,
}
json_response_body = json.dumps(response_body)
headers = {
'content-type': '',
'content-length': str(len(json_response_body))
}
try:
requests.put(
response_url,
data=json_response_body,
headers=headers
)
except Exception as e:
print("send(..) failed executing requests.put(..): " + str(e))
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With
