課題: シンクなしでの同期

異なる地域にある複数のS3バケット間でオブジェクトを同期するのは、データでできた猫を追いかけるようなものです。しかも、その猫たちは目を離すと増えてしまうのです。私たちが直面している主な課題は次のとおりです:

  • 異なる地域からの同時更新
  • ネットワーク分割による一時的な孤立
  • バケット間のバージョンの不一致
  • 可用性を犠牲にせずに最終的な一貫性を確保する必要性

従来のロック機構や中央調整役？それらはサハラ砂漠でのチョコレートティーポットのように役に立ちません。もっとイベントフルなものが必要です。

CRDTの登場: 分散システムの平和維持者

Conflict-free Replicated Data Types (CRDTs) は分散システムの無名の英雄です。これらはネットワーク内の複数のコンピュータに複製できるデータ構造で、複製は独立して同時に更新でき、複製間の調整が不要であり、結果として生じる不整合を常に数学的に解決することが可能です。

私たちのS3レプリケーターでは、Grow-Only Counter (G-Counter) という特定のタイプのCRDTを使用します。これはバージョンの不一致を処理するのに最適で、増加のみを許可し、減少は許可しません。データのバージョン番号のための一方通行のようなものです。

G-Counterの実装

以下はPythonでのG-Counterの簡単な実装です:


class GCounter:
    def __init__(self):
        self.counters = {}

    def increment(self, node_id):
        if node_id not in self.counters:
            self.counters[node_id] = 0
        self.counters[node_id] += 1

    def merge(self, other):
        for node_id, count in other.counters.items():
            if node_id not in self.counters or self.counters[node_id] < count:
                self.counters[node_id] = count

    def value(self):
        return sum(self.counters.values())

このG-Counterは各ノード（私たちの場合、各S3バケット）が独立してカウンターを増加させることを可能にします。同期の時が来たら、カウンターをマージし、各ノードの最大値を取ります。

Lambda@Edge: 分散型ウォッチドッグ

CRDTを手に入れた今、S3バケット間で変更を伝播する方法が必要です。そこで登場するのがLambda@Edgeです。AWSのソリューションで、AWS EdgeロケーションでグローバルにLambda関数を実行できます。世界のあらゆる角に小さく効率的なロボットがいて、すぐに行動を起こす準備ができているようなものです。

Lambda@Edgeを使用して次のことを行います:

  1. S3バケットの変更を検出する
  2. ローカルG-Counterを更新する
  3. 変更を他のバケットに伝播する
  4. 異なるバケットからG-Counterをマージする

Lambda@Edgeの設定

まず、S3オブジェクトの作成または更新時にトリガーされるLambda関数を作成しましょう:


import boto3
import json
from gcounter import GCounter

def lambda_handler(event, context):
    # イベントからバケットとオブジェクト情報を抽出
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    # S3クライアントを初期化
    s3 = boto3.client('s3')

    # オブジェクトメタデータから現在のG-Counterを読み取る
    try:
        response = s3.head_object(Bucket=bucket, Key=key)
        current_counter = json.loads(response['Metadata'].get('g-counter', '{}'))
    except:
        current_counter = {}

    # 新しいG-Counterを作成し、現在のものとマージ
    g_counter = GCounter()
    g_counter.counters = current_counter
    g_counter.increment(bucket)

    # 新しいG-Counterでオブジェクトメタデータを更新
    s3.copy_object(
        Bucket=bucket,
        CopySource={'Bucket': bucket, 'Key': key},
        Key=key,
        MetadataDirective='REPLACE',
        Metadata={'g-counter': json.dumps(g_counter.counters)}
    )

    # 他のバケットに変更を伝播
    propagate_changes(bucket, key, g_counter)

def propagate_changes(source_bucket, key, g_counter):
    # 同期するすべてのバケットのリスト
    buckets = ['bucket1', 'bucket2', 'bucket3']  # バケット名をここに追加

    s3 = boto3.client('s3')

    for target_bucket in buckets:
        if target_bucket != source_bucket:
            try:
                # ソースバケットからオブジェクトを取得
                response = s3.get_object(Bucket=source_bucket, Key=key)
                
                # ターゲットバケットにオブジェクトをコピー
                s3.put_object(
                    Bucket=target_bucket,
                    Key=key,
                    Body=response['Body'].read(),
                    Metadata={'g-counter': json.dumps(g_counter.counters)}
                )
            except Exception as e:
                print(f"Error propagating changes to {target_bucket}: {str(e)}")

このLambda関数は、G-Counterの更新と他のバケットへの変更の伝播を行います。まるでハイパーアクティブなタコのように、同時にすべてのバケットに手を伸ばします。

バージョンの不一致の処理

さて、ここで問題となるのはバージョンの不一致です。ここでG-Counterが役立ちます。増加のみを許可するため、すべてのバケットでオブジェクトの最新バージョンを決定するのに使用できます。

バージョンの不一致を処理するためにLambda関数を次のように変更できます:


def resolve_version_conflict(bucket, key, g_counter):
    s3 = boto3.client('s3')

    # オブジェクトのすべてのバージョンを取得
    versions = s3.list_object_versions(Bucket=bucket, Prefix=key)['Versions']

    # 最も高いG-Counter値を持つバージョンを見つける
    latest_version = max(versions, key=lambda v: GCounter().merge(json.loads(v['Metadata'].get('g-counter', '{}'))))

    # 最新バージョンが現在のバージョンでない場合、更新する
    if latest_version['VersionId'] != versions[0]['VersionId']:
        s3.copy_object(
            Bucket=bucket,
            CopySource={'Bucket': bucket, 'Key': key, 'VersionId': latest_version['VersionId']},
            Key=key,
            MetadataDirective='REPLACE',
            Metadata={'g-counter': json.dumps(g_counter.counters)}
        )

この関数はオブジェクトのすべてのバージョンをチェックし、最も高いG-Counter値を持つバージョンを現在のバージョンとして設定します。まるで時間旅行をする歴史家のように、常に最新の歴史を提示します。

全体像: すべてをまとめる

では、ここで何を構築したのでしょうか？以下に分解してみましょう:

  1. バージョン管理と競合解決を処理するためのG-Counter CRDT
  2. Lambda@Edge関数:
    • S3バケットの変更を検出する
    • G-Counterを更新する
    • 他のバケットに変更を伝播する
    • バージョンの競合を解決する

このシステムは、複数のS3バケット間で最終的な一貫性を維持しながら可用性を犠牲にしません。まるで自己組織化し、自己修復するデータエコシステムのようです。

潜在的な落とし穴と考慮事項

これを本番環境で実装する前に、次の点に注意してください:

  • Lambda@Edgeには制限があり、実行時間やペイロードサイズが含まれます。大きなオブジェクトの場合、チャンク戦略を実装する必要があるかもしれません。
  • このソリューションはネットワーク分割が一時的であることを前提としています。長期間の分割の場合、追加の調整メカニズムが必要になるかもしれません。
  • G-Counterは時間とともに成長します。頻繁に更新される長寿命のオブジェクトの場合、剪定戦略を実装する必要があるかもしれません。
  • 本番環境にデプロイする前に、ステージング環境で徹底的にテストしてください。分散システムは扱いにくいものです！

まとめ: なぜ手間をかけるのか？

「なぜこんなに手間をかけるのか？AWSの組み込みのレプリケーションを使えばいいのでは？」と思うかもしれません。確かにそうです。しかし、私たちのソリューションにはいくつかのユニークな利点があります:

  • 異なるAWSアカウントや地域間で動作し、単一のアカウント内だけではありません。
  • ネットワーク分割や同時更新に直面しても、より強力な一貫性の保証を提供します。
  • より柔軟で、特定のビジネスロジックやデータモデルに合わせてカスタマイズできます。

最終的に、このアプローチはデータ同期プロセスに対するより細かい制御を提供します。まるで分散データオーケストラの指揮者のように、すべての楽器（この場合、すべてのS3バケット）が完璧な調和で演奏することを保証します。

考えるための材料

このソリューションを実装する際、次の質問を考慮してください:

  • 削除を処理するためにこのシステムをどのように変更しますか？
  • このアプローチをS3以外のAWSサービスに拡張することは可能ですか？
  • 分散クラウドアーキテクチャで役立つ他のタイプのCRDTは何ですか？

分散システムの世界では、万能の解決策はありません。しかし、CRDTとLambda@Edgeをツールキットに持っていれば、最も困難なデータ同期の問題にも立ち向かう準備が整っています。さあ、進んで、データが常に同期されることを願っています！