はじめに

分散システムでは、ネットワーク障害、ノードの失敗、データの一貫性の問題など、さまざまなエラーが発生します。これらの課題に対処し、耐障害性と信頼性を確保することは、システムの成功に不可欠です。この記事では、エラー処理のためのパターン（リトライ、サーキットブレーカー）、ロギングとモニタリングのベストプラクティス、そしてPythonを使った実装を解説します。実際のコード例として、CeleryとgRPCを使用した耐障害性の高いシステムを構築します。

分散システムのエラーの種類

分散システムでよく見られるエラーには以下が含まれます：

ネットワーク障害：通信の遅延や切断。
ノードの失敗：サーバーのクラッシュやリソース不足。
データの一貫性：異なるノード間でデータが同期しない問題。
タイムアウト：リクエストが規定時間内に完了しない。

これらに対処するため、リトライ、サーキットブレーカー、フォールバックなどのパターンを活用します。

リトライパターンの実装

tenacityを使用したリトライ

tenacityライブラリは、リトライロジックを簡潔に実装するためのツールです。インストール：

pip install tenacity

以下の例では、gRPCリクエストにリトライを実装します：

from tenacity import retry, stop_after_attempt, wait_exponential
import grpc
import compute_pb2
import compute_pb2_grpc

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def make_grpc_request(number):
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = compute_pb2_grpc.ComputeServiceStub(channel)
        response = stub.ComputeSquare(compute_pb2.ComputeRequest(number=number))
        return response.result

try:
    result = make_grpc_request(5)
    print(f"結果: {result}")
except grpc.RpcError as e:
    print(f"エラー: {e}")

出力例（成功時）：

結果: 25

tenacityは、最大3回のリトライを試み、指数バックオフ（2秒から10秒の待機時間）で再試行します。

サーキットブレーカーの実装

pybreakerを使用したサーキットブレーカー

サーキットブレーカーは、連続するエラーを検知し、一時的にリクエストを停止することでシステムを保護します。インストール：

pip install pybreaker

以下の例では、Celeryタスクにサーキットブレーカーを適用します：

from pybreaker import CircuitBreaker
from celery_app import app as celery_app

# サーキットブレーカーの設定
breaker = CircuitBreaker(fail_max=5, reset_timeout=60)

@breaker
@celery_app.task
def send_email(recipient, subject, body):
    # メール送信をシミュレート（失敗する場合を想定）
    import random
    if random.random() < 0.7:  # 70%の確率で失敗
        raise ValueError("メール送信に失敗")
    return f"メールを {recipient} に送信しました: {subject}"

try:
    task = send_email.delay('user@example.com', 'テスト', 'こんにちは')
    result = task.get(timeout=10)
    print(result)
except CircuitBreakerError:
    print("サーキットブレーカーがオープン状態です")
except Exception as e:
    print(f"エラー: {e}")

出力例（失敗時）：

サーキットブレーカーがオープン状態です

サーキットブレーカーは、5回連続の失敗で回路を「オープン」にし、60秒間リクエストをブロックします。

ロギングとモニタリング

構造化ロギング

structlogを使用した構造化ロギングで、エラー情報を整理します。インストール：

pip install structlog

例：

import structlog
import logging

# ロギング設定
structlog.configure(
    processors=[structlog.processors.JSONRenderer()],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
)
logger = structlog.get_logger()

def process_task(data):
    logger.info("タスク開始", data_id=data['id'])
    try:
        result = data['value'] ** 2
        logger.info("タスク成功", result=result)
        return result
    except Exception as e:
        logger.error("タスク失敗", error=str(e))
        raise

data = {'id': 1, 'value': 5}
process_task(data)

出力例（JSON形式）：

{"event": "タスク開始", "data_id": 1}
{"event": "タスク成功", "result": 25}

structlogは、モニタリングツール（例：Elasticsearch）との統合に適しています。

PrometheusとGrafanaによるモニタリング

PrometheusとGrafanaでタスクのメトリクスを監視します。インストール：

pip install prometheus_client

例：Celeryタスクのメトリクスを収集：

from prometheus_client import Counter, start_http_server
from celery_app import app as celery_app

# メトリクス定義
task_success = Counter('task_success_total', 'Successful tasks')
task_failure = Counter('task_failure_total', 'Failed tasks')

@celery_app.task
def process_data(data):
    try:
        result = data ** 2
        task_success.inc()
        return result
    except Exception as e:
        task_failure.inc()
        raise

# Prometheusサーバー起動
start_http_server(8000)

# タスク実行
process_data.delay(5)

Prometheusをhttp://localhost:8000で確認し、Grafanaで可視化します。

実際の応用例

以下の例では、CeleryとgRPCを組み合わせ、リトライとサーキットブレーカーを適用したシステムを構築します：

# app.py
from flask import Flask, jsonify
from celery_app import app as celery_app
from pybreaker import CircuitBreaker
from tenacity import retry, stop_after_attempt, wait_exponential
import grpc
import compute_pb2
import compute_pb2_grpc
import structlog

app = Flask(__name__)
breaker = CircuitBreaker(fail_max=5, reset_timeout=60)
logger = structlog.get_logger()

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def call_grpc_service(number):
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = compute_pb2_grpc.ComputeServiceStub(channel)
        response = stub.ComputeSquare(compute_pb2.ComputeRequest(number=number))
        return response.result

@celery_app.task
@breaker
def process_task(number):
    logger.info("タスク開始", number=number)
    try:
        result = call_grpc_service(number)
        logger.info("タスク成功", result=result)
        return result
    except Exception as e:
        logger.error("タスク失敗", error=str(e))
        raise

@app.route('/compute/<int:number>')
def trigger_task(number):
    task = process_task.delay(number)
    return jsonify({'task_id': task.id})

if __name__ == '__main__':
    app.run(debug=True)

このシステムは、gRPCで計算を処理し、Celeryで非同期実行、リトライとサーキットブレーカーで耐障害性を確保します。

注意点とベストプラクティス

リトライ設定：過剰なリトライはリソースを浪費するため、適切な上限を設定。
サーキットブレーカー：オープン状態の監視とリカバリ戦略を明確化。
ロギング：構造化ログを使用して、モニタリングツールとの統合を簡素化。
メトリクス：Prometheusでレイテンシやエラーレートを追跡。

まとめ

この記事では、分散システムのエラー処理と耐障害性を高める方法を解説しました。tenacity、pybreaker、structlog、Prometheusを活用し、信頼性の高いシステムを構築しました。次回は、クラウド（Kubernetes、AWS ECS）での分散システムのデプロイを紹介します。

この記事が役に立ったら、いいねやストックをお願いします！コメントで質問やフィードバックもお待ちしています！