HeatWave on AWS で PrivateLink 接続インバウンドレプリケーション時に Aurora のフェイルオーバーに対応する
こちらの記事の続きです。
記事の最後に、
また、ソース DB のフェイルオーバーに追従するには、Aurora クラスターのフェイルオーバーイベントを SNS トピック経由で受け取ってターゲットの IP アドレス（Writer）再登録を行う処理を Lambda などで実装する必要があります。
と書きましたが、これを実装してみました。
PrivateLink 用のターゲットグループ属性を編集
PrivateLink 用のターゲットグループの画面で「属性」タブを開いて右端の「編集」をクリックします。
以下の設定項目を変更して「変更内容の保存」をクリックします。
- ターゲットが異常になったときに接続を終了する : オフ（無効）
- 登録解除時に接続を終了 : オン（有効）
- 登録解除の遅延（ストリーミング間隔） : 0 秒
- クロスゾーン負荷分散 : オン
ソース DB（Aurora）のクラスターに Reader を追加
（RDS のメニューで）ソース DB（Aurora）のクラスターの画面から、「アクション」メニューの「リーダーの追加」を実施します。
ターゲット入れ替え用の Lambda 関数を作成
Aurora クラスターのフェイルオーバーイベントを受信した際に、PrivateLink 用のターゲットグループのターゲットを新しい Writer に入れ替えるための Lambda 関数です。
Lambda の「関数の作成」画面で以下を入力・選択して関数を作成します。
- 一から作成
- 関数名 : 任意
- ランタイム : Python 3.12
- アーキテクチャ : 任意（arm64 を推奨）
関数が作成されたら、「コード」タブでコードソースを入力します（lambda_function.py）。
lambda_function.py
import boto3
import json
import os
client = boto3.client('elbv2')
target_group_arn = os.getenv('TARGET_GROUP_ARN')
ip_dict = json.loads(os.getenv('IP_DICT'))
# インスタンスIDを取得
def get_instance_id(event):
records = event.get('Records', [])
if len(records) == 0:
print('Recordsの要素が見つかりません')
return None
sns = records[0].get('Sns', '')
if sns == '':
print('Snsが見つかりません')
return None
message_json = sns.get('Message', '')
if message_json == '':
print('Messageが見つかりません')
return None
message = json.loads(message_json)
source_id = message.get('Source ID', '')
if source_id == '':
print('Source IDが見つかりません')
return None
event_message = message.get('Event Message', '')
if event_message == '':
print('Event Messageが見つかりません')
return None
index = event_message.rfind(': ')
if index == -1:
print('Instance IDが見つかりません')
return None
return event_message[index + 2:]
# ターゲットとして設定されているIPアドレスを取得
def get_target_ip_address(target_group_arn):
response = client.describe_target_health(TargetGroupArn=target_group_arn)
target_health_descriptions = response.get('TargetHealthDescriptions', [])
if len(target_health_descriptions) == 0:
print('ターゲットが見つかりません')
return None
target = target_health_descriptions[0].get('Target')
if target is None:
print('Targetが見つかりません')
return None
ip_address = target.get('Id')
if ip_address is None:
print('ターゲットとして設定されているIPアドレスが見つかりません')
return None
return ip_address
# 指定IPアドレスをターゲットから登録解除
def deregist_target_ip_address(ip_address):
response = client.deregister_targets(
TargetGroupArn=target_group_arn,
Targets=[
{
'Id': ip_address,
'Port': 3306
},
]
)
return
# 指定IPアドレスをターゲットとして登録
def regist_target_ip_address(ip_address):
response = client.register_targets(
TargetGroupArn=target_group_arn,
Targets=[
{
'Id': ip_address,
'Port': 3306
},
]
)
return
def lambda_handler(event, context):
# インスタンスID取得
instance_id = get_instance_id(event)
if instance_id is None:
# インスタンスIDなし→処理打ち切り
return
# 変更後のIPアドレスを取得
new_ip_address = ip_dict.get(instance_id)
if new_ip_address is None:
print('ターゲット候補の中にフェイルオーバー先のインスタンスIDが見つかりません : (%s)' % (instance_id))
return
# 変更前のIPアドレスを取得
old_ip_address = get_target_ip_address(target_group_arn)
if old_ip_address is None:
return
# 変更前後でターゲットIPアドレスが変わるか？
if new_ip_address == old_ip_address:
print('ターゲットIPアドレスの変更はありません : (%s)' % (new_ip_address))
return
# 古いターゲットIPアドレスを登録解除
deregist_target_ip_address(old_ip_address)
print('古いターゲットIPアドレスを登録解除しました : (%s)' % (old_ip_address))
# 新しいターゲットIPアドレスを登録
regist_target_ip_address(new_ip_address)
print('新しいIPアドレスをターゲットに登録しました : (%s)' % (new_ip_address))
「設定」タブの「一般設定」でタイムアウトを 10 秒程度に設定しておきます。
通常は 2 秒以内に処理が終わるので、デフォルトの 3 秒のままでも支障はありませんが、念のため少し延長しておきます。
「環境変数」で以下の環境変数を登録します。
-
IP_DICT:{"【1つ目のインスタンスID】": "【1つ目のインスタンスのIPアドレス】", "【2つ目のインスタンスID】": "【2つ目のインスタンスのIPアドレス】"}
-
TARGET_GROUP_ARN:- 前の記事で作成したターゲットグループの ARN】
ここで一旦「コード」タブに戻って「Deploy」をクリックしておきます。
Lambda 関数実行用のロールに権限を追加
「設定」タブの「アクセス権限」でロール名のリンクをクリックし、ロール（IAM）画面の「許可」タブで「許可を追加」→「インラインポリシーを作成」をクリックします。
以下のポリシーを JSON 形式で入力して「次へ」をクリックします。
ポリシー（JSON）
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"elasticloadbalancing:DescribeLoadBalancers",
"elasticloadbalancing:DescribeTargetGroupAttributes",
"elasticloadbalancing:DescribeTargetHealth",
"elasticloadbalancing:DescribeTargetGroups"
],
"Resource": "*"
},
{
"Sid": "VisualEditor1",
"Effect": "Allow",
"Action": [
"elasticloadbalancing:RegisterTargets",
"elasticloadbalancing:DeregisterTargets"
],
"Resource": "arn:aws:elasticloadbalancing:ap-northeast-1:【アカウントID】:targetgroup/*/*"
}
]
}
「ポリシーの作成」をクリックします。
SNS トピックとサブスクリプションを作成
Aurora クラスターのフェイルオーバーイベントを受けて Lambda 関数を呼び出すための SNS トピックとサブスクリプションを作成します。
SNS トピックを作成
SNS のメニューから「トピック」を選択し、「トピックの作成」をクリックします。
以下を入力・選択して「トピックの作成」をクリックします。
- タイプ : スタンダード
- 名前 : 任意
SNS サブスクリプションを作成
SNS のメニューから「サブスクリプション」を選択し、「サブスクリプションの作成」をクリックします。
以下を入力・選択して「サブスクリプションの作成」をクリックします。
-
トピック ARN : 直前に作成したトピックを選択
-
プロトコル : AWS Lambda
-
エンドポイント : 先に作成した Lambda 関数を選択
-
サブスクリプションフィルターポリシー : オン（有効化）
- フィルターポリシーのスコープ : メッセージ本文
- JSON エディタ : 以下の JSON を入力
サブスクリプションフィルターポリシー（JSON）
サブスクリプションフィルターポリシー{ "Event ID": [ { "suffix": "RDS-EVENT-0072" }, { "suffix": "RDS-EVENT-0073" } ] }
Aurora（RDS）のイベントサブスクリプションを作成
RDS のメニューから「イベントサブスクリプション」を選択し、「イベントサブスクリプションの作成」をクリックします。
以下を入力・選択して「作成」をクリックします。
- 名前 : 任意
- 次への通知の送信 : Amazon リソースネーム（ARN）
- ARN : 先ほど作成した SNS トピックの ARN を選択
- ソースタイプ : クラスター
- クラスターを含みます : 特定のクラスターを選択
- 特定のクラスター : ソース DB（Aurora）のクラスターを選択
- イベントカテゴリを含みます : 特定のイベントカテゴリを選択
- 特定のイベントカテゴリ : failover
フェイルオーバーを試してみる
実際にソース DB（AUrora）でフェイルオーバーを発生させて、レプリケーションが追従するか試してみます。
ソース DB でテスト用の DB スキーマ・テーブル・データ行を追加
簡単なテーブルを作成してデータ行を入れてみます。
mysql> CREATE DATABASE hoge;
Query OK, 1 row affected (0.00 sec)
mysql> USE hoge;
Database changed
mysql> CREATE TABLE fuga (id int PRIMARY KEY AUTO_INCREMENT, val int);
Query OK, 0 rows affected (0.02 sec)
mysql> INSERT INTO fuga SET val = 100;
Query OK, 1 row affected (0.01 sec)
mysql> INSERT INTO fuga SET val = 150;
Query OK, 1 row affected (0.00 sec)
レプリカ DB で複製確認
レプリカ DB（HeatWave on AWS）で確認してみます。
先ほどソース DB で追加した DB スキーマ・テーブル・データ行が複製されています。
フェイルオーバーを実行
ソース DB（Aurora）の画面から、フェイルオーバーを実行します。
「フェイルオーバー」をクリックします。
しばらく待ってフェイルオーバーが実行されると、クラスターの「最近のイベント」にフェイルオーバー関連のイベントが表示されます。
PrivateLink 用のターゲットグループのターゲットが入れ替わりました。
レプリカ DB（HeatWave on AWS）の Channel 状態が「Needs Attention」に変わりました。
その後、4 ～ 5 分ほど経過して「Active」に戻りました。
インバウンドレプリケーションが復活したようです。
ソース DB でデータ行を追加
インバウンドレプリケーションの状態を（ステータスの表示だけではなく）実際の環境で確かめるために、ソース DB（Aurora）でデータ行を追加してみます。
mysql> INSERT INTO fuga SET val = 200;
ERROR 2013 (HY000): Lost connection to MySQL server during query
No connection. Trying to reconnect...
Connection id: 55
Current database: hoge
Query OK, 1 row affected (0.04 sec)
mysql> INSERT INTO fuga SET val = 250;
Query OK, 1 row affected (0.01 sec)
レプリカ DB でフェイルオーバー後の複製確認
再びレプリカ DB（HeatWave on AWS）で確認してみます。
先ほどソース DB で追加したデータ行が複製されています。
感想など
ちょっと実装が大変ですね。
また、フェイルオーバーイベントが短い間隔で発生（またはイベント通知が重複発火）し、イベント通知が連続する・順序が入れ替わる場合などエッジケースでテストしてみないと実用に耐えるかどうかわかりません。
- イベント通知に含まれる新しい Writer インスタンス（ID）をそのまま「信用」してターゲットに加える仕様で大丈夫か？
- インバウンドレプリケーションの復活が少し（数十秒～ 1 分程度）遅れることになっても「フェイルオーバー完了」イベントを拾って、その時点の Aurora クラスターの Writer インスタンスを
describe_db_clustersを使って確認した上でターゲットに加えるべき？
- インバウンドレプリケーションの復活が少し（数十秒～ 1 分程度）遅れることになっても「フェイルオーバー完了」イベントを拾って、その時点の Aurora クラスターの Writer インスタンスを
- 古いターゲット（IP アドレス）の登録解除を行う際、「ターゲットグループにはターゲットが 1 つしかない」と決めつけて良いのか？
- 運用上 Writer インスタンスの IP アドレスしか含まれない想定だが、イベント通知の連続によりたまたま複数の IP アドレスがターゲットとして登録されてしまう事態を想定すべき？
- イベント通知ではなく定期的に Aurora クラスターの Writer インスタンスとターゲットの「差分」が出ていないかを確認してターゲットを切り替えたほうが良くないか？
- イベント通知を利用する方法と比べると、どうしても切り替えのタイミングが遅くなってしまうし確認処理の分コストが掛かってしまうが、イベント処理の取りこぼしや順序の逆転などで誤った状態になるぐらいなら定期確認を採用すべき？
インバウンドレプリケーションが復活（自動復旧）するまで 4 ～ 5 分掛かってしまうのも気になりますね。
MySQL 8.0.22 で実装された（非同期）レプリケーション接続フェイルオーバー機能が使えると良いのに…と思いつつ、その場合は Egress PrivateLink をあらかじめ複数のソース DB インスタンスに接続しておく必要がありそう（注：現状の仕様ではどちらも不可）なので、コスト増が気になります。
色々悩ましいですね。
Discussion