Issue #140: 運用改善: 監視・分析・運用ダッシュボードの実装

概要

本番運用に向けた包括的な監視・分析・運用管理システムの実装

背景

• 現状: 基本的なヘルスチェックのみ
• 課題: 性能問題、障害の早期発見が困難
• 需要: データドリブンな改善、運用効率化

実装機能

1. システム監視

# Prometheus + Grafana 統合
from prometheus_client import Counter, Histogram, Gauge

# メトリクス定義
memory_operations_total = Counter(
    'mory_memory_operations_total',
    'Total memory operations',
    ['operation_type', 'status']
)

api_response_time = Histogram(
    'mory_api_response_time_seconds',
    'API response time in seconds',
    ['endpoint', 'method']
)

active_connections = Gauge(
    'mory_active_connections',
    'Number of active WebSocket connections'
)

2. アプリケーション分析

# 使用パターン分析
@router.post("/api/memories")
async def create_memory(memory_data: MemoryCreate):
    # ビジネスメトリクス収集
    await analytics.track_event("memory_created", {
        "user_id": current_user.id,
        "content_length": len(memory_data.value),
        "ai_processing_enabled": True,
        "timestamp": datetime.utcnow()
    })
    
    # パフォーマンス測定
    with performance_timer("memory_creation"):
        result = await memory_service.create(memory_data)
    
    return result

3. 運用ダッシュボード

# 管理者向けAPI
@router.get("/api/admin/stats")
async def get_system_stats():
    return {
        "system": {
            "memory_usage": psutil.virtual_memory().percent,
            "cpu_usage": psutil.cpu_percent(),
            "disk_usage": psutil.disk_usage('/').percent
        },
        "application": {
            "total_memories": await get_total_memories(),
            "active_users": await get_active_users_count(),
            "ai_processing_queue": await get_queue_size(),
            "error_rate_24h": await get_error_rate()
        },
        "performance": {
            "avg_response_time": await get_avg_response_time(),
            "requests_per_minute": await get_requests_per_minute(),
            "cache_hit_rate": await get_cache_hit_rate()
        }
    }

技術実装

データ収集基盤

# 構造化ログ
import structlog

logger = structlog.get_logger()

class RequestLoggingMiddleware:
    async def __call__(self, request: Request, call_next):
        start_time = time.time()
        
        # リクエスト情報記録
        logger.info(
            "request_started",
            method=request.method,
            path=request.url.path,
            user_agent=request.headers.get("user-agent"),
            ip=request.client.host
        )
        
        response = await call_next(request)
        
        # レスポンス情報記録
        logger.info(
            "request_completed",
            method=request.method,
            path=request.url.path,
            status_code=response.status_code,
            duration=time.time() - start_time
        )
        
        return response

エラー追跡

# Sentry統合
import sentry_sdk
from sentry_sdk.integrations.fastapi import FastApiIntegration

sentry_sdk.init(
    dsn="your-sentry-dsn",
    integrations=[FastApiIntegration()],
    traces_sample_rate=0.1,
    environment="production"
)

# カスタムエラーハンドリング
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 詳細なエラー情報収集
    error_context = {
        "request_id": str(uuid.uuid4()),
        "user_id": getattr(request.state, "user_id", None),
        "endpoint": request.url.path,
        "method": request.method,
        "timestamp": datetime.utcnow().isoformat()
    }
    
    logger.error(
        "unhandled_exception",
        exc_info=exc,
        extra=error_context
    )
    
    return JSONResponse(
        status_code=500,
        content={
            "error": "Internal server error",
            "request_id": error_context["request_id"]
        }
    )

性能分析

# APM (Application Performance Monitoring)
from elastic_apm.contrib.starlette import ElasticAPM

app.add_middleware(ElasticAPM, config={
    'SERVICE_NAME': 'mory-api',
    'SECRET_TOKEN': 'your-apm-token',
    'SERVER_URL': 'https://your-apm-server.com:8200',
    'ENVIRONMENT': 'production',
})

# N+1クエリ検出
class QueryAnalyzer:
    def __init__(self):
        self.query_count = 0
        self.queries = []
    
    def track_query(self, query: str):
        self.query_count += 1
        self.queries.append({
            "sql": query,
            "timestamp": time.time()
        })
        
        if self.query_count > 10:
            logger.warning(
                "potential_n_plus_one",
                query_count=self.query_count,
                queries=self.queries
            )

ビジネス分析

# ユーザー行動分析
class UserAnalytics:
    async def track_user_session(self, user_id: str, session_data: dict):
        """ユーザーセッション追跡"""
        await self.event_store.save({
            "event_type": "session",
            "user_id": user_id,
            "data": session_data,
            "timestamp": datetime.utcnow()
        })
    
    async def analyze_memory_patterns(self, user_id: str) -> dict:
        """メモリ使用パターン分析"""
        memories = await get_user_memories(user_id)
        
        return {
            "creation_frequency": self._calculate_frequency(memories),
            "popular_tags": self._extract_popular_tags(memories),
            "optimal_times": self._find_optimal_creation_times(memories),
            "content_types": self._categorize_content_types(memories)
        }

データベース拡張

分析用テーブル

-- システムメトリクス
CREATE TABLE system_metrics (
    id TEXT PRIMARY KEY,
    metric_name TEXT NOT NULL,
    metric_value REAL NOT NULL,
    labels JSON,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- ユーザーイベント
CREATE TABLE user_events (
    id TEXT PRIMARY KEY,
    user_id TEXT,
    event_type TEXT NOT NULL,
    event_data JSON,
    session_id TEXT,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- パフォーマンス履歴
CREATE TABLE performance_logs (
    id TEXT PRIMARY KEY,
    endpoint TEXT,
    method TEXT,
    response_time_ms INTEGER,
    status_code INTEGER,
    user_id TEXT,
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);

インデックス最適化

-- 分析クエリ高速化
CREATE INDEX idx_user_events_user_timestamp ON user_events(user_id, timestamp);
CREATE INDEX idx_performance_endpoint_timestamp ON performance_logs(endpoint, timestamp);
CREATE INDEX idx_system_metrics_name_timestamp ON system_metrics(metric_name, timestamp);

可視化・アラート

Grafanaダッシュボード

# grafana-dashboard.json
{
  "dashboard": {
    "title": "Mory System Dashboard",
    "panels": [
      {
        "title": "API Response Time",
        "type": "stat",
        "targets": [
          {
            "expr": "avg(mory_api_response_time_seconds)",
            "legendFormat": "Avg Response Time"
          }
        ]
      },
      {
        "title": "Memory Operations",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(mory_memory_operations_total[5m])",
            "legendFormat": "{{operation_type}}"
          }
        ]
      }
    ]
  }
}

アラート設定

# AlertManager設定
class AlertManager:
    async def check_system_health(self):
        """システムヘルス監視"""
        metrics = await self.collect_metrics()
        
        if metrics['error_rate'] > 0.05:  # 5%以上のエラー率
            await self.send_alert({
                "severity": "critical",
                "message": "High error rate detected",
                "metrics": metrics
            })
        
        if metrics['response_time'] > 2.0:  # 2秒以上のレスポンス
            await self.send_alert({
                "severity": "warning", 
                "message": "High response time detected",
                "metrics": metrics
            })

実装計画

Phase 1: 基本監視 (2週間)

• システムメトリクス収集
• 構造化ログ実装
• 基本ダッシュボード

Phase 2: 高度な分析 (2週間)

• ユーザー行動分析
• 性能プロファイリング
• エラー追跡システム

Phase 3: 運用自動化 (1週間)

• アラートシステム
• 自動復旧機能
• キャパシティプランニング

期待効果

• 信頼性向上: 障害の早期発見・対応
• 性能最適化: データドリブンな改善
• ユーザー体験: 使用パターンに基づく機能改善
• 運用効率: 自動化による工数削減

技術スタック

• 監視: Prometheus + Grafana
• ログ: structlog + ELK Stack
• エラー追跡: Sentry
• APM: Elastic APM
• アラート: AlertManager + Slack/Email