リーダー選定パターン

Azure Blob Storage

他のインスタンスを管理する役割を担うリーダーとして 1 つのインスタンスを選択することで、分散アプリケーション内で連携するインスタンスのコレクションによって実行されるアクションを調整します。これにより、インスタンスが互いに競合して、共有リソースとの競合を引き起こしたり、他のインスタンスが実行されている作業に誤って干渉したりしないようにできます。

コンテキストと問題

一般的なクラウドアプリケーションには、協調した動きをする多くのタスクがあります。これらのタスクはすべて、同じコードを実行し、同じリソースへのアクセスを必要とするインスタンスの可能性があります。また、複雑な計算の個々の部分を実行するために、複数のタスクが並列で動作する場合もあります。

タスクインスタンスは、大半の時間は個々に実行されますが、インスタンスが互いに競合して共有リソースとの競合を引き起こしたり、他のタスクインスタンスが実行されている作業に誤って干渉したりしないように、各インスタンスのアクションを調整する必要もあります。

次に例を示します。

水平方向のスケーリングを実装するクラウドベースシステムでは、同一タスクの複数のインスタンスを、別のユーザーに対する各インスタンスと同時に実行できます。これらのインスタンスが共有リソースに書き込みを行う場合、各インスタンスが他のインスタンスによって行われた変更を上書きしないようにアクションを調整する必要があります。
タスクが複雑な計算の個々の要素を並列で実行している場合、すべてが完了した時点で結果が集計される必要があります。

タスクインスタンスはすべてピアなので、コーディネーターやアグリゲーターとして機能できる自然なリーダーは存在しません。

解決策

リーダーとして機能するように、単一のタスクインスタンスが選定される必要があります。また、このインスタンスは、他の下位のタスクインスタンスのアクションを調整する必要があります。すべてのタスクインスタンスが同じコードを実行している場合、それぞれのインスタンスがリーダーとして機能できます。そのため、2 つ以上のインスタンスが同時にリーダーの役割を担うことがないように、選定のプロセスを注意深く管理する必要があります。

システムは、リーダーを選定するための堅牢なメカニズムを提供する必要があります。選定のメソッドでは、ネットワークの停止やプロセスの失敗などの事態に対処する必要があります。多くのソリューションでは、下位のタスクインスタンスは、任意のタイプのハートビートメソッド経由またはポーリングによって、リーダーを監視します。指定したリーダーが予期せずに終了した場合や、ネットワーク障害によってリーダーを下位のタスクインスタンスで使用できない場合は、新しいリーダーを選定する必要があります。

分散した環境にある一連のタスクからリーダーを選定するには、いくつかの戦略があります。

最低ランクのインスタンス ID またはプロセス ID のタスクインスタンスを選択する。
共有されている分散ミューテックスを獲得するために競わせる。ミューテックスを獲得した最初のタスクインスタンスがリーダーになります。ただし、システムは、リーダーが終了したりシステムの他の部分から切り離されたりした場合に、別のタスクインスタンスがリーダーになれるようにミューテックスが解放されることを保証する必要があります。
Bully Algorithm (ブリーアルゴリズム) や Ring Algorithm (リングアルゴリズム) など、一般的なリーダー選定のアルゴリズムの 1 つを実装する。これらのアルゴリズムでは、選定の各候補が一意の ID を保持し、他の候補と確実に通信できることを前提としています。

問題と注意事項

このパターンの実装方法を決めるときには、以下の点に注意してください。

リーダー選定のプロセスでは、一時的および永続的な障害に対して復元性を備える必要があります。
リーダーに障害が発生した場合や、使用不可能になった場合 (通信障害に起因するなど) に、検出できる必要があります。検出の際に必要とされる迅速さは、システムによって異なります。一部のシステムは、一時的な障害が収束するまでなどの短い時間は、リーダー不在で機能できる場合があります。それ以外の場合は、リーダーの障害をただちに検出して、新しい選定をトリガーする必要があります。
水平方向の自動スケーリングを実装するシステムでは、システムが規模を縮小し、一部のコンピューティングリソースをシャットダウンした場合に、リーダーが終了する場合があります。
共有されている分散ミューテックスを使用すると、ミューテックスを提供している外部サービスへの依存が可能になります。サービスは、単一障害点を構成しています。この障害点が何らかの理由で利用できなくなった場合、システムはリーダーを選定できなくなります。
簡単な方法は、1 つの専用プロセスをリーダーとして使用することです。ただし、プロセスが失敗した場合、再起動時に大幅な遅延が発生することがあります。他のプロセスがリーダーによる操作の調整を待機している場合、結果として生じる待機時間が、それらのプロセスのパフォーマンスと応答時間に影響する可能性があります。
いずれかのリーダー選定アルゴリズムを手動で実装すると、コードの調整と最適化を最も柔軟に行うことができます。
リーダーがシステムのボトルネックにならないようにします。リーダーの目的は、下位のタスクの作業を調整することであり、必ずしもこの作業自体に参加する必要はありません。ただし、タスクがリーダーとして選定されていない場合は、作業に参加できる必要があります。

このパターンを使用する状況

クラウドホストソリューションなどの分散アプリケーション内のタスクに慎重な調整が必要であり、自然なリーダーが存在しない場合は、このパターンを使用します。

以下の場合は、このパターンの使用は適していません。

自然なリーダーまたはリーダーとして常に機能する専用のプロセスがある。たとえば、タスクインスタンスを調整するシングルトンプロセスを実装できる場合があります。このプロセスが失敗したり正常でなくなった場合、システムはプロセスをシャットダウンして、再起動できます。
タスク間の調整が、より簡易的な方法を使用して実現できる。たとえば、複数のタスクインスタンスが共有リソースへのアクセス調整を必要としている場合、アクセスを制御するために楽観的ロックまたは排他的ロックを使用すると、より優れたソリューションになります。
サードパーティのソリューションがよりふさわしい。たとえば、(Apache Hadoop に基づく) Microsoft Azure HDInsight サービスは、Apache Zookeeper が提供するサービスを使用してマップを調整し、データを収集して集計するタスクを減らします。

ワークロード設計

設計者は、Azure Well-Architected Framework の柱で説明されている目標と原則に対処するために、ワークロードの設計でどのようにリーダー選定パターンを使用できるかを評価する必要があります。次に例を示します。

重要な要素	このパターンが柱の目標をサポートする方法
信頼性設計の決定により、ワークロードが誤動作に対して復元力を持ち、障害発生後も完全に機能する状態に回復することができます。	このパターンでは、作業が確実にリダイレクトされるため、ノードの誤動作の影響が軽減します。また、リーダーが誤動作した場合に備えて、コンセンサスアルゴリズムによるフェールオーバーも実装されます。 - RE: 05冗長性 - RE:07 自己復旧:

設計決定と同様に、このパターンで導入される可能性のある他の柱の目標とのトレードオフを考慮してください。

例

GitHub 上の Leader Election サンプルは、Azure Storage の BLOB でリースを使用して、共有分散ミューテックスを実装するメカニズムを提供する方法を示しています。このミューテックスは、利用可能なワーカーインスタンスのグループ間でリーダーを選定するために使用できます。リースを取得した最初のインスタンスがリーダーに選定され、リースを解放するか、リースを更新できなくなるまでリーダーのままになります。他のワーカーインスタンスは、リーダーが利用できなくなった場合に備えて、BLOB リースの監視を続行できます。

Blob リースは、Blob 経由での排他的な書き込みロックです。単一の Blob は、任意の時点で 1 つのリースだけの対象になることができます。ワーカーインスタンスは指定した BLOB に対するリースを要求でき、他のワーカーインスタンスが同じ BLOB に対するリースを保持していなければ、リースが付与されます。そうでない場合、要求は例外をスローします。

リーダーインスタンスがリースを無期限に保持する障害を回避するために、リースには有効期限を指定してください。この有効期限が切れると、リースは使用可能になります。ただし、インスタンスがリースを保持している間、そのインスタンスはリースの更新を要求でき、要求に応じてリース期間が延長されます。リーダーインスタンスは、リースを保持する必要がある場合、このプロセスを継続的に繰り返すことができます。 Blob をリースする方法の詳細については、「Lease Blob (REST API) (Blob のリース (REST API))」を参照してください。

以下の C# の例にある BlobDistributedMutex クラスには、ワーカーインスタンスから指定した BLOB に対するリースの取得を試行できるようにする、RunTaskWhenMutexAcquired メソッドが含まれています。 Blob (名前、コンテナー、およびストレージアカウント) の詳細は、BlobDistributedMutex オブジェクトが作成されたときに (このオブジェクトは、サンプルコードに含まれている単純な構造体です)、BlobSettings オブジェクトのコンストラクターに渡されます。また、コンストラクターは Task も受け取ります。これは、ワーカーインスタンスが BLOB に対するリースの取得に成功してリーダーに選定された場合に、ワーカーインスタンスで実行する必要のあるコードを参照します。リースを取得する低レベルの詳細情報を処理するコードが、BlobLeaseManager という名前の個々のヘルパークラスで実装されていることに注意してください。

public class BlobDistributedMutex
{
  ...
  private readonly BlobSettings blobSettings;
  private readonly Func<CancellationToken, Task> taskToRunWhenLeaseAcquired;
  ...

  public BlobDistributedMutex(BlobSettings blobSettings,
           Func<CancellationToken, Task> taskToRunWhenLeaseAcquired, ... )
  {
    this.blobSettings = blobSettings;
    this.taskToRunWhenLeaseAcquired = taskToRunWhenLeaseAcquired;
    ...
  }

  public async Task RunTaskWhenMutexAcquired(CancellationToken token)
  {
    var leaseManager = new BlobLeaseManager(blobSettings);
    await this.RunTaskWhenBlobLeaseAcquired(leaseManager, token);
  }
  ...

上記のコードサンプル内の RunTaskWhenMutexAcquired メソッドは、実際にリースを取得する下記のサンプルコードに示された RunTaskWhenBlobLeaseAcquired メソッドを呼び出します。 RunTaskWhenBlobLeaseAcquired メソッドが非同期で実行されます。リースが正常に取得された場合は、そのワーカーインスタンスがリーダーに選定されたことになります。 taskToRunWhenLeaseAcquired デリゲートの目的は、他のワーカーインスタンスを調整する作業を実行することです。リースが取得されなかった場合は、別のワーカーインスタンスがリーダーに選定され、現在のワーカーインスタンスは下位のままになります。 TryAcquireLeaseOrWait メソッドは、リースを取得する BlobLeaseManager オブジェクトを使用しているヘルパーメソッドであることに注意してください。

  private async Task RunTaskWhenBlobLeaseAcquired(
    BlobLeaseManager leaseManager, CancellationToken token)
  {
    while (!token.IsCancellationRequested)
    {
      // Try to acquire the blob lease.
      // Otherwise wait for a short time before trying again.
      string? leaseId = await this.TryAcquireLeaseOrWait(leaseManager, token);

      if (!string.IsNullOrEmpty(leaseId))
      {
        // Create a new linked cancellation token source so that if either the
        // original token is canceled or the lease can't be renewed, the
        // leader task can be canceled.
        using (var leaseCts =
          CancellationTokenSource.CreateLinkedTokenSource(new[] { token }))
        {
          // Run the leader task.
          var leaderTask = this.taskToRunWhenLeaseAcquired.Invoke(leaseCts.Token);
          ...
        }
      }
    }
    ...
  }

リーダーによって開始されたタスクは、非同期で実行されます。このタスクの実行中、次のサンプルコードに示された RunTaskWhenBlobLeaseAcquired メソッドは、定期的にリースの更新を試行しています。これにより、ワーカーインスタンスが確実にリーダーのままでいられます。サンプルソリューションでは、他のワーカーインスタンスがリーダーに選定されないように、更新要求間の待ち時間は、リースの期間として指定された時間よりも短くなっています。何らかの理由で更新に失敗した場合、リーダー固有のタスクは取り消されます。

リースの更新に失敗した場合、またはタスクが取り消された場合 (たとえば、ワーカーインスタンスがシャットダウンされた結果として)、リースは解放されます。この時点で、このワーカーインスタンスまたは別のワーカーインスタンスが、リーダーとして選定される可能性があります。以下のコードの抜粋は、プロセスのこの部分を示しています。

  private async Task RunTaskWhenBlobLeaseAcquired(
    BlobLeaseManager leaseManager, CancellationToken token)
  {
    while (...)
    {
      ...
      if (...)
      {
        ...
        using (var leaseCts = ...)
        {
          ...
          // Keep renewing the lease in regular intervals.
          // If the lease can't be renewed, then the task completes.
          var renewLeaseTask =
            this.KeepRenewingLease(leaseManager, leaseId, leaseCts.Token);

          // When any task completes (either the leader task itself or when it
          // couldn't renew the lease) then cancel the other task.
          await CancelAllWhenAnyCompletes(leaderTask, renewLeaseTask, leaseCts);
        }
      }
    }
  }
  ...
}

KeepRenewingLease メソッドは、リースを更新するために BlobLeaseManager オブジェクトを使用するもう 1 つのヘルパーメソッドです。 CancelAllWhenAnyCompletes メソッドは、最初の 2 つのパラメーターに指定されたタスクをキャンセルします。次の図では、リーダーを選定して操作を調整するタスクを実行するために、BlobDistributedMutex クラスを使用しています。

図 1 は、BlobDistributedMutex クラスの関数を示しています

次のコード例は、ワーカーインスタンス内で BlobDistributedMutex クラスを使用する方法を示しています。このコードでは、リースのコンテナーの Azure Blob Storage にある MyLeaderCoordinatorTask という名前の BLOB に対するリースを取得し、ワーカーインスタンスがリーダーに選定された場合に、MyLeaderCoordinatorTask メソッドに定義されたコードが実行されるように指定しています。

// Create a BlobSettings object with the connection string or managed identity and the name of the blob to use for the lease
BlobSettings blobSettings = new BlobSettings(storageConnStr, "leases", "MyLeaderCoordinatorTask");

// Create a new BlobDistributedMutex object with the BlobSettings object and a task to run when the lease is acquired
var distributedMutex = new BlobDistributedMutex(
    blobSettings, MyLeaderCoordinatorTask);

// Wait for completion of the DistributedMutex and the UI task before exiting
await distributedMutex.RunTaskWhenMutexAcquired(cancellationToken);

...

// Method that runs if the worker instance is elected the leader
private static async Task MyLeaderCoordinatorTask(CancellationToken token)
{
  ...
}

サンプルソリューションでは、次の点に注意してください。

Blob は潜在的な単一障害点になります。 Blob service が利用できない場合、またはアクセスできない場合、リーダーはリースを更新できず、他のワーカーインスタンスもリースを取得できなくなります。この場合、どのワーカーインスタンスもリーダーとして機能できません。しかし、Blob サービスは復元性を備えた設計になっているため、Blob サービスの致命的な障害は極めて発生しにくいと考えられています。
リーダーによって実行されているタスクが停止しても、リーダーは引き続きリースの更新を行って、他のワーカーインスタンスによるリースの取得を防ぎ、タスクを調整するためにリーダーの役割を引き継ぐことがあります。実際の運用では、リーダーの正常性を頻繁にチェックする必要があります。
選定プロセスは非決定性です。どのワーカーインスタンスが BLOB リースを取得してリーダーになるかを予測することはできません。
Blob リースのターゲットとして使用される Blob は、他の目的には使用できません。ワーカーインスタンスがこの BLOB にデータを格納しようとした場合、そのワーカーインスタンスがリーダーになっていて BLOB リースを保持していない限り、このデータにはアクセスできなくなります。

次のステップ

このパターンを実装する場合は、次のガイダンスも関連している可能性があります。

このパターンには、ダウンロード可能なサンプルアプリケーションが用意されています。
自動スケールガイダンス。アプリケーションの負荷は変化するため、タスクホストのインスタンスを開始および停止することが可能です。自動スケーリングは、ピーク時の処理中のスループットとパフォーマンスの維持に役立ちます。
タスクベースの非同期パターン。
Bully Algorithm (ブリーアルゴリズム) を示したサンプル。
Ring Algorithm (リングアルゴリズム) を示したサンプル。
Apache ZooKeeper の Apache Curator クライアントライブラリ。
MSDN の記事「Lease Blob (REST API) (Blob のリース (REST API))」。