Script to monitor users logged in on minecraft

minecraft-user-monitor.service

[Unit]
Description=Minecraft User Monitor
After=network.target

[Service]
Type=simple
WorkingDirectory=/var/tmp/minecraft_user_monitor
ExecStart=/var/tmp/minecraft_user_monitor/run.sh
ExecStop=/bin/kill -TERM 
Restart=always

[Install]
WantedBy=multi-user.target

minecraft.service

[Unit]
Description=Minecraft Server
After=minecraft-user-monitor.service

[Service]
Type=simple
EnvironmentFile=/var/tmp/minecraft/env
WorkingDirectory=/var/tmp/minecraft
ExecStart=/var/tmp/minecraft/bedrock_server
ExecStop=/bin/kill -TERM ${MAINPID}
Restart=always

[Install]
WantedBy=multi-user.target

minecraft_user_monitor.py

import re
import select
import subprocess
import time
from systemd import journal
from datetime import datetime, timedelta
from threading import Thread, Event
from typing import Dict

SYSTEMD_UNIT = 'minecraft.service'


class NoUserWatchDog(Thread):
  def __init__(self, termination: Event, max_diff: timedelta) -> None:
    super().__init__()
    self.__termination = termination
    self.__max_diff = max_diff

  def run(self) -> None:
    self.__prev = datetime.now()
    while datetime.now() - self.__prev < self.__max_diff:
      time.sleep(5)
    self.__termination.set()

  def clear(self) -> None:
    self.__prev = datetime.now()


class ConnectedUserMonitor(Thread):
  def __init__(self, poll_interval_ms: float = 500.0, watchdog_timeout_m: float = 15.0) -> None:
    super().__init__()
    self.__termination = Event()
    self.__poll_interval_ms = poll_interval_ms
    self.__wd = NoUserWatchDog(self.__termination, timedelta(minutes=watchdog_timeout_m))
    self.__users: Dict[str, bool] = {}

  def run(self) -> None:
    # Filter log and move to the end of the journal
    j = journal.Reader()
    j.seek_tail()
    j.add_match(_SYSTEMD_UNIT=SYSTEMD_UNIT)

    # Register the journal's file descriptor with the polling object.
    p = select.poll()
    p.register(j, j.get_events())

    self.__wd.start()

    # Poll for new journal entries
    while not self.__termination.is_set():
      if any(list(self.__users.values())):
        self.__wd.clear()

      if p.poll(self.__poll_interval_ms):
        if j.process() != journal.APPEND:
          continue

      # Player connected: hoge, xuid: 2535425559984397
      # Player disconnected: hoge, xuid: 2535425559984397

      messages = [entry['MESSAGE'] for entry in j if 'MESSAGE' in entry and len(entry['MESSAGE'])]
      for m in messages:
        connected = re.match(r'^.* connected: ([\w\d]+), .*$', m)
        disconnected = re.match(r'^.* disconnected: ([\w\d]+), .*$', m)

        if connected:
          name = connected.groups()[0]
          self.__users[name] = True
          print(f'{name} is connected')
        elif disconnected:
          name = disconnected.groups()[0]
          self.__users[name] = False
          print(f'{name} is disconnected')

    self.__wd.join()


if __name__ == '__main__':
  from argparse import ArgumentParser

  parser = ArgumentParser(description='Monitoring users logged in Minecraft and shutdown server if not user')
  parser.add_argument(
    '-t', '--timeout',
    action='store',
    type=float,
    default=15.0,
    help='Timeout value to shutdown server if no user (default: 15 minutes)')
  args = parser.parse_args()

  counter = ConnectedUserMonitor(watchdog_timeout_m=args.timeout)

  try:
    counter.start()
    counter.join()
    subprocess.run(['/usr/sbin/shutdown', '-h', 'now'])
  except KeyboardInterrupt:
    pass

run.sh

#!/usr/bin/env bash
set -eu
/usr/bin/python3 ./minecraft_user_monitor.py --timeout 15

これはなに？

Java版Minecraftサーバを構築したインスタンス上でログインユーザ数を監視し、ユーザ数0をトリガーに自動shutdownするPythonスクリプトとsystemd設定ファイル

簡易的な設定手順

以下をEC2インスタンステンプレートのuser dataに設定しておくと便利。

1. /var/tmp/minecraft にUbuntu向け統合版Minecraftサーバを展開
2. 上記run.shとminecraft_user_monitor.pyを/var/tmp/minecraft_user_monitor/run.shに配置
3. 上記*.serviceファイルを/etc/systemd/system下に配置
4. それぞれpermission調整後にsudo systemctl enable ...等実行して有効化