ログ管理

監査ログ（ユーザー向け）

スキーマ

ユーザー向け監査ログのスキーマについてはshiron log schemaを参照してください。

保存方法

Cloudfalre D1 上での保存
容量削減のために 1 ヶ月で削除(無料枠は 5GB, 容量を見つつ再検討)
※将来的に課金プランによって保管期間を上げるのはあり

Python での実装例

# Hub_idに紐づくログの取得
import os
import requests

def get_audit_logs_by_hub(hub_id: str, limit: int | None = 5) -> list[HubAudit] | None:
    cf_client_id = os.getenv("CF_ACCESS_CLIENT_ID", None)
    cf_client_secret = os.getenv("CF_ACCESS_CLIENT_SECRET", None)
    if cf_client_id is None or cf_client_secret is None:
        return None
    url = f"https://audit.shiron-system.net/hub/{hub_id}?limit={limit}&before=0"
    headers = {
        "CF-Access-Client-Id": cf_client_id,
        "CF-Access-Client-Secret": cf_client_secret,
        "Content-Type": "application/json",
    }
    res = requests.get(url, headers=headers, timeout=5)
    if res.status_code != 200:
        return None
    return [HubAudit(**line, created_at=line["createdAt"], updated_at=line["updatedAt"]) for line in res.json()]

# Transaction Tree取得
def get_audit_tree_by_transaction(hub_id: str, transaction_id: str) -> list[HubAudit] | None:
    cf_client_id = os.getenv("CF_ACCESS_CLIENT_ID", None)
    cf_client_secret = os.getenv("CF_ACCESS_CLIENT_SECRET", None)
    if cf_client_id is None or cf_client_secret is None:
        return None
    url = f"https://audit.shiron-system.net/transaction/{transaction_id}?hub_id={hub_id}"
    headers = {
        "CF-Access-Client-Id": cf_client_id,
        "CF-Access-Client-Secret": cf_client_secret,
        "Content-Type": "application/json",
    }
    res = requests.get(url, headers=headers, timeout=5)
    if res.status_code != 200:
        return None
    return [HubAudit(**line, created_at=line["createdAt"], updated_at=line["updatedAt"]) for line in res.json()]

# 送信
def send_audit_log(
    hub_id: str,
    transaction_type: str,
    extra: dict | None = None,
    depends_on: str | None = None,
) -> None:
    if extra is None:
        extra = {}
    cf_client_id = os.getenv("CF_ACCESS_CLIENT_ID", None)
    cf_client_secret = os.getenv("CF_ACCESS_CLIENT_SECRET", None)
    if cf_client_id is None or cf_client_secret is None:
        return
    url = "https://audit.shiron-system.net/"
    headers = {
        "CF-Access-Client-Id": cf_client_id,
        "CF-Access-Client-Secret": cf_client_secret,
        "Content-Type": "application/json",
    }
    data = {
        "hub_id": hub_id,
        "transaction_id": str(ulid()),
        "transaction_type": transaction_type,
        "prefix": "core",
        "extra": extra,
        "depends_on": depends_on,
    }
    requests.post(url, headers=headers, json=data, timeout=10)

システムログ（運用者向け）

shiron の各システムのシステムログは以下の形式で.log ファイルに保存します。

• timestamp
• level
• name
• message
• extra

収集方法

core システム上に構築した loki を経由して収集し、Grafana Cloud へ送信されます。
また、Prometheus を経由してメトリクスも同様に Grafana Cloud へ送信されます。

Python での実装例

import datetime
from pytz import timezone
from pythonjsonlogger import jsonlogger


class JsonFormatter(jsonlogger.JsonFormatter):

    def parse(self):
        return [
            'timestamp',
            'level',
            'name',
            'message'
        ]

    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)
        if not log_record.get('timestamp'):
            now = datetime.datetime.now(
                timezone('Asia/Tokyo')).strftime('%Y-%m-%dT%H:%M:%S%z')
            log_record['timestamp'] = now
        if log_record.get('level'):
            log_record['level'] = log_record['level'].upper()
        else:
            log_record['level'] = record.levelname

logger = logging.getLogger("shiron-<service name>")
if os.getenv('STAGE') == 'DEVELOPMENT':
    logger.setLevel(logging.DEBUG)
else:
    logger.setLevel(logging.INFO)
handler = logging.handlers.RotatingFileHandler(
    '.logs/shiron-<service name>.log', encoding='utf-8', mode='a', maxBytes=1000000, backupCount=5)
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)