Saga

1. 背景

複数のマイクロサービスを横断したトランザクションを実現したい

• システムをマイクロサービスアーキテクチャとして設計している場合、ある処理を実行するにあたって、複数のマイクロサービスに対してそれぞれデータ投入が必要になる場面がある

2. 課題

如何にしてマイクロサービスを横断したトランザクション/ロールバックを実装するか

• マイクロサービスとして実装されたシステムでは、トランザクションはマイクロサービス単位に行われることになる（＝分散トランザクション）
• データの投入自体はそれぞれAPIに対してデータを送信すればよいが、途中でエラーが発生した場合のリカバリやロールバックをどのようにして制御するかが問題になってくる

3. 解決策

各マイクロサービスにロールバックAPIを実装する

• マイクロサービスは（原則として）独立して稼働している為、RDBMSのトランザクションを利用することはできない
• 各マイクロサービスが独自にロールバック処理を実装しておく必要がある（データの削除or無効化処理を行うAPIを実装）

ローカルトランザクションを管理するモジュールを実装する

• 各マイクロサービスに対してデータの投入とエラー発生時のロールバックを管理するモジュールを実装し、そのモジュールがトランザクションを制御する
• 概念実装はAppendix-1を参照せよ

4. メリット

• 複数のマイクロサービスを跨いだトランザクションの実現

5. デメリット

• 総合的な設計/実装コストの増加
• システム全体の複雑性の増加

6. 注意事項

• システムの複雑性が増加する為、極力控えたいパターンだがマイクロサービスアーキテクチャを採用する以上、避けて通れない問題でもある

参考資料

• TECHSCORE｜マイクロサービスにおけるトランザクション【Saga】 | TECHSCORE BLOG

Appendix-1 シンプルなSagaパターンのサンプル実装

class SagaOperator:
    def __init__(self, *services):
        self.services = services
        self.completed = []

    def run(self):
        try:
            self._execute()
        catch:
            self._rollback()

    def _execute(self):
        for s in self.service:
            s.execute()
            self.completed.append(s)

    def _rollback(self):
        for s in self.completed:
            s.rollback()

class Service(metaclass=ABCMeta):
    @abstractmethod
    def execute(self):
        """ マイクロサービスにデータを登録する処理 """
        pass

    @abstractmethod
    def rollback(self):
        """ マイクロサービスに登録したデータを削除する処理 """
        pass

class Service01(Service):
    def execute(self):
        # APIにデータを送信して結果を評価する処理が書いてあると考えよ
        print("Service01 complete")

    def rollback(self):
        # APIにデータの削除を要求する処理が書いてあると考えよ
        print("Service01 rollback")

class Service02(Service):
    def execute(self):
        # APIにデータを送信して結果を評価する処理が書いてあると考えよ
        print("Service02 complete")

    def rollback(self):
        # APIにデータの削除を要求する処理が書いてあると考えよ
        print("Service02 rollback")

class Service03(Service):
    def execute(self):
        # APIにデータを送信して結果を評価する処理が書いてあると考えよ
        print("Service03 complete")

    def rollback(self):
        # APIにデータの削除を要求する処理が書いてあると考えよ
        print("Service03 rollback")


if __name__ == '__main__':
    op = SagaOperator(
        Service01(),
        Service02(),
        Service03(),
    )
    op.run()

• このサンプルではrollbackが失敗した際のリカバリ処理が省略されていることに注意が必要
• ロールバックもネットワーク経由でマイクロサービスに要求する必要がある以上、ネットワークエラーによってリクエストが到達しないなど、ロールバックが実行されないケースが発生し得る
• したがって、Sagaの実行はリトライが可能であるように実装する必要がある（例えばメッセージキューで実行/完了を管理するなど）