アーカイブ

このサイト(www.teenytinybot.dev)の構成に関して

詳細

最近更新しておりませんでした、IMAXおじさんです。 この頃本サイトに色々機能を足していたので、そろそろ全体をまとめた構成図を作ろうと思い立ちました。

1. コンテナの構成について

本サイトはさくらVPS上にデプロイしています。 当初はUbuntu Server上のネイティブで動かしていたのですが、お仕事でDockerをよく使うこともあり途中からDockerでのデプロイに切り替えました。

各コンポーネントの基本的な構成は以下となります。

  • OS:Ubuntu 18.04
  • 仮想化:Docker
  • リバースプロキシ:Nginx 1.21.6
  • アプリバックエンド:Python3, Django 3.0
  • DB:MySQL 5.7
  • 可視化ツール:Grafana 8.3.4
  • ログ監視:Splunk 8.2.5
  • ログ転送エージェント:UniversalForwarder(@Nginxコンテナ)

また、コンテナ間の関係とデータフローは以下の通りです。

2. Djangoで提供するアプリに関して

また、Djangoでは下記のアプリを作成して公開しています。

  • 質問箱

    • Djangoで初めて作ったWEBアプリです。
    • 中の人に匿名で質問が送れます。line-bot-sdkと連携しており、質問がくるとBotからLINEで中の人に通知が来ます。
    • (最近質問が来てなくてなんか寂しいので質問募集中です)
  • 自宅の温湿度監視

    • DjangoでRESTAPIを実装し、自宅のM5StackからAPI経由で気温・湿度・気圧をサーバに上げています。
    • Grafanaで監視用ダッシュボードを作成し、公開しています。
  • 技術ブログ

    • 技術系の記事を書いていくブログページです。
    • Markdown形式が使えるよう、Djangoプラグインであるmdeditorを入れています。
    • また、コードはシンタックスハイライトが行われるよう、aceというライブラリを入れています。
    • ↓こんな感じ
  • ログ監視

    • また、自明に公開はしていませんが、Nginxのログを見ると公開サーバへの攻撃と思しきアクセスがかなり多いので、監視ツールとしてSplunkを入れています。
    • 法人利用は有償ですが、個人利用であれば制限付きで一定期間は無償利用可能なようです。

3. 今後の予定

今後は以下のことに取り組みたいな〜と考えています。

  • AWSへのデプロイ(AWSできるようになりたい...)
  • elasticsearchとsplunkの比較
  • 機械学習がバックエンドで動くWEBアプリ
  • Vue.js+Django RESTAPIで動く構成に(モダンなWEBの構成にしたい...)

自作競プロジャッジサーバの構成について

詳細

簡単な自作競プロジャッジサーバーを作った話をします。(競プロとは?:こちらを参照) デプロイしてから結構経つのに、詳細な構成に関して何も記事を書いていなかったので、ここらで文章に落とそうと思います。 サーバは一時期まで公開していましたが、Dockerで実行環境を隔離しているとはいえ任意のコードが実行できてしまうのは怖いので最近は公開していません。

1. ジャッジサーバの構成に関して

ジャッジサーバ内のコンポーネント間の構成とデータフローを示します。 図1.1.

回答提出から正誤判定までのジャッジサーバ内での処理の流れ

  1. 回答者からコードが提出されると、一旦Djangoがそれを受け取ります。
  2. DjangoはこれをRedis+Celeryで実装された非同期処理ジャッジキューに突っ込みます。その後、ユーザに対しジャッジ処理中である旨を示します。
  3. ジャッジキューは順に提出コードの正誤判定を行ない、判定終了したコードに対し完了フラグを立てます。
  4. ジャッジ完了後、各コードの正誤判定の詳細をユーザに返します。

2. 難しかったところ

非同期処理の実装

コードを提出してから、ジャッジが完了するまでにある程度時間がかかります。 実行に時間がかかる場合やテストケースが多い場合は、POSTしてからページが表示されるまで何も表示されない状態になってしまいます。 また、実行時間があまりに長い場合はブラウザにリクエストタイムアウトと判定されてしまいま す。 そのため、今回は下記のようなフローで非同期処理を実装することにしました。

バックエンドではジャッジがcelery+redisで実装されたFIFOキューを保持し、提出されたコードに順次正誤判定を行なっています。 ジャッジ未完了の状態ではブラウザ更新を促し、ジャッジ完了の状態では結果を通知します。

TLEの検知

競技プログラミングにおいては、コードの実行時間が非常に重要です。多くの場合、2秒程度の実行時間内に問題を解けるコードを書く必要があります。 制限時間内に解けなかったコードにはTLE(Time Limit Exceeded)判定が与えられ、不正解扱いとなります。

最初は簡単なジャッジサーバができれば良い程度に考えていたので、コードの正解・不正解しか判定しないロジックを書いていました。 ところが、段々と完成度を追求したくなってくるもので、やはりTLEの判定がしたくなってきます。

色々実行時間を計測する術を検討しました。例えば、下記のようにコード実行前後で時刻を計測し、その差分を測るようなやり方です。

start = time.time()
ret = subprocess.run(cmd, shell=True, ...) #提出コードを実行
end = time.time()

timeTaken = end - start # 実行時間を計測

しかし、このやり方ではwhile True: passのような無限ループが発生した場合にTLE判定を出せなくなってしまいます。 一定時間以上実行している場合には途中で実行を打ち切るような動作が必要になるわけです。

最終的にはコードを実行しているsubprocessモジュールで実行時間を計測し、オーバーした場合にTLE判定を出すようにしました。 subprocessモジュールには、実行時間を制限するオプションtimeoutがあるため、ここに実行制限時間の秒数を渡せば上記の問題に対応しつつTLEを判定できます。

# subprocessモジュールによるコードの実行
try:
    ret = subprocess.run(
        cmd, timeout=2, shell=True, stdin=fin, #  timeoutに制限時間を指定
        stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )
except (CalledProcessError, TimeoutExpired) as e:
    if TimeoutExpired:
        tle += 1  # 実行時間が超過した場合はTLE判定を出す
        continue

    elif CalledProcessError:
        ...

3 今後の予定

  • フロントエンドをVue.jsで作り直す。
  • ajaxによる非同期通信を実装し、ユーザがブラウザ更新をする必要をなくす
  • 対応言語を増やす(現在Java, C++, Python3, PyPy3に対応)
  • etc....

ソートアルゴリズムを真面目に実装する(応用情報技術者試験)

詳細

タイトルの通り、ソートアルゴリズムを真面目に実装します。 言語はPython、実装するソートアルゴリズムはクイックソート・ヒープソート・マージソートです。

ソートは基本的に標準ライブラリのsort関数に投げがちなので、内部がどうなっているかを気にすることがないかと思います。 今回は応用情報技術者試験の対策も兼ねて、真面目にソートを実装します。 なお、めんどくさいので昇順ソートしか実装しません()

参考文献

クイックソート

クイックソートは、分割統治の考え方でソートを行うアルゴリズムです。

下記のような配列を考えたとき、その配列に対してpivotと呼ばれる軸を設定し、そのpivot未満の配列とpivot以上の配列に分割していきます。 これを再帰的に繰り返すことで、整列済みの配列を得るアルゴリズムです。また、安定なソートではありません。

計算量

平均計算量は、O(NlogN)となります。 具体的な計算量の導出はこちらが参考になりました。

ざっくり言えば、適切にpivotを選んだ場合上記の画像で示す通りpivotによる分割と結合はそれぞれ回数H = logNで抑えることができます。 それぞれの高さに関して、N回比較が発生するので、計算量はO(NlogN)であるとわかります。

最悪計算量

ただし、予めソートされたデータに対して最大値/最小値をpivotとして取り続けた場合は最悪計算量がO(N^2)となります。

実装

A = list(map(int, input().split()))
l = 0
r = len(A) - 1

def quick_sort(tgt):
    tgt_len = len(tgt)
    if len(tgt) <= 1:
        return tgt

    pivot = tgt[0]
    left = []
    right = []

    for i in range(1, tgt_len):
        if tgt[i] <= pivot:
            left.append(tgt[i])
        else:
            right.append(tgt[i])

    return quick_sort(left) + [pivot] + quick_sort(right)

res = quick_sort(A)
print(res)

ヒープソート

ヒープソートはヒープの再構成がO(logN)でできることを利用したソートアルゴリズムです。 ヒープの根が常に配列の最小値となっているので、ヒープの根をpop→ヒープを再構築...を繰り返し、popした順に配列を作ればO(NlogN)でソートできます。

ヒープに突っ込むため、安定なソートではないですが、常に計算量がO(NlogN)となります。

実装

ヒープを自力で実装し、ヒープからどんどんpopしていく方針を取ります。

class Heap:
    def __init__(self, arr:list):
        self.arr = []

        for item in arr:
            self.push(item)

    def push(self, x):
        self.arr.append(x)
        child = len(self.arr) - 1

        while child > 0:
            parent = (child - 1)//2

            if self.arr[parent] < x:
                break
            else:
                self.arr[child] = self.arr[parent]
                child = parent

        self.arr[child] = x

    def root(self):
        return self.arr[0] if self.arr else None


    def pop(self):
        if not self.arr:
            return None

        root = self.arr[0] 
        self.arr[0] = self.arr[-1]
        self.arr.pop()

        parent = 0
        smallest_index = parent
        heap_size = len(self.arr)

        while 2*parent+1 < heap_size:
            l = 2*parent+1  # 左の子
            r = l+1         # 右の子

            # 自分、左の子、右の子の中で、最小のものを選び取る
            if (r < heap_size) and (self.arr[parent] >= self.arr[r]):
                smallest_index = r

            if (l < heap_size) and (self.arr[smallest_index] > self.arr[l]):
                smallest_index = l

            # 自分が最小になった時点で、ヒープ化されたのでbreak
            if smallest_index == parent:
                break

            # 親と子を入れ替える
            self.arr[parent], self.arr[smallest_index] = self.arr[smallest_index], self.arr[parent]
            parent = smallest_index

        return root

def heap_sort(arr):
    heap = Heap(arr)
    N = len(arr)

    print(heap.arr)

    result = []
    for _ in range(N):
        result.append(heap.pop())
        print(heap.arr)

    return result

if __name__ == "__main__":
    sample = [1, 1, 2, -1, 100, 92]

    result = heap_sort(sample)

    print(result)  # -> [-1, 1, 1, 2, 92, 100]

マージソート

マージソートも、分割統治法の考え方を用いてソートを行うアルゴリズムです。 下記画像のように、まず要素が一つになるまで分割を繰り返します。

その後、分割した配列を2つずつマージしていきます。 このとき、分割した配列を結合する際に、昇順になるように入れていきます。各配列はソート済みであるので、それぞれ先頭のindexから順に比較していけば良いことになります。

計算量

こちらの説明がとてもわかりやすいです。

上の図で示した通り、分割とマージのプロセスはそれぞれH = logNで抑えられます。 マージの際には、それぞれの分割のレイヤーでN回比較が走るので、結果的に計算量はO(2NlogN) = O(NlogN)となります。

実装

再帰的な実装を行います。(実は分割のところがO(N)かかっているのではないかという説があります)

def merge_sort(data):

    if len(data) <= 1: return data

    mid = len(data) // 2

    # 再帰的に分割していく
    left = merge_sort(data[:mid:])
    right = merge_sort(data[mid::])

    return merge(left, right)


def merge(left, right):
    result = []
    i, j = 0, 0

    # 配列の先頭が最小値になっているので、毎回比べながら追加していく
    while (i < len(left)) and (j < len(right)):
        if left[i] <= right[j]:
            result.append(left[i])
            i += 1
        else:
            result.append(right[j])
            j += 1

    # add the remainging part
    if i < len(left):
        result.extend(left[i:])

    if j < len(right):
        result.extend(right[j:])

    return result

if __name__ == "__main__":
    array = list(map(int, input().split()))

    res = merge_sort(array)

    print(res)

HerokuにLINE webhookをデプロイする(heroku/line-bot-sdk/python3)

詳細

HerokuにLINE Webhookをデプロイするまで

今回は無料で使えるホスティングサービスであるHerokuにLINE Webhookをデプロイするまでの手順を備忘録的に記します。

作りたいもの

実家でデジタル難民と化している祖母が他の家族からのLINEメッセージを見れるように、↓のようなシステムを作ろうと思い立ちました。

このシステムを作るにあたり、LINEメッセージを送ると自作のAPIにデータをPOSTするようなWebhookをつくりたいです。 今回はゴールとして、まずはLINE bot(Python)に送ったメッセージをそのままLINEで返信してくるようなエコーボットをデプロイすることにします。

本記事の構成

  1. 前提(heroku/line developerのアカウント作成等)
  2. Webhookの構成
  3. line-bot-sdkについて
  4. herokuへのデプロイ手順

1. 前提

1.1. herokuのアカウント作成

herokuの利用にはアカウント作成が必要です。今回は無料枠の範疇のことしかしないので、Freeアカウントで作成します。 こちらからアカウントを作成することができます。

また、heroku CLIを使うためインストールが必要です。こちらの手順に従い、インストールしてください。

1.2. LINE Developerアカウントの作成

今回のwebhook作成に関しては、LINEbotが必要となります。LINEbotの作成にはLINE developerアカウントが必要となるため、アカウント登録をおこないます。 こちらからアカウント作成をしてください。

2. webhookの構成

今回のwebhookを作成するにあたり、ローカル上で下記のようなフォルダ構成を作成します。

*/
|__main.py              # webhookの処理を記述する
|__requirements.txt  # 使用するライブラリの一覧
|__runtime.txt          # herokuで使用する言語の諸元
|__Procfile              # herokuで実行するコマンド
|__.env                   # herokuで使用する環境変数

2.1. main.py

main.pyにはline-bot-sdkを使って行う処理を記述することになります。 詳細は3節で説明するため、ここでは説明を省きます。

2.2. requirements.txt

requirements.txtには今回使用するライブラリを記述します。これをプロジェクトフォルダに含めることで、デプロイ時に自動でライブラリをインストールしてくれます。

Flask==2.0.3
line-bot-sdk==2.2.1
requests==2.27.1

2.3. runtime.txt

runtime.txtにはherokuで使用するPythonのバージョンを記述します。その際、herokuが対応しているバージョンを選択しなければならないため、現在のherokuが何に対応しているかを確認するようにします。 今回は以下の内容を記述します。一行のみです。

python-3.9.0

2.4. Procfile

Procfileはheroku上でデプロイ時に実行するコマンドを記述します。今回はmain.pyを実行することでサーバーを立ち上げたいので、下記の内容とします。

web: python3 main.py

2.5. .env

.envファイルはherokuの環境に登録したい環境変数を記述します。main.pyなどに直接記載したくない情報(例えばline-bot-sdkのシークレットキーなど)を記述し、herokuの環境に反映します。

3. line-bot-sdkについて

今回は開発者公式が公開しているline-bot-sdkのサンプルプログラムを使用します。サイトはこちら(GitHub)から見れます。

元のサンプルプログラムはアクセストークンとシークレットキーをベタ書きしているため、この二つだけは環境変数から読み込むように変更を加えています。

from flask import Flask, request, abort
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, TextMessage, TextSendMessage
import os

# Flaskでwebhookを立ち上げ
app = Flask(__name__)

# アクセストークンとシークレットキーは環境変数から読み込む
YOUR_CHANNEL_ACCESS_TOKEN = os.environ["YOUR_CHANNEL_ACCESS_TOKEN"]
YOUR_CHANNEL_SECRET = os.environ["YOUR_CHANNEL_SECRET"]

line_bot_api = LineBotApi(YOUR_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(YOUR_CHANNEL_SECRET)

@app.route("/callback", methods=['POST'])
def callback():
    signature = request.headers['X-Line-Signature']

    body = request.get_data(as_text=True)
    app.logger.info("Request body: " + body)

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        print("invalid signatrue. please check channel info")
        abort(400)

    return 'OK'

@handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
    line_bot_api.reply_message(event.reply_token, TextSendMessage(text=event.message.text))


if __name__ == "__main__":
    port = int(os.getenv("PORT"))
    app.run(host="0.0.0.0", port=port)

4. herokuへのデプロイに関して

ここまでのプログラムの準備ができたところで、herokuへのデプロイを行います。大まかな手順は下記の通りです。

  1. heroku上でのプロジェクト作成
  2. gitリポジトリの作成・登録
  3. heroku環境変数の反映
  4. herokuへのpush

4.1. heroku上でのプロジェクト作成

まずは、herokuの個人コンソールから「create new app」を選択し、新しいアプリケーションを作成します。

名前を選ぶように促されるので、適当な名前をつけます。「hogehoge.herokuapp.com」というURLを与えられますが、世界で一意な名前である必要があるため、適度に長い名前にしてください。

4.2. gitリポジトリの作成・登録

ローカルのプロジェクトをgitリポジトリとして登録し、herokuにpushします。

ローカルで

git init
git add *
git commit -m "first commit"

とし、まずはgitのローカルリポジトリを作成します。

次に、herokuのリモートリポを紐付けます。heroku CLIから以下を実行します。

heroku git:remote -a {自分のアプリの名前}

このままgit pushすると、環境変数を設定しないままビルドが始まってしまうので、git pushする前に.envファイルで設定した環境変数をherokuに反映します。

4.3. 環境変数の反映

heroku CLIを用いて、.envファイル内に記述した環境変数をherokuのアプリに反映します。 プロジェクトディレクトリ内でheroku config:pushを実施します。成功すると「Successfully wrote settings to Heroku!」とログが出ます。

venv) hogehoge:~/Desktop/line-webhook-app$ heroku config:push
Successfully wrote settings to Heroku!

4.4. herokuへのpush

ここまで実施できたら、git push heroku masterでheroku上のリモートリポジトリにプロジェクトをpushします。 pushが成功すると自動でビルド→デプロイが行われます。

LINE botにメッセージを送って、全く同じ内容の返事が来れば成功です。

プロジェクト内でheroku logs --tailとすると、デプロイしたアプリのログを確認することができます。 うまくLINE botが返事を返してくれなければ、このログを見てトラブルシュートを行います。

まとめ

これまでVPSをメインで使っていたので、herokuのようなPaaSを使えば無料で簡単にデプロイできるんだなと感心しました。

【資格】ネットワークスペシャリスト試験に合格しました!

詳細

合格発表から時間が空いてしまいました。2023年4月にネットワークスペシャリスト試験を受験し、合格していました。

点数は午前2:8割、午後1:6割、午後2:8割といった具合で、落ちるなら午前1かなぁと思っていたところなんとか耐えていました。

次はセキスペとAWS SAAを受けようと思います。

About

IMAXおじさんが(主に)技術系記事を備忘録として残していくブログです.

Category

  1. Tech
  2. Daily
  3. Job
  4. Other