💓

discord.py の tasks.loop で無料で簡易的に死活監視してみる

に公開

はじめに

discord.py には tasks.loop という機能が組み込まれていて、これを使うとファイル1つ追加するだけで簡易的な死活監視ができます。今回はこの方法を試してみました。

完成イメージ

10分ごとに Discord チャンネルへこんな Embed が届くようになります:

Heartbeat — Healthy
Uptime: 0h 10m 2s | Latency: 22ms | Guilds: 1
Boot: 2026-01-27 21:44 JST

レイテンシに応じて色が変わります:

  • 緑: 200ms 未満(正常)
  • 黄: 200〜500ms(低下)
  • 赤: 500ms 以上(異常)

ログにも出るので、デプロイ先のログビューアでも確認できます。

INFO:src.cogs.health:[Heartbeat] Healthy | uptime=0h 10m 2s latency=22ms guilds=1

環境

  • Python 3.12+
  • discord.py 2.6+

やること

今回やることはざっくり以下の3つだけです。

  1. health.py を書く
  2. ボットに登録する
  3. 送信先チャンネルIDを .env に書く

1. health.py を書く

src/cogs/health.py を新規作成します。全体のコードはこんな感じです。

"""Health monitoring cog for sending periodic heartbeat embeds."""

# Python 3.10+ のアノテーション記法を使うためのインポート
from __future__ import annotations

# 標準ライブラリ
import logging
import time
from datetime import datetime, timedelta, timezone

# discord.py
import discord
from discord.ext import commands, tasks

# 自分のプロジェクトの設定(後述)
from src.config import settings

# このファイル用のロガーを取得
logger = logging.getLogger(__name__)

# ハートビートの間隔(分)
_HEARTBEAT_MINUTES = 10

# 日本時間のタイムゾーン(UTC+9)
# 必要に応じて変更してください
_JST = timezone(timedelta(hours=9))


class HealthCog(commands.Cog):
    """定期的にハートビートを送信する Cog"""

    def __init__(self, bot: commands.Bot) -> None:
        # Bot インスタンスを保存
        self.bot = bot

        # 起動時刻を記録(Uptime 計算用)
        # time.monotonic() はシステム起動からの経過秒数を返す
        # datetime と違って時刻変更の影響を受けない
        self._start_time = time.monotonic()

        # 起動時刻を日本時間で保存(Embed のフッター表示用)
        self._boot_jst = datetime.now(_JST)

    async def cog_load(self) -> None:
        """Cog が読み込まれたときに呼ばれる"""
        # ハートビートのループを開始
        self._heartbeat.start()

    async def cog_unload(self) -> None:
        """Cog がアンロードされたときに呼ばれる"""
        # ハートビートのループを停止
        self._heartbeat.cancel()

    @tasks.loop(minutes=_HEARTBEAT_MINUTES)
    async def _heartbeat(self) -> None:
        """定期的に実行されるハートビート処理"""

        # === Uptime の計算 ===
        # 現在時刻 - 起動時刻 = 経過秒数
        uptime_sec = int(time.monotonic() - self._start_time)
        # divmod で商と余りを同時に取得
        # 例: 3661秒 → 1時間, 余り61秒 → 1分, 余り1秒
        hours, remainder = divmod(uptime_sec, 3600)
        minutes, seconds = divmod(remainder, 60)
        # 表示用の文字列に整形
        uptime_str = f"{hours}h {minutes}m {seconds}s"

        # === Bot の状態を取得 ===
        # Bot が参加しているサーバー(ギルド)の数
        guild_count = len(self.bot.guilds)

        # Discord との接続遅延を取得
        # bot.latency は秒単位なのでミリ秒に変換
        latency_ms = round(self.bot.latency * 1000)

        # === レイテンシに応じてステータスを判定 ===
        if latency_ms < 200:
            status = "Healthy"      # 正常
        elif latency_ms < 500:
            status = "Degraded"     # やや遅い
        else:
            status = "Unhealthy"    # 異常

        # === ログに出力 ===
        logger.info(
            "[Heartbeat] %s | uptime=%s latency=%dms guilds=%d",
            status, uptime_str, latency_ms, guild_count,
        )

        # === Discord チャンネルに送信 ===
        # チャンネルIDが設定されている場合のみ送信
        if settings.health_channel_id:
            # チャンネルIDからチャンネルオブジェクトを取得
            channel = self.bot.get_channel(settings.health_channel_id)

            # チャンネルが存在し、テキストチャンネルであることを確認
            if channel is not None and isinstance(channel, discord.TextChannel):
                # Embed を作成
                embed = self._build_embed(
                    status=status,
                    uptime_str=uptime_str,
                    latency_ms=latency_ms,
                    guild_count=guild_count,
                )
                # 送信
                await channel.send(embed=embed)

    @_heartbeat.before_loop
    async def _before_heartbeat(self) -> None:
        """ループ開始前に1回だけ呼ばれる"""
        # Bot が Discord に接続完了するまで待機
        # これがないと、接続前にハートビートを送ろうとしてエラーになる
        await self.bot.wait_until_ready()

    def _build_embed(
        self,
        *,
        status: str,
        uptime_str: str,
        latency_ms: int,
        guild_count: int,
    ) -> discord.Embed:
        """ハートビート用の Embed を作成する"""

        # === レイテンシに応じて色を決定 ===
        if latency_ms < 200:
            color = discord.Color.green()   # 緑
        elif latency_ms < 500:
            color = discord.Color.gold()    # 黄
        else:
            color = discord.Color.red()     # 赤

        # === Embed を作成 ===
        embed = discord.Embed(
            title=f"Heartbeat — {status}",
            color=color,
            # timestamp は UTC で指定する(Discord が自動でユーザーの
            # タイムゾーンに変換して表示してくれる)
            timestamp=datetime.now(timezone.utc),
        )

        # === フィールドを追加 ===
        # inline=True で横並びに表示
        embed.add_field(name="Uptime", value=uptime_str, inline=True)
        embed.add_field(name="Latency", value=f"{latency_ms}ms", inline=True)
        embed.add_field(name="Guilds", value=str(guild_count), inline=True)

        # === フッターに起動時刻を表示 ===
        embed.set_footer(text=f"Boot: {self._boot_jst:%Y-%m-%d %H:%M JST}")

        return embed


async def setup(bot: commands.Bot) -> None:
    """
    Bot にこの Cog を登録する
    load_extension("src.cogs.health") で呼ばれる
    """
    await bot.add_cog(HealthCog(bot))

こんな感じで動きます。順番に見ていきます。

tasks.loop でループを作る

@tasks.loop(minutes=_HEARTBEAT_MINUTES)
async def _heartbeat(self) -> None:
    ...

@tasks.loop(minutes=10) を付けるだけで、10分ごとに自動実行されるメソッドになります。secondshours も指定できます。

cog_load / cog_unload でライフサイクル管理

async def cog_load(self) -> None:
    self._heartbeat.start()

async def cog_unload(self) -> None:
    self._heartbeat.cancel()

__init__ でタスクを開始するのは避けた方がよさそうです。__init__ は同期メソッドなので、ボットがまだ Discord に接続していない状態でタスクが走ってしまいます。cog_load は非同期メソッドなので、こちらで開始するのが安全かと思います。

before_loop で接続完了を待つ

@_heartbeat.before_loop
async def _before_heartbeat(self) -> None:
    await self.bot.wait_until_ready()

before_loop はループ開始前に1回だけ呼ばれる準備用のフックです。wait_until_ready() で Discord への接続完了を待ちます。

tasks.loopbefore_loop 完了後に即座に初回実行される仕様のようなので、起動直後にハートビートが送られます。before_loop 内でタスク本体を手動で呼ぶと二重実行になるので、そこだけ注意が必要です。

bot.latency で応答性をチェック

latency_ms = round(self.bot.latency * 1000)

bot.latency は Discord Gateway との WebSocket Heartbeat 往復遅延を秒単位で返します。ミリ秒に変換して、200ms / 500ms を閾値にステータスを判定しています。

2. ボットに登録する

setup_hook に1行追加するだけです。

async def setup_hook(self) -> None:
    await self.load_extension("src.cogs.health")

3. 送信先チャンネルIDを設定する

.env に追加します:

HEALTH_CHANNEL_ID=123456789

設定クラス側(pydantic-settings を使っている場合):

class Settings(BaseSettings):
    discord_token: str
    health_channel_id: int = 0  # 0 なら Discord への送信をスキップ

health_channel_id = 0 のままでもログには出力されるので、Discord に送らなくてもOKです。

これで完成

ボットを起動すると、すぐにハートビートが送られます。あとは10分ごとに自動で繰り返されます。

追加のパッケージも不要、外部サービスへの登録も不要。discord.py の標準機能だけで動くのは手軽でいいですね。

もう少しやるなら

エラーハンドリング

@_heartbeat.error
async def _heartbeat_error(self, error: Exception) -> None:
    logger.exception("Heartbeat failed: %s", error)

tasks.loop はデフォルトでエラーが起きるとタスクを停止するようです。error ハンドラを追加すると、エラーをログに記録しつつタスクを継続できます。

実運用では、一時的なネットワークエラーで channel.send が失敗することがあります。エラーハンドラがないとループ自体が止まってしまうので、本番環境では追加しておいた方がよさそうです。

インターバルを動的に変更

self._heartbeat.change_interval(minutes=5)

「死んだ」をどう判断するか

死活監視で一番大事なのは、Bot が死んだことにどうやって気づくかです。

この方式は Bot 自身がハートビートを送る仕組み(Push 型)なので、Bot がフリーズ・クラッシュすると通知も一緒に止まります。つまり:

  • 通知が来ている → 生きている
  • 通知が来ない → 死んでいる かもしれない(ただし人間が気づく必要がある)

10分間隔で送っているので、「20分以上 Embed が来ていない」なら異常と判断できます。しかし、これは Discord チャンネルを目視で確認しないとわかりません。

一方、UptimeRobot のような外部サービス(Pull 型)は「生きてる?」と外から問い合わせる方式です。Bot が応答しなければ即座にアラートが飛ぶため、より確実に異常を検知できます。

自動で検知したい場合は、外部からの監視が必要です。次のセクションで紹介します。

もっと本格的にやるなら

tasks.loop はあくまで簡易的な方法です。本格的な監視サービスを使えば、グラフ付きの通知やフリーズ検知など、より高度なことができます。もっと良いやり方があるかもしれませんが、参考程度に比較表を載せておきます。

サービス 無料枠 グラフ通知 フリーズ検知 導入コスト
tasks.loop(本記事) - × × ファイル1つ
UptimeRobot 50モニター × HTTP エンドポイント追加
Datadog 5ホスト / 1日保持 △(有料) △(有料) エージェント導入
Mackerel 無料プランあり Slack/Discordにグラフ画像付き通知 エージェント導入

UptimeRobot で無料で自動検知する

Bot に aiohttp で簡易 HTTP サーバーを追加し、UptimeRobot から定期的に叩く方法です。

from aiohttp import web

async def health_handler(request: web.Request) -> web.Response:
    return web.json_response({"status": "ok"})

# bot の setup_hook 等で起動
app = web.Application()
app.router.add_get("/health", health_handler)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", int(os.environ.get("PORT", 8080)))
await site.start()

UptimeRobot(無料プラン)で https://your-app.example.com/health を5分間隔で監視し、応答がなければメール通知を受け取れます。Bot がフリーズしても HTTP サーバーごと止まるので、外部から検知できます。

Datadog でやってみる

Datadog Agent をインストールすれば、コード変更なしにログやメトリクスを収集できます。多くの PaaS では Buildpack や Add-on として簡単に導入できます。

本記事の [Heartbeat] ログを Datadog に流せば、Logs Monitor で「一定時間ログが出なければアラート」という監視ができます。ただし、Logs Monitor やグラフ付き通知は有料プランが必要です。

...と思ったけど Mackerel の方が楽かも

Discord Bot の監視なら、Mackerel の方が相性が良いかもしれません。

Mackerel は Slack にグラフ画像のスナップショット付きで通知を飛ばせます。そして Discord は Slack 互換の Webhook に対応しているので、Webhook URL の末尾に /slack を付ければ Discord チャンネルにも同じ通知が届きます。

# Discord の Slack 互換 Webhook URL
https://discord.com/api/webhooks/xxxxx/yyyyyy/slack

Bot の監視通知を Discord に統一できるのは地味に便利です。グラフ画像付きで「CPU 使用率が跳ねた」「メモリが逼迫している」といった状況が一目でわかります。

まずは本記事の tasks.loop で始めて、運用が安定してきたら外部サービスを検討するのが良いのではないでしょうか。

まとめ

  • ファイル1つ + 設定1行で死活監視が動く
  • tasks.loop は discord.py 標準機能。追加パッケージ不要
  • before_loop + wait_until_ready() で接続完了を待ってから即時実行
  • まずはこれで始めて、必要に応じて外部監視を追加するのが良さそう

参考

Discussion