[Python]Upstash de Logging(Flask)
はじめに
前回の記事多くの方に読んでいただいたみたいで誠にありがとうございます！
本当にUpstashは便利なのですが、煽りタイトルでただユーザーを増やしただけみたいな感じになってしまいこのまま終わるのもどうなの？ と思うところがあるので今回実際にUpstashを使ってみた事例を紹介したいと思います。
私はActivityPubという今話題のMastodonやMisskeyで使われている分散型SNS技術に関するOSSの開発を行っており、PSHというActivityPub実装サーバーをVercel上で動かしていたりします。
上記記事を書いた頃と比べて現在は多機能になりましたが未だにデータベース機能がありません。
そもそもUpstashはこのPSHに組み込むデータベースを探している最中に見つけたサービスなんですね。
じゃあ早速UpstashにPSHにデータベースを実装しよう！ ……とはいっても数日前に使い始めたばかりで永続性のあるRedisのDBをどう実装してよいのか全くわからないですし、変な実装をして後から破壊的変更をする羽目になっても困ります。
そんなわけでまずはPSH以外にも使えるようなウェブアプリケーションの永続的なロギング機能を実装して、どんな感じでRedisやUpstashを使えばよいのか一度やっていきたいと思います。
実装
PSHはPythonで書かれており、WebアプリケーションフレームワークにFlask、HTTPリクエストクライアントにRequestsを使用しています。
またActivityPubに使われている署名用にCryptographyを使用していますが今回は使いません。
@app.before_request
def before():
requests.post(
os.getenv("UPSTASH_REDIS_REST_URL") + "/pipeline",
json=[
[
"LPUSH",
"example.com.list",
json.dumps(
{
"log": {
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"headers": dict(request.headers),
"body": request.get_data().decode(),
}
},
ensure_ascii=False,
),
]
],
headers={"Authorization": "Bearer " + os.getenv("UPSTASH_REDIS_REST_TOKEN")},
)
このように書けばUpstashにFlaskのheaderとbodyのアクセスログを保存することができます。
環境変数からUPSTASH_REDIS_REST_URLとUPSTASH_REDIS_REST_TOKENを取ってきて、Requestsを使いUpstashのREST APIへPOSTを行っています。
/pipelineというエンドポイントを使えば二次元配列のJSONデータにRedisコマンドを組み込めるので、今回はLPUSH example.com.list "..."コマンドを叩きlist型で末尾に保存する方法を採用しました。
from datetime import datetime
import json
import os
from flask import Flask, Response, request
import requests
app = Flask(__name__)
@app.route("/")
def home():
return Response("Zenn", headers={"Content-Type": "text/plain; charset=utf-8"})
@app.before_request
def before():
requests.post(
os.getenv("UPSTASH_REDIS_REST_URL") + "/pipeline",
json=[
[
"LPUSH",
"example.com.list",
json.dumps(
{
"log": {
"timestamp": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
"headers": dict(request.headers),
"body": request.get_data().decode(),
}
},
ensure_ascii=False,
),
]
],
headers={"Authorization": "Bearer " + os.getenv("UPSTASH_REDIS_REST_TOKEN")},
)
if __name__ == "__main__":
app.run()
$ pip install Flask requests gunicorn
$ export UPSTASH_REDIS_REST_URL=<UPSTASH_REDIS_REST_URL>
$ export UPSTASH_REDIS_REST_TOKEN=<UPSTASH_REDIS_REST_TOKEN>
$ gunicorn app:app -b 127.0.0.1:8080
ローカルでもこのように書けば動くと思うので試してみてください。
Vercelで動かす場合はUpstashのIntegrationsを使う方法もありますが、Vercel上に環境変数を設定しても問題なく動きます。
UPSTASH_REDIS_REST_URLとUPSTASH_REDIS_REST_TOKENを設定した後は再度Redeployする必要があります。
$ curl -H "Authorization: Bearer <UPSTASH_REDIS_REST_TOKEN>" <UPSTASH_REDIS_REST_URL>/lrange/example.com.list/0/100
{"result":["{\"log\": {\"timestamp\": \"...\", \"headers\": {...}, \"body\": \"...\"}}}"]}
$ curl -H "Authorization: Bearer <UPSTASH_REDIS_REST_TOKEN>" <UPSTASH_REDIS_REST_URL>/lrange/example.com.list/0/100 | jq -r '.result[]' | jq -s '.' > log.json
あとはLRANGE example.com.list 0 -1コマンドを叩きjqやyqを使ってうまいことやればいい感じにログを取得できます。
$ curl -H "Authorization: Bearer <UPSTASH_REDIS_REST_TOKEN>" <UPSTASH_REDIS_REST_URL>/del/example.com.list
ログが必要なくなったらDEL example.com.listコマンドを叩けば全部消えます。
ログ1行1Kbぐらいなのでシングルリージョンデータベースの上限256Mbにすぐ到達することはないと思いますが気になる人はたまにやりましょう。
おわりに
RedisとUpstashのおかげで簡単に永続的なロギング機能を実装することができました。
皆さんも様々なアイデアを形にできるRedisとUpstashを好きになりましょう。
Discussion