プライベートでDockerコンテナで動作するPalworldゲームサーバーを監視・制御するためのDiscord Botを作ったことがあるので紹介します。

主な機能は以下の通り

プレイヤーの監視と非アクティブ時の自動停止
Discordメッセージによるサーバーの手動起動・停止
サーバーのステータス確認（稼働状況、リソース使用率）
起動・停止時のDiscord通知

サーバー

スペック
- メモリ:4GB
- CPU:仮想4コア
- 容量:100GB
- OS:Ubuntu 22.04

言語はPythonです

※ Discordのボット機能の解説については今回は触れないでおきます。（いろんな解説記事が世に転がってますし）

1. ボットの初期設定と依存関係
2. イベントハンドラとユーティリティ関数
3. 監視・自動停止機能
- 3.1 定期的なログ監視タスク
- 3.2 サーバー停止処理
4. Discord コマンド機能
5. ボットの実行
6. 動作フロー概要
まとめ

1. ボットの初期設定と依存関係

1.1 モジュールのインポート

import subprocess
import time
import discord
from discord.ext import commands, tasks
import asyncio
import json

subprocess: シェルコマンド（Docker コマンドなど）を実行するためのモジュール
time: タイムスタンプ管理と時間計測
discord: Discord API とのインターフェース
commands, tasks: Discord Bot のコマンド機能と定期実行タスク
asyncio: 非同期処理のためのモジュール
json: JSON データの解析

1.2 Discord Bot の設定

# Discordボットの設定
intents = discord.Intents.default()
intents.messages = True
intents.message_content = True
intents.guilds = True
bot = commands.Bot(command_prefix="palworld ", intents=intents)

# Discordで通知するチャンネルID
CHANNEL_ID = 0000

Intents の設定: Discord API v9以降で必要となる権限設定
- messages: メッセージ関連のイベントを受け取る権限
- message_content: メッセージの内容を読み取る権限
- guilds: サーバー（ギルド）関連の情報にアクセスする権限
コマンドプレフィックス: palworld （スペース含む）をコマンドの接頭辞として設定
CHANNEL_ID: 通知を送信する Discord チャンネルのID（実際の ID に変更が必要）

経緯) Discodeもコマンドが使用できますが、普段ゲームをする人たち大半がCUIに馴染みがないので、まあコマンド入力が浸透しない…。ということでメッセージから拾うカタチで実装。

1.3 Docker 関連の設定

# コンテナ名
container_name = "palworld-server"

# docker-compose.ymlの場所
docker_compose_path = "/root/palworld/docker-compose.yml"

container_name: Docker 上で動作する Palworld サーバーのコンテナ名
docker_compose_path: サーバーを起動するための docker-compose.yml ファイルの場所

1.4 グローバル変数の初期化

# プレイヤーが最近ログインしたかどうかを追跡
player_logged_in_recently = False

# 停止予告メッセージを送信したかどうかのフラグ
notified_for_shutdown = False

# プレイヤーがいなくなってからの時刻
zero_player_since = None

# グローバル変数に起動時刻を追加
startup_time = None

# グローバル変数に監視状態を追加
monitoring_active = False

player_logged_in_recently: プレイヤーがログインしているかを追跡
notified_for_shutdown: 停止予告メッセージを既に送信したかのフラグ
zero_player_since: プレイヤーがいなくなった時刻を記録
startup_time: サーバー起動時刻を記録（起動直後の監視を制御するため）
monitoring_active: 監視タスクが動作中かを示すフラグ

2. イベントハンドラとユーティリティ関数

2.1 Bot の準備完了イベント

# ボットが準備完了した後に呼び出されるイベント
@bot.event
async def on_ready():
    print(f'Bot has logged in as {bot.user}')
    # 起動時には監視を開始しない

on_ready(): Bot が Discord に接続し準備完了したときに呼び出される
起動時には自動的に監視を開始せず、明示的なコマンドで開始する設計
- クーロンで常時立ち上げるのでbotの立ち上げはシェルに寄せています

2.2 プレイヤー情報取得関数

# プレイヤー情報を取得する関数を追加
def get_player_info():
    try:
        result = subprocess.run(
            ["docker", "exec", "-i", container_name, "rest-cli", "players"],
            capture_output=True,
            text=True
        )
        if result.returncode == 0:
            return json.loads(result.stdout)
        return None
    except Exception as e:
        print(f"プレイヤー情報の取得中にエラーが発生: {e}")
        return None

Docker コンテナ内の rest-cli players コマンドを実行
コンテナ内の REST API を使用してプレイヤー一覧を JSON 形式で取得
エラー処理を含み、失敗時は None を返す

2.3 サーバー情報取得関数

# サーバー情報を取得する関数を追加
def get_server_info():
    try:
        result = subprocess.run(
            ["docker", "exec", "-i", container_name, "rest-cli", "info"],
            capture_output=True,
            text=True
        )
        if result.returncode == 0:
            return json.loads(result.stdout)
        return None
    except Exception as e:
        print(f"サーバー情報の取得中にエラーが発生: {e}")
        return None

Docker コンテナ内の rest-cli info コマンドを実行
コンテナ内の REST API を使用してサーバー情報を JSON 形式で取得
サーバー名やバージョンなどの基本情報を含む

2.4 Discord 通知関数

# Discordに通知を送る関数
async def send_discord_notification(message):
    # Discordメッセージの送信
    channel = bot.get_channel(CHANNEL_ID)
    if channel:
        await channel.send(message)

指定したチャンネル ID に Discord メッセージを送信
監視タスクからの通知やエラーメッセージの送信に使用

2.5 フラグリセット関数

# 停止後のフラグをリセットする関数
def reset_shutdown_flag():
    global notified_for_shutdown, player_logged_in_recently, zero_player_since
    notified_for_shutdown = False  # 停止予告をリセット
    player_logged_in_recently = False  # プレイヤーがログインしているかどうかもリセット
    zero_player_since = None  # プレイヤーがいるためリセット

サーバー停止時などに呼び出され、各種フラグを初期状態にリセット
global キーワードでグローバル変数を参照・変更

3. 監視・自動停止機能

3.1 定期的なログ監視タスク

# 定期的にログをチェックするタスク
@tasks.loop(seconds=10)  # 10秒ごとにログをチェック
async def check_logs():
    global player_logged_in_recently, notified_for_shutdown, zero_player_since, startup_time

    # 起動直後1分間は監視をスキップ
    if startup_time and time.time() - startup_time < 60:
        print("サーバー起動直後のため、監視をスキップします")
        return

    if startup_time and time.time() - startup_time >= 60:
        startup_time = None

    # コンテナの状態確認
    try:
        container_status = subprocess.check_output(
            ["docker", "ps", "-q", "-f", f"name={container_name}"]
        ).decode("utf-8").strip()
    except subprocess.CalledProcessError as e:
        print(f"コンテナ状態の確認中にエラーが発生しました: {e}")
        return

    if not container_status:
        print("コンテナが停止しています。フラグをリセットします。")
        reset_shutdown_flag()
        return

    # プレイヤー情報の取得
    player_info = get_player_info()
    if player_info is None:
        return

    player_count = len(player_info.get("players", []))
    print(f"現在のプレイヤー数: {player_count}")

    if player_count > 0:
        if not player_logged_in_recently:
            print("プレイヤーがログインしました。停止予告をリセットします。")
        player_logged_in_recently = True
        notified_for_shutdown = False
        zero_player_since = None
    else:
        if player_logged_in_recently:
            print("プレイヤーがログアウトしました。警告メッセージを送信します。")
            player_logged_in_recently = False
            if not notified_for_shutdown:
                await send_discord_notification("⚠️【システム】\n現在、プレイヤーがいないため、1分後にサーバーを停止します。")
                notified_for_shutdown = True
            if zero_player_since is None:
                zero_player_since = time.time()
        else:
            if zero_player_since:
                elapsed_time = time.time() - zero_player_since
                print(f"プレイヤーがいなくなってからの経過時間: {elapsed_time} 秒")
                if elapsed_time > 60:
                    print("1分経過しました。サーバーを停止します。")
                    await send_discord_notification("🛑【システム】\nサーバーは非アクティブのため停止されました。再起動の際は手動で実行してください。")
                    await stop_server()

@tasks.loop(seconds=10): 10秒ごとに実行される定期タスク
起動直後の監視スキップ: サーバー起動直後の1分間は監視をスキップ（起動処理中の誤停止防止）
コンテナ状態確認: Docker メッセージでコンテナが稼働中か確認
プレイヤー数の監視:
- プレイヤーがいる場合: 停止予告をリセット
- プレイヤーがいなくなった場合: 停止予告を送信し、タイマーを開始
- プレイヤーが1分間いない場合: サーバーを停止

経緯) 立ち上げている状態が続くとゲーム内時間が進んでイベントなど発生してしまうので、基本的に誰もやっていないときは落とすようにしました。落とすの忘れたときの防止。

3.2 サーバー停止処理

# サーバーを手動で停止
async def stop_server():
    global monitoring_active
    try:
        # docker compose stop を実行してサーバーを停止
        result = subprocess.run(["docker", "stop", container_name], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

        # 監視を停止
        if monitoring_active:
            check_logs.stop()
            monitoring_active = False

        # 停止後にフラグをリセット
        reset_shutdown_flag()

    except subprocess.CalledProcessError as e:
        await send_discord_notification(f"🛑【エラー】\nサーバー停止中にエラーが発生しました: {e.stderr}")

Docker の stop コマンドでコンテナを停止
監視タスクも停止
停止後にフラグをリセット
エラー時は Discord に通知

4. Discord コマンド機能

4.1 サーバー起動コマンド

# サーバーを手動で起動
@bot.command()
async def up(ctx):
    global startup_time, monitoring_active
    try:
        # 起動開始メッセージを送信
        await ctx.send("🔄【システム】\nPalworldゲームサーバーを起動中です...")

        # コンテナの存在を確認する関数
        def check_container_exists(container_name):
            result = subprocess.run(
                ["docker", "ps", "-a", "--filter", f"name={container_name}", "--format", "{{.Names}}"],
                check=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            return container_name in result.stdout

        # コンテナの存在を確認
        if check_container_exists(container_name):
            await ctx.send("【システム】\n既存のサーバーコンテナを起動します...")
            # 既存のコンテナを起動
            result = subprocess.run(
                ["docker", "start", container_name],
                check=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            if result.returncode != 0:
                await ctx.send(f"🛑【エラー】\nサーバー起動中にエラーが発生しました: {result.stderr}")
        else:
            await ctx.send("🟢【システム】\n新しいサーバーコンテナを作成します...")
            # 新しいコンテナを作成して起動
            result = subprocess.run(
                ["docker", "compose", "-f", docker_compose_path, "up", "-d", "--no-recreate"],
                check=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            if result.returncode != 0:
                await ctx.send(f"🛑【エラー】\nサーバー起動中にエラーが発生しました: {result.stderr}")

        # 起動時刻を記録
        startup_time = time.time()

        # 監視を開始
        if not monitoring_active:
            check_logs.start()
            monitoring_active = True

        # 正常終了時のメッセージ
        await ctx.send("🟢【システム】\nPalworldゲームサーバーの起動を開始しました。\n起動完了までしばらくお待ちください...")

        # 起動完了を監視するタスクを開始
        await asyncio.sleep(10)  # 10秒待機
        check_startup_complete.start(ctx.channel)

    except subprocess.CalledProcessError as e:
        await ctx.send(f"🛑【エラー】\nサーバー起動中にエラーが発生しました: {e.stderr}")

@bot.command(): palworld up メッセージを定義
コンテナ存在確認: 既存コンテナがあるか確認
- 既存コンテナがある場合: docker start で起動
- 存在しない場合: docker compose up で新規作成
監視開始: 起動時刻を記録し、監視タスクを開始
起動完了監視: check_startup_complete タスクを開始

4.2 起動完了監視タスク

# 起動完了を監視するタスク
@tasks.loop(seconds=5)
async def check_startup_complete(channel):
    server_info = get_server_info()
    if server_info and "version" in server_info:
        await channel.send(f"✅【システム】\nPalworldゲームサーバーの起動が完了しました！\nサーバー名: {server_info.get('servername', 'Unknown')}\nバージョン: {server_info.get('version', 'Unknown')}")
        check_startup_complete.stop()

@tasks.loop(seconds=5): 5秒ごとに実行されるタスク
サーバー情報を取得し、バージョン情報が返ってくれば起動完了と判断
起動完了時にサーバー名とバージョンを含む通知を送信
タスク自身を停止

経緯) 立ち上げ処理を投げてからすぐアクセスを試みる→できない！という人がだいたいなので、わかりやすく起動確認も処理に追加

4.3 サーバー停止コマンド

# サーバーを手動で停止
@bot.command()
async def down(ctx):
    try:
        # サーバー停止タスクを実行
        await stop_server()

        # 正常終了時のメッセージ
        await ctx.send("🛑【システム】\nPalworldゲームサーバーを停止しました！")
    except subprocess.CalledProcessError as e:
        await ctx.send(f"🛑【エラー】\nサーバー停止中にエラーが発生しました: {e.stderr}")

@bot.command(): palworld down メッセージを定義
stop_server() 関数を呼び出してサーバーを停止
結果を Discord チャンネルに通知

4.4 ステータス確認コマンド

# "status"コマンドでコンテナの稼働状況とリソース使用率を確認
@bot.command()
async def status(ctx):
    try:
        # docker ps -q -f name=container_name を実行してコンテナが稼働中か確認
        ps_result = subprocess.run(
            ["docker", "ps", "-q", "-f", f"name={container_name}"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )

        if ps_result.returncode == 0 and ps_result.stdout.strip():
            # コンテナが稼働中の場合、CPUとメモリの使用率を取得
            stats_result = subprocess.run(
                ["docker", "stats", container_name, "--no-stream", "--format", "{{.CPUPerc}} {{.MemUsage}}"],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )

            if stats_result.returncode == 0:
                cpu_usage, mem_usage = stats_result.stdout.strip().split(' ', 1)
                status_message = (
                    "🟢【システム】\n"
                    "Palworldゲームサーバーは現在稼働中です。\n\n"
                    f"📊 **リソース使用率:**\n"
                    f"CPU使用率: {cpu_usage}\n"
                    f"メモリ使用率: {mem_usage}"
                )
            else:
                status_message = (
                    "🟢【システム】\n"
                    "Palworldゲームサーバーは現在稼働中です。\n"
                    "ただし、リソース使用率の取得中にエラーが発生しました。"
                )
        else:
            status_message = "🛑【システム】\nPalworldゲームサーバーは現在停止しています。"

        await ctx.send(status_message)
    except subprocess.CalledProcessError as e:
        await ctx.send(f"🛑【エラー】\nステータス取得中にエラーが発生しました: {e.stderr}")

@bot.command(): palworld status メッセージを定義
docker ps コマンドでコンテナの稼働状態を確認
稼働中の場合は docker stats コマンドでリソース使用率を取得
CPU使用率とメモリ使用率を含む詳細なステータスを表示

5. ボットの実行

# ボットを実行
bot.run('xxxxx')

Discord Bot トークンを指定して Bot を起動
実運用時には 'xxxxx' を実際の Discord Bot トークンに置き換える必要がある

6. 動作フロー概要

初期化:
- Bot の起動と Discord への接続
- 自動的な起動は開始しない
手動起動:
- palworld up メッセージで起動
- コンテナを起動または新規作成
- 監視タスクを開始
- 起動完了を監視
監視サイクル:
- 10秒ごとにプレイヤー数を確認
- プレイヤーがログアウトすると1分後の停止を予告
- 1分間プレイヤーがいないとサーバーを停止
手動停止:
- palworld down メッセージでサーバーを停止
- 監視タスクも停止
ステータス確認:
- palworld status メッセージでサーバーの状態とリソース使用率を確認

まとめ

bot自体つくるのはそんな大変でもなかったですが、いかに「システムに馴染みがない人が簡単に扱えるか」を考えて処理を組まないといけないという、使用者目線を改めて意識するいい機会になりました。なので、botというより「チャットボット」のような…受け答えの動き的にはそれが近いかもしれません。

あと、パルが可愛かったです。

ELW株式会社テックブログ

リアルなログをそのままお届けします。

Discordでゲームのマルチサーバを制御するbotを作ってみた