ピアボーナス(社内用通貨)システム作ってみた

本記事は GCP(Google Cloud Platform) Advent Calendar 2022 1日目の記事です。
2日目: GCSにIP制限を掛ける方法について @koduki さん

普段はUnityエンジニアとして働いているcovaです。
今回は普段の業務時間の一部を使って「社内用のピアボーナスシステム」を構築してみたので
その経緯や制作過程について紹介いたします。

ピアボーナスとは

Peer Bonusとは「社員同士」でお互いの仕事・作業内容などに対して誉める+ボーナスを送り合うというものです。
人事施策の一環として一部企業では採用されてたりしていて、そのボーナスを用いてお菓子や飲み物の雑貨購入を始め、一部購入補助にボーナスを充てたりするため「社内用の仮想通貨」といっても差し支えないと思います。

ピアボーナスを作った経緯

弊社には「Head For Ambition」「Blend Tech in Life」「Brave Decision」という3つの「Value」を設定しています。
しかしながら、このような会社の考え方は社員全体に中々浸透しづらいという問題がありました。
そこで各バリューを指針としたピアボーナス用トークンを社員全員に毎月配布して、そのバリュー指針にあった良いことをしたらトークンを送り合うようにすれば、自然とバリューが浸透しやすく、またお互いの良いところを褒める文化の醸成につながるのでは？と思い作った形です。

製作まわり

設計その1

用件定義

ピアボーナスシステムで必要な機能としては

• 定期的にピアボーナス用トークン（以下トークン）を社員に付与
• トークンの管理
• トークンを特定の人に送る
• トークンの送付履歴の確認
• トークンの受信履歴の確認
• 購入用トークン（以下、ゴールド）の消費
• トークンの消費履歴の確認

これであれば「トークン」および「ゴールド」の履歴と値を社員（ユーザー）毎に保持する仕組みであれば十分そうです。

より良いUXを目指して

基本的にユーザーは社員で、必ずしも技術者だけではありません。
そのため「誰でも」「簡単に」利用できる環境を作る必要があります。

そこで、今回は入出力を業務でも利用している「Slack」をトリガーに、
また、履歴に関しては経理も絡んでくるので一部情報をSpreadSheet に吐き出せるようにすることで社内の利便性を高めようと考えました。

構成

たまたま社内にはビルド用のJenkins マシンがあったため、そこにDocker で環境構築してしまおうと考えました。

幸い、Slack のAPI操作用のSDK(BOLT: https://github.com/slackapi/bolt-python) が Python で作られていること、また、社内的に「Unityエンジニア」と「サーバーエンジニア」の共通言語として Python を読み書きできるようになろうという流れが出来ていたため、「エンジニアメンバーなら誰でもメンテ対応を可能にする」ためにも Python を採用しました。

Dockerの内容としてはロジック等を担当する「Pythonコンテナ」とデータ保存用の「MySQLコンテナ」を用意したシンプルなものです。

トラブル発生

Jenkins マシンの再起動問題

どうせ「社内システム」なので一度立ち上げたらしばらく立ち上げっぱなしでOKだろうと高を括ってました。
しかし「UnityビルドするにあたりJenkins環境更新したのでマシンを再起動したい」と他のメンバーから言われ再起動しました。

その結果、Docker上に保存していたDBのデータを外部ファイルに吐き出していなかったため、DBの中身が全て吹き飛びました。

完全にコピペコードで動かしていた弊害です。（ちゃんと勉強しような、1敗）

Slack のUXの問題

このときSlackのBOLT を利用していましたが、BOTにメンションしてコマンドを入力する方式でした。

勤怠管理のジョブカン のようにスラッシュコマンドでやりたいよねっていうメンバーからのフィードバックを受けました。

しかし、このスラッシュコマンド、エンドポイントのURLが必要です。

さらに、ビルドマシンを公開するのはセキュリティ的に問題なのと、Docker上でエンドポイント用のAPPサーバー的なのを構築するのは、自分のスキル的にコストが高いと考えました。

設計その2

ということで、先ほどの課題を解決するにあたり、エンジニアメンバー全体に相談を投げかけたところ「クラウド環境移行しかないよね」と言う結論に至りました。
そこで移行にあたり

• Python で書いたコードはなるべく流用したい
• スタートアップなので費用はなるべく安く抑えたい (by 経理担当)
• Endpoint発行の手間が限りなく小さい

を重視して色々調べたところ、GCPの契約がスタートアップ用で一定利用範囲内であれば無料枠が通常よりも多い状態でしたのでGCPの利用を検討しました。

GCP 移行時の構成

ちょうどGoogleCloudFunction (以下GCF)がPython にも対応していたこと、GoogleCloudSQL (以下GCSql) と連携すれば丸っとクラウド上で構築できそうと思い、以下のような環境構成を考えました。

これなら「業務時間の10~20%」を使えば実装できそうな目処がたちました。

クラウド移行中のトラブル

CloudFunction とCloudSQLが接続できない

Python にはMySQLを扱うためのFrameworkがいくつか存在します。
中でも簡単に操作可能なPyMySQLを最初検討していましたが、接続部分でなぜかうまくいかきませんでした。

ここは公式のコード通り SQLAlchemy を採用したところ無事動いたので、Pythonﾁｮｯﾄﾜｶﾙ という状態でなければ、おとなしく公式ドキュメントに乗っかるのが吉です。

しかし、Webページ上のサンプルを使うとうまくつながらないことがわかりました。

    pool = sqlalchemy.create_engine(
        # Equivalent URL:
        # mysql+pymysql://<db_user>:<db_pass>@/<db_name>?unix_socket=<socket_path>/<cloud_sql_instance_name>
        sqlalchemy.engine.url.URL.create(
            drivername="mysql+pymysql",
            username=db_user,
            password=db_pass,
            database=db_name,
            query={"unix_socket": unix_socket_path},
        ),
        # ...
    )

ここの コメントアウトで省略されている最後の部分が問題です。
そこで公式を見ると、Githubのリンクがあるので確認すると、Timeout等のパラメータを設定してあります。

よって、Githubのコードを丸っと持ってきて設定してあげれば接続ができました。

Slack のBOTメンションとSlashコマンドでEndpointの作りが違う

BOT のエンドポイントは

botEndpoint.py

def bot_mention(request):
 ...

となっており、Slashコマンドは

slashCommandEndpoint.py

def main(request):
 ...

となっていますが、それぞれ引数のデータ構造が異なります。
普段 C#, C++, などの静的型付け言語を書いている身としては、データ構造把握するために型を書いてくれ・・・と思ってしまいます。

(なお、実際の運用コードではのちに判明した型情報を用いて :型 を明記して書いてます）

Bot メンション

このrequest のデータ構造ですが、「う〜ん、わからん」と思っていたところ、全く同じようなことをやっている方がいらっっしゃいました。

Validationもしっかり描かれているコードなので、素直に採用して問題ありませんでした。

json をパースした結果の変数で request_json.get("event") と呼んであげることで今回の投稿に必要な情報は大体格納されているので、あとは辞書型データからパラメータを引っ張ってくればOKです。

SlashCommand

SlashCommandの引数を扱うためにはPython用Webフレームワークの　Flask を用います。
実際に、引数の型は flask.Request となっています。
このflask.Requestの form プロパティ にデータが辞書型で格納されているのでそれを利用してあげればOKです。

Slashコマンドのパラメータについては Slack公式の記事の途中のコメント が一番参考になりました。

パラメータ数が多いのでこれらの情報を元に簡易的なパーサーを作って各種パラメータにアクセスしやすいとメンテが楽になるのでおすすめです

SlashCommandParser.py

class Parser:

    def __init__(self, token: str,
                 team_id: str, team_domain: str,
                 slack_channel_id: str, slack_channel_name: str,
                 slack_user_id: str, slack_user_name: str,
                 command: str,
                 text: str,
                 response_url: str,
                 trigger_id: str
                 ):
        self.__token = token
        self.__team_id = team_id
        self.__team_domain = team_domain
        self.__slack_channel_id = slack_channel_id
        self.__slack_channel_name = slack_channel_name
        self.__slack_user_id = slack_user_id
        self.__slack_user_name = slack_user_name
        self.__command = command
        self.__text = text
        self.__response_url = response_url
        self.__trigger_id = trigger_id

    @classmethod
    def parse(cls, form):
        # Text のmention が<@slack_user_id> ではなく @slack_user_name になっているため
        # SlackBoltと同様のレスポンスに整形が必要
        text_message = "No Text"
        if "text" in form:
            tmp_msg = str(form["text"]).split()
            text_message = ""
            count = 0
            for msg in tmp_msg:
                if count > 0:
                    text_message += " "
                if msg[0] == "@":
                    text_message += "<@"+ str(form["user_id"]) + ">"
                else:
                    text_message += msg
                count += 1

        return cls(
            str(form["token"]) if "token" in form else "No Token",
            str(form["team_id"]) if "team_id" in form else "No Team ID",
            str(form["team_domain"]) if "team_domain" in form else "No Team Domain",
            str(form["channel_id"]) if "channel_id" in form else "No Slack Channel ID",
            str(form["channel_name"]) if "channel_name" in form else "No SlackChannelName",
            str(form["user_id"]) if "user_id" in form else "No UserId",
            str(form["user_name"]) if "user_name" in form else "No UserName",
            str(form["command"]) if "command" in form else "No Command",
            text_message,
            str(form["response_url"]) if "response_url" in form else "No URL",
            str(form["trigger_id"]) if "trigger_id" in form else "No TriggerID",
        )

    @property
    def token(self) -> str:
        return self.__token

    @property
    def team_id(self) -> str:
        return self.__team_id

    @property
    def team_domain(self) -> str:
        return self.__team_domain

    @property
    def slack_channel_id(self) -> str:
        return self.__slack_channel_id

    @property
    def slack_channel_name(self) -> str:
        return self.__slack_channel_name

    @property
    def slack_user_id(self) -> str:
        return self.__slack_user_id

    @property
    def slack_user_name(self) -> str:
        return self.__slack_user_name

    @property
    def command(self) -> str:
        return self.__command

    @property
    def text(self) -> str:
        return self.__text

    @property
    def response_url(self) -> str:
        return self.__response_url

    @property
    def trigger_id(self) -> str:
        return self.__trigger_id

    def dump(self) -> str:
        dump_text = "[dump]\n"
        dump_text += "token:"+self.token+"\n"
        dump_text += "team_domain:"+self.team_domain+"\n"
        dump_text += "slack_channel_id:"+self.slack_channel_id+"\n"
        dump_text += "slack_channel_name:"+self.slack_channel_name+"\n"
        dump_text += "slack_user_id:"+self.slack_user_id+"\n"
        dump_text += "slack_user_name:"+self.slack_user_name+"\n"
        dump_text += "command:"+self.command+"\n"
        dump_text += "text:"+self.text+"\n"
        dump_text += "response_url:"+self.response_url+"\n"
        dump_text += "trigger_id:"+self.trigger_id+"\n"

        return dump_text

Slack 側のレスポンスでタイムアウトが起きる

やはりDB操作は処理がだんだん時間がかかってきてしまい、数秒レスポンスにかかってしまうと処理途中にも関わらずSlack でBOTがEphemeralなメッセージ(※1)でTimeout と通知してきます。

こちらの問題に関しては別途記事かしてありますので、是非とも参考にしていただければと思います。

※1 Ephemeralメッセージ: 「短命の」という意味のメッセージで実行者のみに表示されるメッセージのことです

CloudFunction のページからコード等のリソースが表示されない

諦めです。
この警告が出るあたりから、コードのデプロイはShellScript を作ってCLI上で実行していました。

deploy.sh

# For SlashCommand
gcloud functions deploy --region=asia-northeast1 GCFの関数名 --max-instances=1 --runtime python39 --trigger-http --allow-unauthenticated

# PubSub
gcloud functions deploy --region=asia-northeast1 GCFの関数名 --max-instances=1 --runtime python39 --trigger-topic=設定したTopic名

なお、HTTPトリガーとPub/Subトリガーでオプションの設定方法が違うのでご注意ください(2敗)

処理が多重実行される

CloudFunction のMax Instance の数が問題でした。 1 にしておくことで複数台での平行処理をブロックできるので多重実行されるという思わぬトラブルを防止できます。(Deploy 用のShell にオプション引数がないと上書きされるので正しく設定しましょう)

しばらく運用してみて

一旦Slack コマンドとGCFとGCSqlがつながったので社内運用してみました。
その結果実装として以下に問題がありました

1. 定期的にピアボーナス用トークンを社員に付与
   • 手動じゃやってられない
2. トークンの消費履歴の確認
   • 経理担当がゴールドの消費と経費申請をリンクさせるための情報を出力したい
   • 経理担当がGCSqlでSQLコマンドを叩くのはハードルが高すぎる
   • みんな大好きExcel で出力してほしい
     • 妥協でSpreadSheet
3. サーバー費用削減

トークンを毎月Reset＆固定量付与

毎月送付可能なトークンをResetするために、手動でSQLを実行しましたが、社員が増えたり等今後を考えるとあまり賢い選択ではありません。
そこでGoogle Cloud Sheduler を導入して Pub/Sub を可能にすることで自動的に実行する仕組みを追加しました。

Pub/Sub 対応は結構簡単でcronの知識があると設定が簡単です。

実際にScheduler や専用のCloudFunction の設定をするにあたり、以下の記事が参考になりました。

GoogleSpreadsheet 連携

購入履歴をGoogleSpreadSheet にも出力する対応ですが、 こちらの通りに実行することでSpreadSheet の内容の読み書きをPythonで行えるようになります。

なお、CloudFunction上で実行するには pip install の代わりにrequirements.txt　に

gspread
oauth2client

を追記すれば使えるようになりました。

費用削減

最初はスタートアップ用の無料枠で行なっていたのですが、運用途中で無料期間が終了したため運用コストが発生しました。
運用費用を見ると、CloudScheduler / CloudFunction は無料枠の範囲で収まっていましたが、GCSqlのみ費用が発生しました。

そこで社員が使うであろうタイミングでのみSql Instanceを起動、その後停止する様に変更をかけました。
具体的には平日の9:00~23:59 の間は稼働で、平日の深夜〜早朝帯と土日はInstanceを停止する様にしました。

これでおおよそ30%ほど費用を削減できます。

やり方としては下記の記事と、その参考リンク先を読めば簡単に対応が可能です(1~2h程度)

具体的なコードとしてはこのような感じのコードを用意してあげれば十分です。
(各種パラメータはScheduler とGCF双方で定義しておいてます)

endpoint.py

import base64
import core
import json
from pprint import pprint

# SQLインスタンス制御APIのエンドポイント
def control_sql_instance(event, context) -> None:
    """Triggered from a message on a Cloud Pub/Sub topic.
    Args:
         event (dict): Event payload.
         context (google.cloud.functions.Context): Metadata for the event.
    """
    # {"instance": "hogehoge", "status": "hoge" } のようなpayloadが送られてくるはず
    try:
        pubsub_message = base64.b64decode(event['data']).decode('utf-8')
        print(pubsub_message)
        information = json.loads(pubsub_message)
        instance_name = information["instance"]
        command_name = information["status"]

        is_succeeded = core.execute(instance_name, command_name)
    except Exception as e:
        pprint(e.with_traceback())

core.py

import os
import logging
import googleapiclient.discovery
from googleapiclient.errors import HttpError
from oauth2client.client import GoogleCredentials


# @brief インスタンス操作処理
# @param instance_name - Target Instance name
# @param command_name - CommandName
# @return Is Succeeded ? or Not
def execute(instance_name: str, command_name: str) -> bool:

    # Validation
    if instance_name is None or len(instance_name) < 1:
        print("Invalid Instance Name")
        return False

    if command_name is None or len(command_name) < 1:
        print("Invalid CommandName")
        return False

    credentials = GoogleCredentials.get_application_default()
    if credentials is None:
        logging.error("Invalid Environment. credentials is None")
        return False

    sqladmin = googleapiclient.discovery.build('sqladmin', 'v1beta4', credentials=credentials)
    instance = instance_name
    project = os.environ['SQL_PROJECT_ID']

    if command_name == "start":
        print("Awake Instance")
        try:
            body = {'settings': {'activationPolicy': 'ALWAYS', }}  # インスタンスの起動
            request = sqladmin.instances().patch(project=project, instance=instance, body=body)
            response = request.execute()
            print(response)
        except HttpError as err:
            logging.error("Failed Awake Instance: {}".format(err))
            return False
        else:
            logging.info("Awake instance is succeeded : {}".format(response))
            return True

    elif command_name == "stop":
        print("Halt Instance")
        try:
            body = {'settings': {'activationPolicy': 'NEVER', }}  # インスタンスの停止
            request = sqladmin.instances().patch(project=project, instance=instance, body=body)
            response = request.execute()
            print(response)
        except HttpError as err:
            logging.error("Failed halt instance: {}".format(err))
            return False
        else:
            logging.info("Halt request is succeeded".format(response))
            return True
    else:
        print("Unexpected Command:"+command_name)
        return False

最終的な構成

構成図

フォルダ構成

.
├── core.py # main.py から呼ばれる実処理を記述するクラス
├── deployShells
│   ├── deploy_main.sh
│   ├── deploy_sandbox.sh
│   └── deploy_token_reset.sh
├── google-api-key.json # Git ignore 対象です
├── graffity_util
│   ├── db
│   │   ├── DB操作系Script.py 
│   │   ├── テーブル操作系Script.py 
│   │   ...
│   ├── gcp
│   │   └── pub_sub操作系Script.py
│   ├── graffity_googledrive
│   │   └── google_drive_api.py
│   ├── graffity_slack
│   │   ├── request_parser.py
│   │   ├── slack_api_util.py
│   │   └── slack_post_util.py
│   └── token_command.py # Enum定義やパラメータをとりまとめたクラス定義用
├── main.py # 各種エンドポイントまとめ(ロジックが肥大化するのが見えていたのでファイルを分割)
├── requirements.txt # CloudFunction でpip install の代わりの依存パッケージ定義ファイル
└── venv # PyCharm 用のファイル群

requirement.txt

requirement.txt

# SQL/JSON
Flask
SQLAlchemy

# Slack
slack_sdk
slack_bolt
certifi

# GAS
gspread
oauth2client

# google
google-cloud-pubsub
google-api-python-client

システムを構築してみて

業務では新卒1年目の頃にPHPで運用スマホゲームのバックエンドを書いた程度（当時はまだLaravel とかなかった時代）の知見で、いきなりPython触ってシステム構築できるか見通しがなかった状態でしたが、Web記事が充実していることもあり結構思ったより簡単に実装できたなという印象です。

項目	工数
ビルドマシン上にDockerで構築	5営業日程度.
クラウド環境移行	10営業日程度.

で実装が終わりました。

参考