Dealing with Ambiguity

CDK アプリケーション用設定スクリプトを書いてみた

December 28, 2021
CDK

背景

CDK を使っている際に外部 API の URL や Route53 ドメイン、ACM 証明書の ARN 等をハードコードさせず、動的に与えたい場合があります。このような場合は aws_cdk.aws_ssm.StringParameter クラスの value_from_lookup や value_for_string_parameter といったメソッドを用いて、Systems Manager Parameter Store からパラメータの値もしくはデプロイ時に値を参照可能なトークンを取得する形になると思います。しかし、アプリの構造が複雑化するにつれ、設定するパラメータも多くなり、いちいちデプロイ前にコンソールにログインして適切なパラメータを設定するのが面倒になるかと思います。
なので、今回は、パラメータの作成・更新・削除を簡単に行えるようなスクリプトを作成してみました。ちなみに Secrets Manager へのシークレットの保存についても同時にやろうか迷いましたが、現状せいぜい GitHub OAuth Token くらいでそこまで必要性に駆られていないので今回は見送ることにしました。

パラメータ定義ファイルの作成

今回作成するスクリプトの目的の 1 つとして「汎用性」が挙げられます。つまり、どんな CDK アプリケーションでもパラメータの定義さえ決めてあげればあとはスクリプトを実行、値を渡してあげるだけで他の CDK アプリケーションと重複することなく値が参照可能であることを目指します。そのため、まずは以下のようなパラメータ定義ファイルを作成します。

config.params.json
{
  "Namespace": "/CDKWorkshop/",
  "Parameters": [
    {
      "Name": "CloudFrontAliasCertArn",
      "CLIFormat": "cloudfront-alias-cert-arn",
      "Description": "The ARN for the ACM certificate used for CloudFront destribution"
    },
    {
      "Name": "CloudFrontDomainName",
      "CLIFormat": "cloudfront-domain-name",
      "Description": "The domain name for CloudFront destribution"
    },
    {
      "Name": "HostedZoneName",
      "CLIFormat": "hosted-zone-name",
      "Description": "The name of Route 53 Hosted Zone"
    }
  ],
  "DefaultOptions": [
    {
      "CLIFormat": "delete",
      "ShortCLIFormat": "d",
      "Action": "store_true",
      "Help": "delete all AWS SSM Parameters (after CDK stack was destroyed)"
    },
    {
      "CLIFormat": "interactive",
      "ShortCLIFormat": "i",
      "Action": "store_true",
      "Help": "run in interactive mode"
    },
    {
      "CLIFormat": "test",
      "ShortCLIFormat": "t",
      "Action": "store_true",
      "Help": "run in test mode (only creates config.cache.json, but does not store parameters to SSM Parameter Store)"
    }
  ]
}

まず、Namespace ですが、こちらは対象の CDK アプリケーション用名前空間となります。実際に SSM Parameter Store に保存されるパラメータは Namespace/Parameter という形となり、これにより他アプリケーションとの重複を避けます。そのため、この値はアプリケーション間で一意なものであることが想定されます。
次に、Parameters ですが、この下に設定したいパラメータを書いていきます。具体的な項目の説明は以下の通りです。

項目 説明
Name パラメータの名前
CLIFormat コマンドオプションとして渡す際のフォーマット
Description パラメータの説明

例えば上の例の場合、/CDKWorkshop/CloudFrontAliasCertArn のようなパラメータが生成される形となります。

最後に DefaultOptions についてですが、これはスクリプトのオプションとなり、基本的にはいじらなくて OK です。もし何か機能を追加したい場合はここに追記し、スクリプト内でその機能を実装することで実現可能です。現時点で用意されるオプションは以下の通りです。

オプション 説明
—delete 指定された名前空間のパラメータを、SSM Parameter Store 上から削除する
—interactive 対話モードでスクリプトを実行する
—test テストモードでスクリプトを実行する (SSM Parameter Store にパラメータを保存せずに、ローカルのファイルにのみ出力する)

パラメータ定義ファイルについては以上です。実装に移りましょう。

コマンドラインオプション解析

まずはオプション解析です。今回は Python の argparse を利用させていただきました。なお、設定するオプションは先に作成したパラメータ定義ファイルから動的に決定されます。

loaded_params = json.load(open('config.params.json', 'r'))
cache_file = 'config.cache.json'
params = loaded_params['Parameters']
namespace = loaded_params['Namespace']
options = loaded_params['DefaultOptions']

...

def get_args():
  usage = 'Usage: python {} [--help]'.format(os.path.basename(__file__))
  for option in options:
    usage += ' [--{}]'.format(option['CLIFormat'])
  for param in params:
    usage += ' [--{} <value>]'.format(param['CLIFormat'])

  description = 'Use this script for configuring SSM parameters for CDK deployment.'
  argparser = ArgumentParser(usage=usage, description=description)

  for option in options:
    argparser.add_argument(
      '-' + option['ShortCLIFormat'], '--' + option['CLIFormat'],
      action=option['Action'],
      help=option['Help']
    )

  for param in params:
    argparser.add_argument(
      '--' + param['CLIFormat'],
      help=param['Description']
    )

  args = argparser.parse_args()
  return vars(args)

help を表示させるといい感じです。

$ python configure.py --help
usage: Usage: python configure.py [--help] [--delete] [--interactive] [--test] [--cloudfront-alias-cert-arn <value>] [--cloudfront-domain-name <value>] [--hosted-zone-name <value>]

Use this script for configuring SSM parameters for CDK deployment.

optional arguments:
  -h, --help            show this help message and exit
  -d, --delete          delete all AWS SSM Parameters (after CDK stack was destroyed)
  -i, --interactive     run in interactive mode
  -t, --test            run in test mode (only creates config.cache.json, but does not store parameters to SSM Parameter Store)
  --cloudfront-alias-cert-arn CLOUDFRONT_ALIAS_CERT_ARN
                        The ARN for the ACM certificate used for CloudFront destribution
  --cloudfront-domain-name CLOUDFRONT_DOMAIN_NAME
                        The domain name for CloudFront destribution
  --hosted-zone-name HOSTED_ZONE_NAME
                        The name of Route 53 Hosted Zone

SSM Parameter Store とのやりとり

次に実際に SSM Parameter Store にパラメータを保存(更新)・取得・削除するような処理を書いていきます。まずは渡されたコマンドラインオプションの値を変換してあげます。

def convert_args_to_ssm_params(args):
  ssm_params = []

  for param in params:
    if args[param['CLIFormat'].replace('-', '_')] is not None:
      ssm_params.append({
        'Name': namespace + param['Name'],
        'Value': args[param['CLIFormat'].replace('-', '_')],
        'Description': param['Description'],
      })

  return ssm_params

あとは保存・取得・削除用に関数を用意してあげるだけです。

def store_ssm_parameters(ssm_params):
  print('Storing parameters to SSM Parameter Store.')

  for ssm_param in ssm_params:
    try:
      response = ssm.put_parameter(
        Name=ssm_param['Name'],
        Value=ssm_param['Value'],
        Type='String',
        Overwrite=True,
        Description='{} under {}'.format(ssm_param['Description'], namespace)
      )
      print('The parameter "{}" has been stored in SSM Parameter Store.'.format(ssm_param['Name']))

    except Exception as e:
      print(e)


def load_ssm_parameters():
  loaded_ssm_params = []
  for param in params:
    try:
      response = ssm.get_parameter(
        Name=namespace + param['Name']
      )
      parameter = response['Parameter']
      loaded_ssm_params.append({
        'Name': parameter['Name'],
        'Value': parameter['Value'],
      })

    except ssm.exceptions.ParameterNotFound as e:
      print('Parameter "{}" has not been stored in SSM Parameter Store.'.format(param['Name']))
      loaded_ssm_params.append({
        'Name': namespace + param['Name'],
        'Value': ssm_not_defined,
      })

    except Exception as e:
      print(e)

  return loaded_ssm_params


def delete_ssm_parameters():
  print('Deleting all parameters starting with "{}" from SSM Parameter Store.'.format(namespace))
  
  try:
    response = ssm.delete_parameters(Names=[namespace + param['Name'] for param in params])
    if len(response['DeletedParameters']) == 0:
      print('No parameter stored with the namespace of "{}".'.format(namespace))
    else:
      print('Deleted parameters: {}'.format(", ".join(response['DeletedParameters'])))
  
  except Exception as e:
    print(e)

なお、取得時に保存されていないパラメータがある場合は “not-defined” として返すようにしました。

ssm_not_defined = 'not-defined'

キャッシュファイル作成処理

SSM パラメータストアから取得した値をローカルで確認する、テストモードで実行する、といったような目的より、今回キャッシュファイル (config.params.json) を作成するようにしました。その辺りの処理は以下の通りです。

def store_parameters_to_cache(_params):
  print('Writing parameters to {} .'.format(cache_file))

  with open(cache_file, 'w') as f:
    json.dump(_params, f, ensure_ascii=False)


def load_cached_parameters():
  if os.path.exists(cache_file):
    cached_params = json.load(open(cache_file, 'r'))
    return cached_params

  else:
    return []

対話モードの実装

本スクリプトは一気にパラメータの値を全て指定し実行することも可能ですが、対話モードで一つずつ指定することも可能です (というかこっちの方が需要は多いんじゃないかと思ってます)。対話モードで SSM パラメータストアに保存するパラメータを取得する処理は別で書く必要がありました。具体的には以下の通りです。

def get_ssm_params_interactive_mode():
  print('Please provide values for each parameters.')
  
  cached_params = load_cached_parameters()
  if len(cached_params) == 0:
    cached_params = get_temp_cache()

  ssm_params = []
  for param in params:
    prompt_str = namespace + param['Name']
    cached_val = get_cached_value(namespace + param['Name'], cached_params)
    
    if cached_val == ssm_not_defined:
      prompt_str += ': '
      
    else:
      prompt_str += ' ({}): '.format(cached_val)

    val = input(prompt_str)

    if val == '' and cached_val is not ssm_not_defined:
      val = cached_val

    if val != '' and val != ssm_not_defined:
      ssm_params.append({
        'Name': namespace + param['Name'],
        'Value': val,
        'Description': param['Description']
      })
      
  if len(ssm_params) > 0:
    return ssm_params
    
  else:
    print('At least one parameter needs to be provided if there is no cache available.')
    sys.exit()

なお、ローカルキャッシュに既に保存された値がある場合はその値をデフォルト値として使用し、またプロンプトにも表示してくれるようになってます。
例えば 1 回目の実行で HostedZoneName のみ指定し、2 回目の実行を対話モードで行った場合、HostedZoneName 入力時には前回の値が表示され、特に値を入力しなかった場合は前回の値が使用されます。

[ 1 回目 ]

$ python configure.py --hosted-zone-name hoge.com
Storing parameters to SSM Parameter Store.
The parameter "/CDKWorkshop/HostedZoneName" has been stored in SSM Parameter Store.
Parameter "CloudFrontAliasCertArn" has not been stored in SSM Parameter Store.
Parameter "CloudFrontDomainName" has not been stored in SSM Parameter Store.
Writing parameters to config.cache.json .

[ 2 回目 ]

$ python configure.py -i
Please provide values for each parameters.
/CDKWorkshop/CloudFrontAliasCertArn: arn:xxxxxxx
/CDKWorkshop/CloudFrontDomainName:      
/CDKWorkshop/HostedZoneName (hoge.com): 
Storing parameters to SSM Parameter Store.
The parameter "/CDKWorkshop/CloudFrontAliasCertArn" has been stored in SSM Parameter Store.
The parameter "/CDKWorkshop/HostedZoneName" has been stored in SSM Parameter Store.
Parameter "CloudFrontDomainName" has not been stored in SSM Parameter Store.
Writing parameters to config.cache.json .

最終的なスクリプト

あとはメインの処理やその他細かい実装を行い、スクリプトを完成させます。最終的なコードは以下の通りです。

configure.py
import boto3
import json
import os
import sys
from argparse import ArgumentParser

loaded_params = json.load(open('config.params.json', 'r'))
cache_file = 'config.cache.json'
params = loaded_params['Parameters']
namespace = loaded_params['Namespace']
options = loaded_params['DefaultOptions']
ssm = boto3.client('ssm')
ssm_not_defined = 'not-defined'


def get_args():
  usage = 'Usage: python {} [--help]'.format(os.path.basename(__file__))
  for option in options:
    usage += ' [--{}]'.format(option['CLIFormat'])
  for param in params:
    usage += ' [--{} <value>]'.format(param['CLIFormat'])

  description = 'Use this script for configuring SSM parameters for CDK deployment.'
  argparser = ArgumentParser(usage=usage, description=description)

  for option in options:
    argparser.add_argument(
      '-' + option['ShortCLIFormat'], '--' + option['CLIFormat'],
      action=option['Action'],
      help=option['Help']
    )

  for param in params:
    argparser.add_argument(
      '--' + param['CLIFormat'],
      help=param['Description']
    )

  args = argparser.parse_args()
  return vars(args)


def is_parameters_provided(args):
  none_value_count = list(args.values()).count(None)
  return none_value_count < len(args) - len(options)


def convert_args_to_ssm_params(args):
  ssm_params = []

  for param in params:
    if args[param['CLIFormat'].replace('-', '_')] is not None:
      ssm_params.append({
        'Name': namespace + param['Name'],
        'Value': args[param['CLIFormat'].replace('-', '_')],
        'Description': param['Description'],
      })

  return ssm_params


def store_ssm_parameters(ssm_params):
  print('Storing parameters to SSM Parameter Store.')

  for ssm_param in ssm_params:
    try:
      response = ssm.put_parameter(
        Name=ssm_param['Name'],
        Value=ssm_param['Value'],
        Type='String',
        Overwrite=True,
        Description='{} under {}'.format(ssm_param['Description'], namespace)
      )
      print('The parameter "{}" has been stored in SSM Parameter Store.'.format(ssm_param['Name']))

    except Exception as e:
      print(e)


def load_ssm_parameters():
  loaded_ssm_params = []
  for param in params:
    try:
      response = ssm.get_parameter(
        Name=namespace + param['Name']
      )
      parameter = response['Parameter']
      loaded_ssm_params.append({
        'Name': parameter['Name'],
        'Value': parameter['Value'],
      })

    except ssm.exceptions.ParameterNotFound as e:
      print('Parameter "{}" has not been stored in SSM Parameter Store.'.format(param['Name']))
      loaded_ssm_params.append({
        'Name': namespace + param['Name'],
        'Value': ssm_not_defined,
      })

    except Exception as e:
      print(e)

  return loaded_ssm_params


def delete_ssm_parameters():
  print('Deleting all parameters starting with "{}" from SSM Parameter Store.'.format(namespace))
  
  try:
    response = ssm.delete_parameters(Names=[namespace + param['Name'] for param in params])
    if len(response['DeletedParameters']) == 0:
      print('No parameter stored with the namespace of "{}".'.format(namespace))
    else:
      print('Deleted parameters: {}'.format(", ".join(response['DeletedParameters'])))
  
  except Exception as e:
    print(e)


def store_parameters_to_cache(_params):
  print('Writing parameters to {} .'.format(cache_file))

  with open(cache_file, 'w') as f:
    json.dump(_params, f, ensure_ascii=False)


def load_cached_parameters():
  if os.path.exists(cache_file):
    cached_params = json.load(open(cache_file, 'r'))
    return cached_params

  else:
    return []


def get_temp_cache():
  return [{'Name': namespace + _param['Name'], 'Value': ssm_not_defined} for _param in params]


def get_cached_value(param_name, cached_params):
  return list(filter(lambda item: item['Name'] == param_name, cached_params))[0]['Value']


def get_ssm_params_interactive_mode():
  print('Please provide values for each parameters.')
  
  cached_params = load_cached_parameters()
  if len(cached_params) == 0:
    cached_params = get_temp_cache()

  ssm_params = []
  for param in params:
    prompt_str = namespace + param['Name']
    cached_val = get_cached_value(namespace + param['Name'], cached_params)
    
    if cached_val == ssm_not_defined:
      prompt_str += ': '
      
    else:
      prompt_str += ' ({}): '.format(cached_val)

    val = input(prompt_str)

    if val == '' and cached_val is not ssm_not_defined:
      val = cached_val

    if val != '' and val != ssm_not_defined:
      ssm_params.append({
        'Name': namespace + param['Name'],
        'Value': val,
        'Description': param['Description']
      })
      
  if len(ssm_params) > 0:
    return ssm_params
    
  else:
    print('At least one parameter needs to be provided if there is no cache available.')
    sys.exit()


def get_cached_params_for_test(ssm_params):
  cached_params = []

  for param in params:
    searched_param = list(filter(lambda item: item['Name'] == namespace + param['Name'], ssm_params))
    if len(searched_param) == 0:
      cached_params.append({
        'Name': namespace + param['Name'],
        'Value': ssm_not_defined,
      })
    else:
      cached_params.append({
        'Name': searched_param[0]['Name'],
        'Value': searched_param[0]['Value'],
      })
  
  return cached_params


def main():
  args = get_args()
  
  if args['delete']:
    delete_ssm_parameters()

    if os.path.exists(cache_file):
      os.remove(cache_file)
    sys.exit()

  if args['test']:
    if args['interactive']:
      test_ssm_params = get_ssm_params_interactive_mode()
      ssm_params = get_cached_params_for_test(test_ssm_params)
      store_parameters_to_cache(ssm_params)
      sys.exit()
    
    else:
      if is_parameters_provided(args):
        test_ssm_params = convert_args_to_ssm_params(args)
        ssm_params = get_cached_params_for_test(test_ssm_params)
        store_parameters_to_cache(ssm_params)
        sys.exit()

      else:
        print('At least one parameter needs to be provided for non-interactive mode.')
        sys.exit()

  if args['interactive']:
    ssm_params = get_ssm_params_interactive_mode()
    store_ssm_parameters(ssm_params)
    loaded_ssm_params = load_ssm_parameters()
    store_parameters_to_cache(loaded_ssm_params)

  else:
    if is_parameters_provided(args):
      ssm_params = convert_args_to_ssm_params(args)
      store_ssm_parameters(ssm_params)
      loaded_ssm_params = load_ssm_parameters()
      store_parameters_to_cache(loaded_ssm_params)

    else:
      print('At least one parameter needs to be provided for non-interactive mode.')

  
if __name__ == '__main__':
  main()

動作確認

早速動作確認してみます。実際に CDK アプリケーションで使用する際はアプリケーションルートに configure.py と config.params.json を配置しておくと良いと思います。
まずはパラメータの作成です。

$ python configure.py --cloudfront-alias-cert-arn arn:hogehoge --cloudfront-domain-name hoge.com --hosted-zone-name hoge.com
Storing parameters to SSM Parameter Store.
The parameter "/CDKWorkshop/CloudFrontAliasCertArn" has been stored in SSM Parameter Store.
The parameter "/CDKWorkshop/CloudFrontDomainName" has been stored in SSM Parameter Store.
The parameter "/CDKWorkshop/HostedZoneName" has been stored in SSM Parameter Store.
Writing parameters to config.cache.json .

SSM のコンソールから以下のようにパラメータが保存できていることが確認できます。

cdk_configuration_script_1

削除も以下のように行えます。CDK アプリケーションの CloudFormation スタックを削除後に SSM Parameter Store 内のパラメータを削除するのに利用可能です。

$ python configure.py -d
Deleting all parameters starting with "/CDKWorkshop/" from SSM Parameter Store.
Deleted parameters: /CDKWorkshop/CloudFrontAliasCertArn, /CDKWorkshop/CloudFrontDomainName, /CDKWorkshop/HostedZoneName

次に対話モードの実行です。-i オプションで指定可能です。これは対話モードに限りませんが、以下のように一部のパラメータのみを指定することもできます。

$ python configure.py -i
Please provide values for each parameters.
/CDKWorkshop/CloudFrontAliasCertArn: 
/CDKWorkshop/CloudFrontDomainName: 
/CDKWorkshop/HostedZoneName: hoge.com
Storing parameters to SSM Parameter Store.
The parameter "/CDKWorkshop/HostedZoneName" has been stored in SSM Parameter Store.
Parameter "CloudFrontAliasCertArn" has not been stored in SSM Parameter Store.
Parameter "CloudFrontDomainName" has not been stored in SSM Parameter Store.
Writing parameters to config.cache.json .

最後にテストモードです。実際に SSM Parameter Store にパラメータを保存することはなく、config.cache.json のみをアップデートします。

$ python configure.py -it
Please provide values for each parameters.
/CDKWorkshop/CloudFrontAliasCertArn: arn:hogehoge
/CDKWorkshop/CloudFrontDomainName: hoge.com
/CDKWorkshop/HostedZoneName (hoge.com): 
Writing parameters to config.cache.json .
$ cat config.cache.json 
[{"Name": "/CDKWorkshop/CloudFrontAliasCertArn", "Value": "arn:hogehoge"}, {"Name": "/CDKWorkshop/CloudFrontDomainName", "Value": "hoge.com"}, {"Name": "/CDKWorkshop/HostedZoneName", "Value": "hoge.com"}]

CDK スタックから値を参照する

それでは SSM Parameter Store にパラメータが保存できたところで、実際に CDK スタックからその値を参照してみましょう。面倒だったのでこれまで散々触ってきた HitCounter をここでも使います。具体的には Lambda 関数の環境変数として SSM パラメータの値を設定しているだけで特に意味は無いですが、こんな感じで参照できるんだなぁというニュアンスがわかればいいと思います。

cdk_workshop/hitcounter.py
class HitCounter(Construct):

  @property
  def handler(self):
    return self._handler

  @property
  def table(self):
    return self._table

  def __init__(self, scope: Construct, id: str, downstream: _lambda.IFunction, read_capacity: int = 5, **kwargs):
    if read_capacity < 5 or read_capacity > 20:
      raise ValueError("read_capacity must be greater than 5 or less than 20")
    
    super().__init__(scope, id, **kwargs)

    self._table = ddb.Table(
      self, 'Hits',
      partition_key={'name': 'path', 'type': ddb.AttributeType.STRING},
      encryption=ddb.TableEncryption.AWS_MANAGED,
      read_capacity=read_capacity,
    )

    self._handler = _lambda.Function(
      self, 'HitCountHandler',
      runtime=_lambda.Runtime.PYTHON_3_9,
      handler='hitcount.handler',
      code=_lambda.Code.from_asset('lambda'),
      environment={
        'DOWNSTREAM_FUNCTION_NAME': downstream.function_name,
        'HITS_TABLE_NAME': self._table.table_name,
        'HOSTED_ZONE_NAME': ssm.StringParameter.value_for_string_parameter(self, "/CDKWorkshop/HostedZoneName"),
        'CLOUD_FRONT_DOMAIN_NAME': ssm.StringParameter.value_for_string_parameter(self, "/CDKWorkshop/CloudFrontDomainName"),
      }
    )

    self._table.grant_read_write_data(self.handler)
    downstream.grant_invoke(self.handler)

実際にデプロイさせると Lambda 関数の環境変数として CLOUD_FRONT_DOMAIN_NAME と HOSTED_ZONE_NAME が設定されていることがわかります。

cdk_configuration_script_2

おわりに

いかがでしたでしょうか。こんな感じのスクリプトを用意しておくと、今後 CDK でインフラやアプリケーションをデプロイする際の手間が省け、より充実した CDK ライフが送れそうな気がしますね。

　© 2023, Dealing with Ambiguity