二年前にやったこれを今度はLINE Botを使ってみることにしました。
https://qiita.com/hamasan05/items/8b778bf0aa2398fd4b28
構成
なんでこんな事するのか？
この構成の最大のメリットは常時起動しているサーバを自分で用意しなくて良いことです。
ゲームを遊んでいるときだけGCEのサーバを起動しておくことによって
アイドル時間のCPUとメモリの料金を払わずにすみます。
クラウドの料金体系を活かして高スペックなサーバを短い時間使って安価に済ます事ができるのです。
今回Webhook＋サーバーレスのランタイムの組み合わせでこれを実現します。
BOTがWebhookではなく常時接続のサーバを要求するチャットサービスでは常時起動しているサーバが必要です。
GCEの無料インスタンスや宅内のRaspberry Piを使うなどアイディアもありますが、
やはりインフラの管理が面倒なので解放されたいです。
ということでいくつかチャットサービスを比較してみましたが、
手軽さでLINE Botを選びました。
LINE側でやること
この記事を参考にプロバイダとチャンネルの作成を行い
チャンネルシークレットとチャンネルアクセストークンを発行しておきます。
https://www.tomotaku.com/line-repeat-bot/
Cloud Funcitons側でやること
Google CloudのアカウントをサインアップしたらCloud Functionsを作成します。
トリガーをHTTPにして 未認証の呼び出しを許可 にします。
ランタイムに Python3.10を選択します
requirements.txtは下記のとおりに設定しました。
cachetools==4.2.1
certifi==2020.12.5
chardet==4.0.0
click==7.1.2
Flask==1.1.2
future==0.18.2
google-api-core==1.26.0
google-api-python-client==1.12.8
google-auth==1.27.0
google-auth-httplib2==0.0.4
googleapis-common-protos==1.53.0
httplib2==0.19.0
idna
itsdangerous==1.1.0
Jinja2==2.11.3
line-bot-sdk==1.19.0
MarkupSafe==1.1.1
oauth2client==4.1.3
packaging==20.9
protobuf==3.15.3
pyasn1==0.4.8
pyasn1-modules==0.2.8
pyparsing==2.4.7
pytz==2021.1
requests==2.25.1
rsa==4.7.2
six==1.15.0
uritemplate==3.0.1
urllib3==1.26.3
Werkzeug==1.0.1
main.pyには150行ほどのコードを書きました。
import base64
import hashlib
import hmac
import googleapiclient.discovery
from linebot import LineBotApi
from linebot.exceptions import LineBotApiError
from linebot.models import TextSendMessage
from oauth2client.client import GoogleCredentials
from config import (
CHANNEL_ACCESS_TOKEN,
CHANNEL_SECRET,
DICTIONARY,
INSTANCE_NAME,
MACHINE_TYPE_DICT,
MACHINE_TYPE_SUFFIX,
PROJECT,
USER_IDS,
ZONE,
)
class MineCraftServer:
def __init__(self, project: str, zone: str, instance_name: str):
credentials = GoogleCredentials.get_application_default()
self.service = googleapiclient.discovery.build(
"compute", "v1", credentials=credentials, cache_discovery=False
)
self.project = project
self.zone = zone
self.instance_name = instance_name
self.instance = (
self.service.instances()
.get(project=project, zone=zone, instance=instance_name)
.execute()
)
def start(self) -> bool:
if self.instance["status"] == "TERMINATED":
self.service.instances().start(
project=self.project, zone=self.zone, instance=self.instance_name
).execute()
return True
return False
def stop(self) -> bool:
if self.instance["status"] == "RUNNING":
self.service.instances().stop(
project=self.project, zone=self.zone, instance=self.instance_name
).execute()
return True
return False
def scale(self, up: bool) -> bool:
if self.instance["status"] == "TERMINATED":
after_type = (
MACHINE_TYPE_DICT["high"] if up else MACHINE_TYPE_DICT["default"]
)
if not self.instance["machineType"].endswith(after_type):
self.instance["machineType"] = MACHINE_TYPE_SUFFIX + after_type
self.service.instances().update(
project=self.project,
zone=self.zone,
instance=self.instance_name,
body=self.instance,
).execute()
return True
return False
def get_machine_type_str(self) -> str:
machine_type = self.instance["machineType"]
return machine_type[machine_type.rfind("/") + 1 :]
def is_machine_type_default(self) -> bool:
machine_type = self.instance["machineType"]
return (
machine_type[machine_type.rfind("/") + 1 :] == MACHINE_TYPE_DICT["default"]
)
class LineMineCraft:
def __init__(self, request, server: MineCraftServer):
payload = request.get_json(silent=True)
self.body = request.get_data()
self.user_id = payload["events"][0]["source"]["userId"]
self.reply_token = payload["events"][0]["replyToken"]
self.type = payload["events"][0]["type"]
self.text = payload["events"][0]["message"]["text"]
self.signature = request.headers.get("x-line-signature")
self.mine_craft_server = server
def auth(self) -> bool:
body_hash = hmac.new(
CHANNEL_SECRET.encode("utf-8"), self.body, hashlib.sha256
).digest()
signature = base64.b64encode(body_hash)
if signature.decode("utf-8") != self.signature:
print(f"invalid signature: {signature} header: {self.signature}")
self.response(DICTIONARY["NotAuthorized"])
return False
if self.user_id not in USER_IDS:
print(self.user_id)
return False
return True
def response(self, message: str) -> None:
line_bot_api = LineBotApi(CHANNEL_ACCESS_TOKEN)
try:
line_bot_api.reply_message(self.reply_token, TextSendMessage(text=message))
except LineBotApiError as e:
print(e)
def start(self) -> bool:
ret = self.mine_craft_server.start()
if ret:
self.response(
DICTIONARY["started"]
+ DICTIONARY["machineType"].format(
self.mine_craft_server.get_machine_type_str()
)
)
else:
self.response(DICTIONARY["alreadyStarted"])
return ret
def stop(self) -> bool:
ret = self.mine_craft_server.stop()
if ret:
self.response(DICTIONARY["stopped"])
if not self.mine_craft_server.is_machine_type_default():
self.scale(False)
else:
self.response(DICTIONARY["alreadyStopped!"])
return ret
def scale(self, up: bool) -> bool:
ret = self.mine_craft_server.scale(up)
if ret:
self.response(DICTIONARY["changed"])
else:
self.response(DICTIONARY["alreadyStarted!"])
return ret
def line(request):
line_minecraft = LineMineCraft(
request, MineCraftServer(PROJECT, ZONE, INSTANCE_NAME)
)
if not line_minecraft.auth():
return
if line_minecraft.type == "message":
if line_minecraft.text == DICTIONARY["start"]:
line_minecraft.start()
elif line_minecraft.text == DICTIONARY["stop"]:
line_minecraft.stop()
elif line_minecraft.text == DICTIONARY["scaleUp"]:
line_minecraft.scale(True)
elif line_minecraft.text == DICTIONARY["scaleDown"]:
line_minecraft.scale(False)
else:
line_minecraft.response(DICTIONARY["notSupportedFunction"])
main.pyには150行ほどのコードを書きました。
# GCP設定
PROJECT = "GCPのプロジェクトのID"
ZONE = "対象のGCEのゾーン"
INSTANCE_NAME = "対象のGCEのインスタンス名"
MACHINE_TYPE_SUFFIX = f"zones/{ZONE}/machineTypes/"
MACHINE_TYPE_DICT = {"default": "e2-highcpu-2", "high": "e2-highcpu-4"}
# LINE設定
USER_IDS = ["利用するLINEのID(動かしてみてログから判断します)"]
CHANNEL_SECRET = "LINE Developerで取得したチャンネルシークレット"
CHANNEL_ACCESS_TOKEN = "LINE Developerで取得したチャンネルアクセストークン"
DICTIONARY: dict = {
"started": "起動しました",
"stopped": "停止しました",
"changed": "変更しました",
"alreadyStarted": "すでに起動しています",
"alreadyStopped": "すでに停止しています",
"notSupportedFunction": "その操作はサポートしていません",
"notAuthorized": "権限がありません",
"start": "起動",
"stop": "停止",
"scaleUp": "スペックアップ",
"scaleDown": "スペックダウン",
"machineType": "現在のマシンタイプ{}",
}
設定ファイルとしてconfig.pyを用意しました。
Cloud Functionsのお作法としては認証情報は環境変数に設定するのが良いみたいですね
このあたりはいずれ環境変数に移そうと思います。
- PROJECT
- ZONE
- USER_IDS
- INSTANCE_NAME
- CHANNEL_SECRET
- CHANNEL_ACCESS_TOKEN
ソースコードはGithubにも上げておきました
操作ができるユーザーについて
LINE BotはBOTアカウントを他人に知られてしまうと誰でもサーバの操作ができてしまうので
操作をできるユーザーを絞っておきます。
送信元の署名を確認したあとに送信元のユーザーIDを抽出します。
自分のLINE IDの取得方法がわからなかったので(おそらくBOTごとで異なるIDが振られる)
一旦は未知のIDがあったらログに出力するようにします。
自分でアクセスしてログに出力されたIDを見てconfig.pyに書き込んでアップロードをしました。
※このあたりを環境変数にするとコード書き換えの手間が省けると実感しています。
とりあえず中締め
改めて見返してみるとLineBotApiがほとんどのことをやってくれるので非常に少ないコードでサーバの起動停止ができました
やっていることは単純に下記だけですね。
- payloadの中から必要な情報を取り出して認証する
- メッセージの命令を解釈する
- サーバの操作をする
LINEがスマートスピーカーと比べて良いなと思ったのがレスポンスが遅くても良いということです。
スマートスピーカーだとレスポンスは短い必要がありますが、テキストチャットだと人間がそこまでの即応性を求めないということなのでしょう
これは実装する上で結構適当に気軽に作れるという印象を受けました。
また実機のテストも手元のスマホやPCから可能なのでテストはかなりスムーズでした。
とても手軽にできたということで今回はマイクラで遊んでいてスペックが足りないなと感じることもあって
状況に応じてサーバを強化するということにチャレンジしてみました。
サーバ強化
マイクラで遊んでいるとスペックが足りないときいうのは案外シチュエーションが限られています。
自分が遊んでいて感じたのはおおよそ下記の3つです。
- 初めて訪れてチャンクの生成が発生するとき
- モブが大量に発生しているとき(村人がたくさんいるなど)
- 複数人数で遊んでいてチャンクの読み込みが広範囲に渡るとき
こういうときはオンラインでサーバ強化しなくても
今からサーバ落として強化するわ、で問題ないと思っています。
ということでLINEでサーバ強化と元に戻すを実装してみました。
(すでに上で紹介したコードに含まれています)
やっていることは停止中のGCEのマシンタイプを好みのものに変更するだけです。
machineTypeパラメーターに指定するのは zones/{ZONE}/machineTypes/e2-highcpu-2 のような形式になるのでそこだけ注意すれば update() するだけです。
本当に簡単にマシンタイプが変更できてしまいました。
必要な時間だけ必要なスペックのマシンを借りることができる
というクラウドの長所を最大限に生かしたマイクラサーバの運用が
LINE Botを使ってChatOpsで実現することができました。