Djangoの非同期タスク処理用ECS Serviceをスケールインから保護するためにECS Task Protectionを使う

SRE AWS Python

はじめに

SREの大木 ( @2357gi )です。最近美味しいお茶漬けを探しています。
今回は非同期タスクの処理を行うECS Serviceのオートスケーリングをいい感じにした話です。
非同期タスクが大量に積まれた際に、それを実行するworkerを柔軟にスケールアウトさせたいですよね。ただ、スケールインする際に処理中のタスクどうするの？中断されちゃったりしない？みたいな課題があると思います。
そこに対して、ECS Task Protectionを使用することによりいい感じにすることができたので紹介したいと思います。
関連技術はだいたい以下の通りです。

  • Django
  • Celery
  • ECS Service
  • Auto Scaling (ECS Service)

背景

Park DirectのバックエンドはAPIを提供するbackend-api(Django)と非同期タスクの処理をおこなうworker(Celery)がそれぞれECS Serviceとして実行されています。
backend-api側の ECS Service は CPU メトリクスをもとにスケーリングをしています。
worker側は当初はオートスケーリング無しでも問題なかったのですが、機能拡張やサービス拡大によってqueue内にタスクが滞留することが多くなってきました。
ビジネス的に重要な処理も非同期タスクで行っていたので、非同期タスクの滞留による遅延に対処することにしました。

オートスケーリングの実装

backend-api - worker間の非同期タスクの受け渡しはRedisをbackendとした分散タスクキューツールを使用しています。
Redisを直接叩くことにより queue 内に積んであるタスクの個数がわかるので、これをCloudWatch Metricsに登録することによりオートスケーリングの設定が可能になりました。
スケールアウトまでは問題ないのですが、スケールインを実施しようとしたときに「実行中のECS Taskで非同期タスクを実行していた場合、実行中のものを巻き込んで終了してしまう問題」が出てきます。
アプリケーション側で全ての非同期タスクに「最後まで実行されたことを確認する機能」や「処理が中断された場合に自動で再実行する機能」を担保することも考えましたが、既存の非同期タスクは全て完全に冪等性が担保されているわけではないので、インフラ面でのアプローチを考えました。

ECS Task保護の実装

ECS Task は ECS API (もしくは ECS エージェントエンドポイント) を叩くことにより 特定のタスクをスケールインから保護することができます。

aws.amazon.com

こちらを利用して、worker内で非同期タスク実行開始時にタスクの保護開始、実行終了時に保護を解除するようにしました。
下記のような関数を開始時、終了時に呼び出しています。

# 実装例
def control_task_protection(is_enabled: bool):
    ecs_agent_url = f"{AWS_ECS_AGENT_URI}/task-protection/v1/state" # Task Protectionを有効にするためのURL

    response = requests.put(
        ecs_agent_url, headers={'Content-Type': 'application/json'},
        data=json.dumps({
            "ProtectionEnabled": is_enabled,
            "ExpiresInMinutes": 1440  # defaultで24時間保持
        })
    )

    if response.status_code != 200:
        _logger.error(f'task protectionに失敗しました.ProtectionEnabled: ${is_enabled} error: {response.text}')

ecs タスク内から自身のタスクを保護する場合は ${AWS_ECS_AGENT_URI}task-protection/v1/state エンドポイントを使用することが推奨されています。

ECS Task保護の排他制御化

workerは設定によって同時に複数個のタスクを処理することができます。
愚直にタスク開始/終了時に保護/解除をしていると、他のタスクが未だ実行されている最中に解除してしまう場合があるので、ecs task arnをkey、実行中のタスクid配列をvalueとしてRedisに保存し、現在該当のworkerでタスクが実行されているかを状態として持つようにしました。
保護や解除を行うときにRedisを確認することで、排他的にTask Protectionを実施できるようになります。
すでに保護されている状態で保護開始のAPIを叩いても何も影響はないので、実装当時は保護開始の際はRedisの値を見ずに行っていました。
しかし、これだとタスク数がスパイクした際にレートリミットに引っかかってしまうことがあった為、現在はRedisの値を見て必要であれば保護開始のAPIを叩くように修正しました。

実装結果

それらの実装を行った結果、以下の画像のようにスケールインが行われてもタスクが保護され終了していない状態になりました。

スケールインされてECS Task数が1個が適切という状態になっていますが、保護された ECS Taskが存在するため、2個稼働している状態になっています🎉

タスク保護の解除に失敗した際、ECS Serviceを更新しても古いタスクが生き残り続けることになるので何らかのアラートを仕込んでおくことをお勧めします。
(我々はTask Protectionの失敗ログをトリガーとするアラートとランブックを作成しています。)

これにより、非同期タスクが積まれた際に柔軟なスケールアウトを行い、効率的に捌くことができるようになりました 🎉

左から[queueに積まれている非同期タスク数]、[workerのECS Task数]、[処理された非同期タスク数]

それはそれとして、auto scaling はできてもDBの負荷など他の要因でスケールアウトの上限があり、一定時間queueの待機時間が発生する場合もありました。ユーザー影響的に遅延が許容できない非同期タスクも存在したので、別途以下の対応もSREチームで行いました。

nealle-dev.hatenablog.com

採用しなかったアプローチ

重要度の高いタスクはアドホックなworkerで実行する

当初は長時間かかり且つ中断が許されない非同期タスクを実行する際に、独立した専用のworkerをアドホックに立ち上げるというアプローチも考えました。 しかし、アドホックタスクとして実装するタスクとそれ以外の線引きをどうするのかといった線引きが難しいことや、前述した解決方法に比べて劣ると判断したため採用しませんでした。

おしまい