【テクニカル・上級編】CloudFormation Registryを用いたプライベート拡張プロバイダーの自作：カスタムリソースを書かずに独自AWSリソースやSaaS連携をIaC化する裏技 – インフラ構成管理（IaC）活用バイブル

CloudFormation Registryの深淵：カスタムリソースの呪縛を断ち切り、真の宣言的IaCを降臨させる方法

世のインフラエンジニアの大半は、CloudFormationの表現力の限界に直面したとき、反射的に「Lambdaカスタムリソース（`AWS::CloudFormation::CustomResource`）」に逃げる。

SaaSのAPIを叩きたい？ Lambdaを書け。
社内独自DBにレコードを挿入したい？ Lambdaを書け。
複雑なプロビジョニングフローが必要？ Lambdaを書け。

ちょっと待て。それはIaCではなく、「Lambdaスクリプトを非同期で雑多に呼び出すスパゲッティコードの量産」に他ならない。エラーハンドリングの不備によるスタックのロールバック地獄、タイムアウトの恐怖、そして何より「真の宣言的（Declarative）な状態管理の崩壊」。

AWSのエキスパートであり、数々の修羅場をくぐり抜けてきたSREなら知っているはずだ。CloudFormationには、サードパーティ製リソースや社内独自システムを、第一級（First-Class）のAWSリソースと同等に扱うための究極のメカニズムが備わっている。

それが CloudFormation Registry と CloudFormation CLI（`cfn`） によるプライベート拡張プロバイダー（Private Extension Provider）の自作である。

今回は、カスタムリソースという名の「技術的負債の隠れ家」を永遠に封印し、CloudFormationの内部アーキテクチャの深部にまで踏み込んだ、極限の拡張プロバイダー開発の奥義を授けよう。

—

1. 圧倒的なパラダイムシフト：カスタムリソース vs 拡張プロバイダー

まずは、敵を知るために両者のアーキテクチャ的違いを明確にしておく。

カスタムリソースは、言ってみれば「バッチ処理の寄せ集め」だ。それに対し、拡張プロバイダーは 「AWSのネイティブAPI群の延長線上に、独自のインフラドメインを完全に定着させる」 手法である。

—

2. 実践：CloudFormation CLIによるプロバイダー開発の全貌

今回は、例として「社内の独自ドメイン管理システム（Internal DNS API）」をCloudFormationで直接宣言的に管理するためのResource ProviderをPythonで構築する手順を解説する。

01. 開発環境の構築とスキャッフォールディング

CloudFormation CLIはPython環境で動作する。まずはツールチェーンを導入し、プロジェクトの骨組みを生成する。

仮想環境の作成と有効化
python3 -m venv .venv
source .venv/bin/activate

cfn-cliのインストール
pip install cloudformation-cli cloudformation-cli-python-plugin

プロジェクトの初期化（Resource Nameは Vendor::Service::Resource の形式）
cfn init –project-generator python –resource-type Acme::DNS::Record

これにより、以下のディレクトリ構造が自動生成される。

acme-dns-record/
├── acorn.yml # プロジェクト設定
├── docs/ # 自動生成ドキュメント
├── pyproject.toml # 依存関係定義
├──
├── acme-dns-record.json # 【最重要】リソースのJSON Schema
├── handlers.py # 【最重要】CRUDLロジックの実装
└──

02. スキーマ定義（`acme-dns-record.json`）の極意

このJSON Schemaは、CloudFormationがテンプレートの構文解析や入力値検証を行うための契約書だ。厳密に定義せよ。

{
“typeName”: “Acme::DNS::Record”,
“description”: “Internal DNS Record Provider for Acme Corp”,
“sourceUrl”: “https://github.com/example/acme-dns-record”,
“definitions”: {
“RecordType”: {
“type”: “string”,
“enum”: [“A”, “CNAME”, “TXT”]
}
},
“properties”: {
“DomainName”: {
“type”: “string”,
“description”: “The fully qualified domain name.”
},
“RecordType”: {
“$ref”: “#/definitions/RecordType”
},
“Value”: {
“type”: “string”,
“description”: “Target IP or CNAME value.”
},
“Ttl”: {
“type”: “integer”,
“default”: 300
},
“RecordId”: {
“type”: “string”,
“description”: “System-generated unique identifier.”
}
},
“required”: [
“DomainName”,
“RecordType”,
“Value”
],
“primaryIdentifier”: [
“/properties/DomainName”
],
“readOnlyProperties”: [
“/properties/RecordId”
],
“handlers”: {
“create”: {
“permissions”: []
},
“read”: {
“permissions”: []
},
“update”: {
“permissions”: []
},
“delete”: {
“permissions”: []
},
“list”: {
“permissions”: []
}
}
}

Expert Tip: `primaryIdentifier` と `readOnlyProperties` の設計が不適切だと、スタック更新時のリソース置換（Replacement）が正しく動作せず、CloudFormationエンジンと内部ステートが乖離する。

03. ハンドラー実装（`handlers.py`）の完全最適化

CloudFormation CLIは、`CREATE`, `READ`, `UPDATE`, `DELETE`, `LIST` (CRUDL) の5つのハンドラーを実装することを要求する。

ここで最も重要なのは、「冪等性の完全な担保」と「例外ハンドリングによる適切なステータスコードの返却」である。

以下は、社内DNS API（外側のREST APIを想定）を叩くPythonハンドラーの実装例だ。

import logging
from typing import Any, MutableMapping, Optional
from cloudformation_cli_python_lib import (
Action,
HandlerErrorCode,
OperationStatus,
ProgressEvent,
Resource,
SessionProxy,
exceptions,
)
import requests

ロギング設定（CloudWatch Logsに出力される）
LOG = logging.getLogger(__name__)
LOG.setLevel(logging.INFO)

resource = Resource(type_name=”Acme::DNS::Record”)
TYPE_NAME = resource.type_name

内部APIクライアントの設定（実際にはSecret Manager等からクレデンシャルを取得する）
API_ENDPOINT = “https://api.internal.acme.corp/v1/dns”
API_TOKEN = “sec3rt-token-xyz” # 本番では環境変数やSecrets Managerを活用すること

def get_headers():
return {
“Authorization”: f”Bearer {API_TOKEN}”,
“Content-Type”: “application/json”
}

@resource.handler(Action.CREATE)
def create_handler(
session: Optional[SessionProxy],
request: MutableMapping[str, Any],
callback_context: MutableMapping[str, Any],
) -> ProgressEvent:
model = request.desired_state

payload = {
“domain”: model.DomainName,
“type”: model.RecordType,
“value”: model.Value,
“ttl”: model.Ttl
}

try:
response = requests.post(API_ENDPOINT, json=payload, headers=get_headers(), timeout=10)

if response.status_code == 409:
# 既に存在する場合（冪等性の担保：必要に応じてUPDATEにフォールバックするかエラーにする）
raise exceptions.AlreadyExists(f”DNS Record {model.DomainName} already exists.”)

response.raise_for_status()
res_data = response.json()

# 外部システムが付与した一意なIDをモデルに格納
model.RecordId = res_data.get(“id”)

LOG.info(f”Successfully created DNS record for {model.DomainName}”)

return ProgressEvent(
status=OperationStatus.SUCCESS,
resource_model=model,
message=”Creation complete”
)

except requests.exceptions.RequestException as e:
LOG.error(f”Failed to create DNS record: {str(e)}”)
return ProgressEvent(
status=OperationStatus.FAILED,
error_code=HandlerErrorCode.ServiceInternalError,
message=str(e)
)

@resource.handler(Action.READ)
def read_handler(
session: Optional[SessionProxy],
request: MutableMapping[str, Any],
callback_context: MutableMapping[str, Any],
) -> ProgressEvent:
model = request.desired_state

try:
response = requests.get(f”{API_ENDPOINT}/{model.DomainName}”, headers=get_headers(), timeout=10)

if response.status_code == 404:
return ProgressEvent(
status=OperationStatus.FAILED,
error_code=HandlerErrorCode.NotFound,
message=f”DNS Record {model.DomainName} not found.”
)

response.raise_for_status()
res_data = response.json()

# モデルへのマッピング
model.DomainName = res_data.get(“domain”)
model.RecordType = res_data.get(“type”)
model.Value = res_data.get(“value”)
model.Ttl = res_data.get(“ttl”)
model.RecordId = res_data.get(“id”)

return ProgressEvent(
status=OperationStatus.SUCCESS,
resource_model=model
)

except requests.exceptions.RequestException as e:
return ProgressEvent(
status=OperationStatus.FAILED,
error_code=HandlerErrorCode.GeneralException,
message=str(e)
)

@resource.handler(Action.UPDATE)
def update_handler(
session: Optional[SessionProxy],
request: MutableMapping[str, Any],
callback_context: MutableMapping[str, Any],
) -> ProgressEvent:
model = request.desired_state

payload = {
“type”: model.RecordType,
“value”: model.Value,
“ttl”: model.Ttl
}

try:
response = requests.put(f”{API_ENDPOINT}/{model.DomainName}”, json=payload, headers=get_headers(), timeout=10)
response.raise_for_status()

return ProgressEvent(
status=OperationStatus.SUCCESS,
resource_model=model
)
except requests.exceptions.RequestException as e:
return ProgressEvent(
status=OperationStatus.FAILED,
error_code=HandlerErrorCode.ServiceInternalError,
message=str(e)
)

@resource.handler(Action.DELETE)
def delete_handler(
session: Optional[SessionProxy],
request: MutableMapping[str, Any],
callback_context: MutableMapping[str, Any],
) -> ProgressEvent:
model = request.desired_state

try:
response = requests.delete(f”{API_ENDPOINT}/{model.DomainName}”, headers=get_headers(), timeout=10)

if response.status_code == 404:
# 既に消えている場合は成功とみなす（冪等性）
pass
else:
response.raise_for_status()

return ProgressEvent(
status=OperationStatus.SUCCESS,
resource_model=None # 削除時はモデルを空にする
)
except requests.exceptions.RequestException as e:
return ProgressEvent(
status=OperationStatus.FAILED,
error_code=HandlerErrorCode.ServiceInternalError,
message=str(e)
)

@resource.handler(Action.LIST)
def list_handler(
session: Optional[SessionProxy],
request: MutableMapping[str, Any],
callback_context: MutableMapping[str, Any],
) -> ProgressEvent:
# 複数リソースの一覧取得（スタックドリフトやリソーススキャンで使用される）
try:
response = requests.get(API_ENDPOINT, headers=get_headers(), timeout=10)
response.raise_for_status()
items = response.json()

models = []
for item in items:
models.append(
resource.get_model({
“DomainName”: item.get(“domain”),
“RecordType”: item.get(“type”),
“Value”: item.get(“value”),
“Ttl”: item.get(“ttl”),
“RecordId”: item.get(“id”)
})
)

return ProgressEvent(
status=OperationStatus.SUCCESS,
resource_models=models
)
except requests.exceptions.RequestException as e:
return ProgressEvent(
status=OperationStatus.FAILED,
error_code=HandlerErrorCode.ServiceInternalError,
message=str(e)
)

—

3. レジストリへのデプロイメントとアカウント内への登録自動化

コードが書けたら、これをAWSアカウントのCloudFormation Registryにプライベート拡張機能として登録する。このプロセスも完全自動化のパイプラインに組み込むべきだ。

01. パッケージングとアカウント登録のコマンド

1. 依存関係を含めてパッケージをビルド
cfn submit –dry-run
cfn submit –region ap-northeast-1

出力される型ID（例: arn:aws:cloudformation:ap-northeast-1:123456789012:type/resource/Acme-DNS-Record）を控え、
アカウントのデフォルトバージョンに設定する
aws cloudformation set-type-default-version \
–arn arn:aws:cloudformation:ap-northeast-1:123456789012:type/resource/Acme-DNS-Record \
–region ap-northeast-1

これで、あなたのAWSアカウント内において、`AWS::S3::Bucket` や `AWS::EC2::Instance` と全く同じように、以下のCloudFormationテンプレートが記述できるようになる。

AWSTemplateFormatVersion: ‘2010-09-09’
Resources:
MyInternalDns:
Type: Acme::DNS::Record
Properties:
DomainName: “app.internal.acme.corp”
RecordType: “A”
Value: “10.0.100.50”
Ttl: 60

信じられるか？ Lambdaコードを一切書くことなく、CloudFormationが直接サードパーティAPIや社内システムを宣言的に管理しているのだ。

—

4. 低レイヤ＆エキスパート知見：パフォーマンス最適化とトラブルシューティングハック

数々のプロダクション環境で拡張プロバイダーを運用してきたアーキテクトとして、現場で絶対に知っておくべき「地雷と最適化の知見」を共有する。

1. コールドスタートとメモリ消費の最適化

CloudFormation Registryで実行されるハンドラーは、AWS内部のセキュアなコンテナサンドボックス内で実行される。Pythonプラグインの場合、インポートする外部ライブラリの肥大化はそのまま初期化レイテンシー（コールドスタート）に直結する。

`requests` などの軽量なHTTPクライアントを使い、余計な重いライブラリ（`boto3`はデフォルトで利用可能なので不要なインポートを避けるなど）を持ち込まないこと。
依存関係は `pyproject.toml` に厳格に固定し、ビルドサイズを最小限（数MB以下）に抑える。

2. タイムアウト制約のハック

ハンドラーの最大実行時間は、仕様上制限がある（通常最大60秒程度）。もし外部SaaSのプロビジョニングに数分かかる場合（例：データベースの作成やDNS伝搬待ち）、同期的に待つのは悪手である。

このような場合は、`callback_context` を用いてポーリング機構（Asynchronous Polling）を実装する。
初回の `CREATE` でAPIに非同期ジョブをキックし、ジョブIDを `callback_context` に保存して `OperationStatus.IN_PROGRESS` を返す。
CloudFormationエンジンが自動的に数秒〜数十秒後に再度ハンドラーを呼び出し、ステータスが `COMPLETED` になるまでループさせる。これにより、タイムアウトのエラーを華麗に回避できる。

3. デバッグのためのCloudWatch Logs解析

拡張プロバイダーが意図せぬエラー（`InternalFailure` 等）を吐いた場合、通常のCloudFormationのエラーメッセージは不親切である。

ログは、プロバイダーをデプロイしたリージョンのCloudWatch Logs内、ロググループ `AWS/CloudFormation/Registry` に出力される。
例外発生時のスタックトレースや `print`/`logging` の出力は全てここに集約されるため、開発時は必ずこのロググループをテールしながら `cfn submit` を回すこと。

—

5. 結び：インフラストラクチャの「境界線」を消し去れ

カスタムリソースの時代は終わった。

Lambdaのコードを貼り付け、イベントペイロードの `RequestType` を分岐させ、エラー時に `CfnResponse` モジュールへHTTPリクエストを送りつける泥臭いハックは、もう過去の遺物にすべきだ。

CloudFormation Registryを用いたプロバイダー自作は、インフラストラクチャの定義言語としてのCloudFormationの領域を、AWSのクラウドフロンティアの向こう側へと無限に拡張する。SaaS、オンプレミス、社内独自API——あらゆるリソースをあなたのテンプレートの支配下に置き、真の宣言的インフラストラクチャの自動化を極限まで押し上げろ。

これぞ、プロフェッショナルなSREが到達するべき、自動化の最高峰である。