はじめに

こんにちは、バックエンドエンジニアのtakayukiです。
私のあるリリースをきっかけとしてRedisのコネクション数が増加しました。この記事にはRedisのコネクション数が増加した際の調査内容と修正の備忘録を書きます。

起きたこと

くらしのマーケットでは、非同期にタスクを実行する仕組みとしてCeleryというタスクキューを使用し、メッセージングキュー(メッセージブローカー)にはRedisを採用しています。

先日、このCeleryを利用した機能をリリースしたのですが、リリースの翌日にRedisのコネクション数が一定の基準値を超え、障害検知のアラートが鳴りました。
※くらしのマーケットではAWSのElastiCacheのRedisを使用しており最大接続数は65000ですがアラートは余裕をもって設定しています。

Redisのkeyとサイトへのリクエストを調査すると、私のリリースでCeleryのtaskを呼ぶ回数が増加し、それに伴ってコネクション数も増加していました。

調査

くらしのマーケットではバックエンドにPython製webフレームワークのDjangoを使用しており、そこからCeleryを経由してRedisにエンキューしています。今回の私が作成したエンドポイントや他のエンドポイントでは、以下のようなCeleryのラッパークラスをインスタンス化してエンキューしていました。

from datetime import datetime
from celery import Celery

class CeleryClass(object):
    def __init__(self):
        self.app = Celery()
        self.app.config_from_object("config")

    def send_task(self, task_name: str, args: any, queue: str, eta: datetime=None):
        self.app.send_task(task_name, args=args, queue=queue, eta=eta)

# 使用するとき
Celery = CeleryClass()
Celery.send_task("SampleTask", args=[""], queue="user")

(一部名前など改変しています)

擬似コードなどを作って調べたところ、CeleryはCelery()がインスタンス化される時にコネクションを接続しているようです。しかし、このクラスを見ると共通クラスがインスタンス化される度にCelery()を呼んでいます。
どうやらこれが原因のようです。

解決策

解決策としては以下の2つが考えられました。

Celeryにコネクションを明示的に閉じるコマンドが存在するので、コネクションが貼られる度に閉じる
生成するインスタンスを1つに制限するSingletonパターンを参考にしてCelery()インスタンスを使い回す

Celeryにはコネクションプール機能があり、明示的に閉じてしまうとその機能を生かせないため、2を選択し以下のように修正しました。

from datetime import datetime
from celery import Celery

class CeleryClass(object):
    __instance = None
    @staticmethod
    def get_instance():
        if CeleryClass.__instance is None:
            CeleryClass()
        return CeleryClass.__instance

    def __init__(self):
        if CeleryClass.__instance is not None:
            raise Exception("error")
        else:
            self.app = Celery()
            self.app.config_from_object("config")
            CeleryClass.__instance = self

    def send_task(self, task_name: str, args: any, queue: str, eta: datetime=None):
        self.app.send_task(task_name, args=args, queue=queue, eta=eta)

# 使用するとき
Celery = CeleryClass.get_instance()
Celery.send_task("SampleTask", args=[""], queue="user")