Datadog入門 第5部 - ログ管理の実践完全ガイド ​

# Datadog Log Management の核心概念
class DatadogLogPhilosophy:
    """
    Datadogログ管理の基本原則
    """
    
    def __init__(self):
        self.core_principles = {
            # 1. すべてのログデータの統合
            "unified_logging": {
                "description": "異なるソースからのログを統一プラットフォームで管理",
                "benefits": [
                    "クロスサービス分析の実現",
                    "単一検索インターフェース",
                    "一貫したアクセス制御"
                ]
            },
            
            # 2. 構造化ログの推進
            "structured_logging": {
                "description": "JSON構造でのログ出力推奨",
                "advantages": [
                    "高速検索とフィルタリング",
                    "自動フィールド抽出",
                    "ビジュアライゼーション対応"
                ]
            },
            
            # 3. コンテキスト保持
            "contextual_logging": {
                "description": "メタデータとタグによる豊富なコンテキスト",
                "elements": [
                    "サービス識別情報",
                    "環境・バージョン情報",
                    "ユーザー・セッション情報"
                ]
            },
            
            # 4. 相関分析
            "correlation_analysis": {
                "description": "ログ・メトリクス・トレースの自動相関",
                "capabilities": [
                    "トレースIDによるログ関連付け",
                    "異常メトリクスとログの連携",
                    "タイムライン一元表示"
                ]
            },
            
            # 5. コスト効率性
            "cost_efficiency": {
                "description": "インテリジェントなデータ管理によるコスト最適化",
                "strategies": [
                    "ログレベル別保持期間設定",
                    "自動アーカイブとコールドストレージ",
                    "サンプリングとフィルタリング"
                ]
            }
        }
    
    def demonstrate_value_proposition(self):
        """
        Datadogログ管理の価値提案
        """
        return {
            "開発チーム": [
                "デバッグ時間の大幅短縮",
                "プロダクション環境での詳細可視性",
                "リアルタイムエラー検知とアラート"
            ],
            "運用チーム": [
                "一元化されたログ管理",
                "自動化されたインシデント対応",
                "予防的問題検知"
            ],
            "セキュリティチーム": [
                "セキュリティイベントの統合監視",
                "コンプライアンス監査対応",
                "異常行動パターン検出"
            ],
            "ビジネス": [
                "ユーザー行動の深い洞察",
                "ビジネスKPI追跡",
                "データ駆動型意思決定支援"
            ]
        }

# Python Flask アプリケーションの構造化ログ設定
import json
import logging
import time
from datetime import datetime
from flask import Flask, request, g
from ddtrace import tracer
import structlog

app = Flask(__name__)

# 構造化ログの設定
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

class DatadogLogFormatter:
    """
    Datadog用カスタムログフォーマッター
    """
    
    def format_log(self, level, message, **kwargs):
        """
        構造化ログの生成
        """
        # 基本ログ構造
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": level.upper(),
            "message": message,
            "service": "ecommerce-api",
            "version": "2.1.0",
            "env": "production"
        }
        
        # トレース情報の追加
        span = tracer.current_span()
        if span:
            log_entry.update({
                "dd.trace_id": str(span.trace_id),
                "dd.span_id": str(span.span_id),
                "dd.service": span.service,
                "dd.version": span.get_tag("version")
            })
        
        # リクエストコンテキストの追加
        if hasattr(g, 'request_context'):
            log_entry.update({
                "request": {
                    "id": g.request_context.get("request_id"),
                    "method": g.request_context.get("method"),
                    "path": g.request_context.get("path"),
                    "user_id": g.request_context.get("user_id"),
                    "ip_address": g.request_context.get("ip_address"),
                    "user_agent": g.request_context.get("user_agent")
                }
            })
        
        # 追加コンテキスト
        log_entry.update(kwargs)
        
        return json.dumps(log_entry, ensure_ascii=False)

# カスタムログ関数
def log_info(message, **kwargs):
    print(DatadogLogFormatter().format_log("info", message, **kwargs))

def log_error(message, error=None, **kwargs):
    if error:
        kwargs.update({
            "error": {
                "type": type(error).__name__,
                "message": str(error),
                "stack_trace": str(error.__traceback__) if hasattr(error, '__traceback__') else None
            }
        })
    print(DatadogLogFormatter().format_log("error", message, **kwargs))

def log_warning(message, **kwargs):
    print(DatadogLogFormatter().format_log("warning", message, **kwargs))

# リクエストミドルウェア
@app.before_request
def before_request():
    g.start_time = time.time()
    g.request_context = {
        "request_id": generate_request_id(),
        "method": request.method,
        "path": request.path,
        "ip_address": request.remote_addr,
        "user_agent": request.headers.get("User-Agent"),
        "user_id": get_current_user_id()  # 認証システムから取得
    }
    
    log_info("Request started", 
        request_details=g.request_context,
        query_params=dict(request.args),
        headers={k: v for k, v in request.headers if k.lower() not in ['authorization', 'cookie']}
    )

@app.after_request
def after_request(response):
    duration = (time.time() - g.start_time) * 1000  # ミリ秒
    
    log_info("Request completed",
        request_id=g.request_context.get("request_id"),
        status_code=response.status_code,
        duration_ms=duration,
        response_size=len(response.get_data()) if response.get_data() else 0
    )
    
    return response

# ビジネスロジックでのログ活用例
@app.route('/api/products/<int:product_id>')
def get_product(product_id):
    try:
        log_info("Product fetch initiated",
            product_id=product_id,
            operation="product_fetch"
        )
        
        # キャッシュから取得試行
        cached_product = cache.get(f"product:{product_id}")
        if cached_product:
            log_info("Product cache hit",
                product_id=product_id,
                cache_key=f"product:{product_id}",
                data_source="cache"
            )
            return cached_product
        
        # データベースから取得
        log_info("Database query initiated",
            product_id=product_id,
            query_type="product_select",
            data_source="database"
        )
        
        product = Product.query.get(product_id)
        
        if not product:
            log_warning("Product not found",
                product_id=product_id,
                result="not_found"
            )
            return {"error": "Product not found"}, 404
        
        # キャッシュに保存
        cache.set(f"product:{product_id}", product.to_dict(), timeout=3600)
        
        log_info("Product fetch completed",
            product_id=product_id,
            product_name=product.name,
            product_category=product.category,
            operation_result="success"
        )
        
        return product.to_dict()
        
    except DatabaseError as e:
        log_error("Database error occurred",
            product_id=product_id,
            error=e,
            operation="product_fetch",
            error_category="database"
        )
        return {"error": "Internal server error"}, 500
        
    except Exception as e:
        log_error("Unexpected error occurred",
            product_id=product_id,
            error=e,
            operation="product_fetch",
            error_category="unexpected"
        )
        return {"error": "Internal server error"}, 500

def generate_request_id():
    """リクエストID生成"""
    import uuid
    return str(uuid.uuid4())

def get_current_user_id():
    """現在のユーザーID取得（認証システムから）"""
    # 実際の認証システムと連携
    return request.headers.get("X-User-ID")

# Datadog Log Pipeline API を使用したパーサー作成例
import requests
import json

class DatadogLogPipelineManager:
    """
    Datadog Log Pipeline 管理クラス
    """
    
    def __init__(self, api_key, app_key):
        self.api_key = api_key
        self.app_key = app_key
        self.base_url = "https://api.datadoghq.com/api/v1"
        self.headers = {
            "Content-Type": "application/json",
            "DD-API-KEY": api_key,
            "DD-APPLICATION-KEY": app_key
        }
    
    def create_grok_parser(self, name, grok_pattern, samples=None):
        """
        Grokパーサーの作成
        """
        parser_config = {
            "type": "grok-parser",
            "name": name,
            "is_enabled": True,
            "source": "message",
            "grok": {
                "support_rules": "",
                "match_rules": grok_pattern
            }
        }
        
        if samples:
            parser_config["samples"] = samples
            
        return self._create_parser(parser_config)
    
    def create_json_parser(self, name, source_field="message"):
        """
        JSONパーサーの作成
        """
        parser_config = {
            "type": "json-parser",
            "name": name,
            "is_enabled": True,
            "source": source_field
        }
        
        return self._create_parser(parser_config)
    
    def create_date_parser(self, name, date_format, source_field="timestamp"):
        """
        日付パーサーの作成
        """
        parser_config = {
            "type": "date-parser",
            "name": name,
            "is_enabled": True,
            "source": source_field,
            "target": "@timestamp",
            "format": date_format
        }
        
        return self._create_parser(parser_config)
    
    def create_nginx_pipeline(self):
        """
        Nginxアクセスログ用パイプライン
        """
        # Nginxログフォーマット: 
        # log_format combined '$remote_addr - $remote_user [$time_local] '
        #                     '"$request" $status $body_bytes_sent '
        #                     '"$http_referer" "$http_user_agent"';
        
        grok_pattern = """
        nginx_access %{IPORHOST:network.client.ip} - %{USER:user.name} \\[%{HTTPDATE:timestamp}\\] "%{WORD:http.method} %{URIPATH:http.url_details.path}(?:%{URIPARAM:http.url_details.queryString})? HTTP/%{NUMBER:http.version}" %{INT:http.status_code} %{INT:network.bytes_written} "%{DATA:http.referer}" "%{DATA:http.useragent}"
        """
        
        pipeline_config = {
            "name": "Nginx Access Logs",
            "is_enabled": True,
            "filter": {
                "query": "source:nginx"
            },
            "processors": [
                # 1. Grokパーサー
                {
                    "type": "grok-parser",
                    "name": "Parse Nginx Access Log",
                    "is_enabled": True,
                    "source": "message",
                    "grok": {
                        "support_rules": "",
                        "match_rules": grok_pattern
                    }
                },
                # 2. 日付パーサー
                {
                    "type": "date-parser",
                    "name": "Parse Timestamp",
                    "is_enabled": True,
                    "source": "timestamp",
                    "target": "@timestamp",
                    "format": "dd/MMM/yyyy:HH:mm:ss Z"
                },
                # 3. ステータスコードカテゴリ
                {
                    "type": "category-processor",
                    "name": "Categorize Status Code",
                    "is_enabled": True,
                    "target": "http.status_category",
                    "categories": [
                        {"filter": {"query": "@http.status_code:[200 TO 299]"}, "name": "OK"},
                        {"filter": {"query": "@http.status_code:[300 TO 399]"}, "name": "Redirect"},
                        {"filter": {"query": "@http.status_code:[400 TO 499]"}, "name": "Client Error"},
                        {"filter": {"query": "@http.status_code:[500 TO 599]"}, "name": "Server Error"}
                    ]
                },
                # 4. User Agentパーサー
                {
                    "type": "user-agent-parser",
                    "name": "Parse User Agent",
                    "is_enabled": True,
                    "source": "http.useragent",
                    "target": "http.useragent_details"
                },
                # 5. URLパーサー
                {
                    "type": "url-parser",
                    "name": "Parse URL",
                    "is_enabled": True,
                    "source": "http.url_details.path",
                    "target": "http.url_details"
                },
                # 6. 地理的情報の抽出
                {
                    "type": "geo-ip-parser",
                    "name": "Enrich with GeoIP",
                    "is_enabled": True,
                    "source": "network.client.ip",
                    "target": "network.client.geoip"
                }
            ]
        }
        
        return self._create_pipeline(pipeline_config)
    
    def create_application_error_pipeline(self):
        """
        アプリケーションエラーログ用パイプライン
        """
        pipeline_config = {
            "name": "Application Error Logs",
            "is_enabled": True,
            "filter": {
                "query": "source:python level:error"
            },
            "processors": [
                # 1. JSONパーサー
                {
                    "type": "json-parser",
                    "name": "Parse JSON Log",
                    "is_enabled": True,
                    "source": "message"
                },
                # 2. ログレベルの正規化
                {
                    "type": "string-builder-processor",
                    "name": "Normalize Log Level",
                    "is_enabled": True,
                    "target": "level",
                    "template": "%{level}",
                    "is_replace_missing": False
                },
                # 3. エラータイプの分類
                {
                    "type": "category-processor",
                    "name": "Categorize Error Type",
                    "is_enabled": True,
                    "target": "error.category",
                    "categories": [
                        {"filter": {"query": "error.type:DatabaseError"}, "name": "Database"},
                        {"filter": {"query": "error.type:ValidationError"}, "name": "Validation"},
                        {"filter": {"query": "error.type:PermissionError"}, "name": "Authorization"},
                        {"filter": {"query": "error.type:TimeoutError"}, "name": "Timeout"},
                        {"filter": {"query": "error.type:ConnectionError"}, "name": "Network"}
                    ]
                },
                # 4. スタックトレースの抽出
                {
                    "type": "trace-id-remapper",
                    "name": "Map Trace ID",
                    "is_enabled": True,
                    "sources": ["dd.trace_id"]
                },
                # 5. アラート用タグ追加
                {
                    "type": "string-builder-processor",
                    "name": "Add Alert Tag",
                    "is_enabled": True,
                    "target": "alert.required",
                    "template": "true",
                    "is_replace_missing": False
                }
            ]
        }
        
        return self._create_pipeline(pipeline_config)
    
    def _create_parser(self, parser_config):
        """
        パーサー作成の内部メソッド
        """
        url = f"{self.base_url}/logs/config/processors"
        response = requests.post(url, headers=self.headers, data=json.dumps(parser_config))
        return response.json()
    
    def _create_pipeline(self, pipeline_config):
        """
        パイプライン作成の内部メソッド
        """
        url = f"{self.base_url}/logs/config/pipelines"
        response = requests.post(url, headers=self.headers, data=json.dumps(pipeline_config))
        return response.json()

# 使用例
pipeline_manager = DatadogLogPipelineManager(
    api_key="your-api-key",
    app_key="your-app-key"
)

# Nginxパイプラインの作成
nginx_pipeline = pipeline_manager.create_nginx_pipeline()

# アプリケーションエラーパイプラインの作成
error_pipeline = pipeline_manager.create_application_error_pipeline()

# Datadog Log Archive API を使用したアーカイブ設定
class DatadogLogArchiveManager:
    """
    Datadogログアーカイブ管理クラス
    """
    
    def __init__(self, api_key, app_key):
        self.api_key = api_key
        self.app_key = app_key
        self.base_url = "https://api.datadoghq.com/api/v2"
        self.headers = {
            "Content-Type": "application/json",
            "DD-API-KEY": api_key,
            "DD-APPLICATION-KEY": app_key
        }
    
    def create_s3_archive(self, name, bucket, path, access_key_id, secret_access_key, region="us-east-1"):
        """
        S3ログアーカイブの作成
        """
        archive_config = {
            "data": {
                "type": "archives",
                "attributes": {
                    "name": name,
                    "query": "*",  # すべてのログをアーカイブ
                    "destination": {
                        "type": "s3",
                        "container": bucket,
                        "path": path,
                        "region": region,
                        "account_id": "",
                        "integration": {
                            "access_key_id": access_key_id,
                            "secret_access_key": secret_access_key
                        }
                    },
                    "rehydration_tags": ["env", "service", "version"],
                    "include_tags": True
                }
            }
        }
        
        url = f"{self.base_url}/logs/config/archives"
        response = requests.post(url, headers=self.headers, data=json.dumps(archive_config))
        return response.json()
    
    def create_tiered_archive_strategy(self):
        """
        階層化アーカイブ戦略の実装
        """
        strategies = []
        
        # ホットデータ（0-7日）: インデックス済み、高速検索
        hot_archive = {
            "name": "Hot Data Archive",
            "query": "*",
            "retention_days": 7,
            "storage_class": "hot",
            "description": "Recent logs for real-time analysis"
        }
        strategies.append(hot_archive)
        
        # ウォームデータ（8-30日）: 圧縮済み、中速検索
        warm_archive = {
            "name": "Warm Data Archive", 
            "query": "level:(warn OR error OR critical)",
            "retention_days": 30,
            "storage_class": "warm",
            "description": "Important logs for trend analysis"
        }
        strategies.append(warm_archive)
        
        # コールドデータ（31日-1年）: 高圧縮、検索には時間要
        cold_archive = {
            "name": "Cold Data Archive",
            "query": "level:error OR source:security",
            "retention_days": 365,
            "storage_class": "cold",
            "description": "Critical logs for compliance and audit"
        }
        strategies.append(cold_archive)
        
        return strategies
    
    def create_compliance_archive(self):
        """
        コンプライアンス要件に対応するアーカイブ設定
        """
        compliance_configs = {
            # GDPR対応: 個人データの適切な管理
            "gdpr_archive": {
                "name": "GDPR Compliance Archive",
                "query": "source:application @user_id:*",
                "retention_days": 2555,  # 7年間
                "encryption": True,
                "immutable": True,
                "audit_logging": True,
                "data_classification": "personal_data"
            },
            
            # SOX対応: 財務関連ログの長期保存
            "sox_archive": {
                "name": "SOX Compliance Archive", 
                "query": "source:financial OR service:payment",
                "retention_days": 2555,  # 7年間
                "encryption": True,
                "immutable": True,
                "audit_logging": True,
                "data_classification": "financial_data"
            },
            
            # PCI DSS対応: 決済関連ログ
            "pci_archive": {
                "name": "PCI DSS Compliance Archive",
                "query": "source:payment OR @credit_card:*",
                "retention_days": 365,  # 1年間
                "encryption": True,
                "immutable": True,
                "audit_logging": True,
                "data_classification": "payment_data"
            }
        }
        
        return compliance_configs
    
    def calculate_storage_costs(self, daily_log_volume_gb, retention_days, storage_tier="warm"):
        """
        ストレージコストの計算
        """
        # AWS S3ストレージ料金例（2024年基準）
        storage_rates = {
            "hot": 0.023,    # $0.023/GB/月 (S3 Standard)
            "warm": 0.0125,  # $0.0125/GB/月 (S3 Standard-IA)
            "cold": 0.004    # $0.004/GB/月 (S3 Glacier)
        }
        
        # 月間ストレージ容量の計算
        monthly_storage_gb = daily_log_volume_gb * retention_days
        
        # 月間コスト
        monthly_cost = monthly_storage_gb * storage_rates[storage_tier]
        
        # 年間コスト
        annual_cost = monthly_cost * 12
        
        return {
            "daily_volume_gb": daily_log_volume_gb,
            "retention_days": retention_days,
            "storage_tier": storage_tier,
            "monthly_storage_gb": monthly_storage_gb,
            "monthly_cost_usd": round(monthly_cost, 2),
            "annual_cost_usd": round(annual_cost, 2),
            "cost_per_gb_month": storage_rates[storage_tier]
        }
    
    def optimize_log_retention(self, services_config):
        """
        サービス別ログ保持期間の最適化
        """
        optimized_config = {}
        
        for service, config in services_config.items():
            log_volume = config.get("daily_volume_gb", 0)
            criticality = config.get("criticality", "medium")
            compliance_required = config.get("compliance_required", False)
            
            # サービス重要度に基づく保持期間
            if criticality == "critical":
                hot_retention = 30
                warm_retention = 90
                cold_retention = 365
            elif criticality == "high":
                hot_retention = 14
                warm_retention = 60
                cold_retention = 180
            elif criticality == "medium":
                hot_retention = 7
                warm_retention = 30
                cold_retention = 90
            else:  # low
                hot_retention = 3
                warm_retention = 14
                cold_retention = 30
            
            # コンプライアンス要件の調整
            if compliance_required:
                cold_retention = max(cold_retention, 2555)  # 7年間
            
            optimized_config[service] = {
                "hot_retention_days": hot_retention,
                "warm_retention_days": warm_retention,
                "cold_retention_days": cold_retention,
                "estimated_costs": {
                    "hot": self.calculate_storage_costs(log_volume, hot_retention, "hot"),
                    "warm": self.calculate_storage_costs(log_volume, warm_retention, "warm"),
                    "cold": self.calculate_storage_costs(log_volume, cold_retention, "cold")
                }
            }
        
        return optimized_config

# 使用例
archive_manager = DatadogLogArchiveManager(
    api_key="your-api-key",
    app_key="your-app-key"
)

# サービス設定例
services = {
    "web-api": {
        "daily_volume_gb": 5.2,
        "criticality": "critical",
        "compliance_required": True
    },
    "background-worker": {
        "daily_volume_gb": 1.8,
        "criticality": "high", 
        "compliance_required": False
    },
    "frontend": {
        "daily_volume_gb": 0.5,
        "criticality": "medium",
        "compliance_required": False
    }
}

# アーカイブ戦略の最適化
optimized_config = archive_manager.optimize_log_retention(services)
print(json.dumps(optimized_config, indent=2))

# Datadog Log Search API を使用した高度検索
class DatadogLogSearcher:
    """
    Datadog高度ログ検索クラス
    """
    
    def __init__(self, api_key, app_key):
        self.api_key = api_key
        self.app_key = app_key
        self.base_url = "https://api.datadoghq.com/api/v1"
        self.headers = {
            "Content-Type": "application/json",
            "DD-API-KEY": api_key,
            "DD-APPLICATION-KEY": app_key
        }
    
    def search_error_patterns(self, service, time_range="1h", limit=100):
        """
        エラーパターンの検索と分析
        """
        query = f"service:{service} level:error"
        
        search_params = {
            "query": query,
            "time": {
                "from": f"now-{time_range}",
                "to": "now"
            },
            "sort": "desc",
            "limit": limit
        }
        
        url = f"{self.base_url}/logs-queries/list"
        response = requests.post(url, headers=self.headers, data=json.dumps(search_params))
        
        if response.status_code == 200:
            logs = response.json()
            return self._analyze_error_patterns(logs)
        else:
            return {"error": "Search failed", "status": response.status_code}
    
    def _analyze_error_patterns(self, logs):
        """
        エラーパターンの分析
        """
        error_analysis = {
            "total_errors": len(logs.get("logs", [])),
            "error_types": {},
            "error_frequency": {},
            "affected_endpoints": {},
            "time_distribution": {}
        }
        
        for log in logs.get("logs", []):
            # エラータイプの集計
            error_type = log.get("attributes", {}).get("error", {}).get("type", "Unknown")
            error_analysis["error_types"][error_type] = error_analysis["error_types"].get(error_type, 0) + 1
            
            # エンドポイント分析
            endpoint = log.get("attributes", {}).get("request", {}).get("path", "Unknown")
            error_analysis["affected_endpoints"][endpoint] = error_analysis["affected_endpoints"].get(endpoint, 0) + 1
            
            # 時間分布
            timestamp = log.get("attributes", {}).get("@timestamp")
            if timestamp:
                hour = timestamp[:13]  # YYYY-MM-DDTHH形式
                error_analysis["time_distribution"][hour] = error_analysis["time_distribution"].get(hour, 0) + 1
        
        return error_analysis
    
    def search_performance_anomalies(self, service, duration_threshold_ms=5000, time_range="4h"):
        """
        パフォーマンス異常の検索
        """
        query = f"service:{service} @duration:>[{duration_threshold_ms}]"
        
        search_params = {
            "query": query,
            "time": {
                "from": f"now-{time_range}",
                "to": "now"
            },
            "sort": "desc",
            "limit": 100,
            "aggregation": {
                "type": "count",
                "group_by": [
                    {"field": "@http.method"},
                    {"field": "@http.url_details.path"}
                ]
            }
        }
        
        url = f"{self.base_url}/logs-queries/list"
        response = requests.post(url, headers=self.headers, data=json.dumps(search_params))
        
        return response.json()
    
    def search_security_events(self, time_range="24h"):
        """
        セキュリティイベントの検索
        """
        security_queries = [
            # 認証失敗
            "source:auth level:warning @event_type:authentication_failed",
            # 権限エラー
            "level:error @http.status_code:403",
            # 異常なアクセスパターン
            "@network.client.ip:* @http.status_code:404",
            # SQLインジェクション試行
            "message:*UNION* OR message:*SELECT* AND level:warning"
        ]
        
        security_analysis = {}
        
        for query in security_queries:
            search_params = {
                "query": query,
                "time": {
                    "from": f"now-{time_range}",
                    "to": "now"
                },
                "sort": "desc",
                "limit": 50
            }
            
            url = f"{self.base_url}/logs-queries/list"
            response = requests.post(url, headers=self.headers, data=json.dumps(search_params))
            
            if response.status_code == 200:
                security_analysis[query] = response.json()
        
        return security_analysis
    
    def create_saved_view(self, name, query, columns=None):
        """
        保存されたビューの作成
        """
        if columns is None:
            columns = ["@timestamp", "level", "service", "message"]
        
        view_config = {
            "data": {
                "type": "logs_saved_views",
                "attributes": {
                    "name": name,
                    "query": query,
                    "columns": columns
                }
            }
        }
        
        url = f"{self.base_url}/logs/config/saved_views"
        response = requests.post(url, headers=self.headers, data=json.dumps(view_config))
        return response.json()
    
    def live_tail_errors(self, service, callback_function):
        """
        エラーログのライブテール（WebSocket使用）
        """
        import websocket
        import json
        
        def on_message(ws, message):
            log_data = json.loads(message)
            if log_data.get("level") in ["error", "critical", "fatal"]:
                callback_function(log_data)
        
        def on_error(ws, error):
            print(f"WebSocket error: {error}")
        
        def on_close(ws, close_status_code, close_msg):
            print("WebSocket connection closed")
        
        def on_open(ws):
            # ライブテールクエリの送信
            subscribe_message = {
                "action": "subscribe",
                "query": f"service:{service} level:error"
            }
            ws.send(json.dumps(subscribe_message))
        
        websocket_url = f"wss://api.datadoghq.com/v1/logs/tail?dd-api-key={self.api_key}"
        
        ws = websocket.WebSocketApp(
            websocket_url,
            on_open=on_open,
            on_message=on_message,
            on_error=on_error,
            on_close=on_close
        )
        
        ws.run_forever()

# 使用例
searcher = DatadogLogSearcher(
    api_key="your-api-key",
    app_key="your-app-key"
)

# エラーパターン分析
error_patterns = searcher.search_error_patterns("web-api", "2h")
print("Error Patterns:", json.dumps(error_patterns, indent=2))

# パフォーマンス異常検索
performance_issues = searcher.search_performance_anomalies("web-api", 3000, "1h")

# セキュリティイベント検索
security_events = searcher.search_security_events("6h")

# ライブテールの設定
def handle_error_log(log_data):
    print(f"ALERT: New error in {log_data.get('service')}: {log_data.get('message')}")
    # Slack通知やPagerDuty連携などの処理

# searcher.live_tail_errors("web-api", handle_error_log)

# Datadog Log-based Metrics API
class DatadogLogMetricsManager:
    """
    Datadogログベースメトリクス管理クラス
    """
    
    def __init__(self, api_key, app_key):
        self.api_key = api_key
        self.app_key = app_key
        self.base_url = "https://api.datadoghq.com/api/v2"
        self.headers = {
            "Content-Type": "application/json",
            "DD-API-KEY": api_key,
            "DD-APPLICATION-KEY": app_key
        }
    
    def create_error_rate_metric(self):
        """
        エラー率メトリクスの作成
        """
        metric_config = {
            "data": {
                "type": "logs_metrics",
                "id": "error_rate_by_service",
                "attributes": {
                    "compute": {
                        "aggregation_type": "count"
                    },
                    "filter": {
                        "query": "level:error"
                    },
                    "group_by": [
                        {
                            "facet": "service",
                            "limit": 10,
                            "sort": {
                                "aggregation": "count",
                                "order": "desc"
                            }
                        }
                    ]
                }
            }
        }
        
        return self._create_metric(metric_config)
    
    def create_response_time_metric(self):
        """
        レスポンス時間メトリクスの作成
        """
        metric_config = {
            "data": {
                "type": "logs_metrics", 
                "id": "avg_response_time",
                "attributes": {
                    "compute": {
                        "aggregation_type": "distribution",
                        "path": "@duration"
                    },
                    "filter": {
                        "query": "source:nginx @duration:*"
                    },
                    "group_by": [
                        {
                            "facet": "service",
                            "limit": 10
                        },
                        {
                            "facet": "@http.url_details.path",
                            "limit": 20
                        }
                    ]
                }
            }
        }
        
        return self._create_metric(metric_config)
    
    def create_business_kpi_metrics(self):
        """
        ビジネスKPIメトリクスの作成
        """
        business_metrics = []
        
        # 1. ユーザー登録数
        user_registration_metric = {
            "data": {
                "type": "logs_metrics",
                "id": "user_registrations",
                "attributes": {
                    "compute": {
                        "aggregation_type": "count"
                    },
                    "filter": {
                        "query": "source:application @event_type:user_registered"
                    },
                    "group_by": [
                        {
                            "facet": "@user.source",
                            "limit": 10
                        }
                    ]
                }
            }
        }
        business_metrics.append(user_registration_metric)
        
        # 2. 購入金額合計
        purchase_amount_metric = {
            "data": {
                "type": "logs_metrics",
                "id": "total_purchase_amount",
                "attributes": {
                    "compute": {
                        "aggregation_type": "sum",
                        "path": "@order.total"
                    },
                    "filter": {
                        "query": "source:ecommerce @event_type:purchase_completed"
                    },
                    "group_by": [
                        {
                            "facet": "@order.payment_method",
                            "limit": 5
                        },
                        {
                            "facet": "@user.segment",
                            "limit": 10
                        }
                    ]
                }
            }
        }
        business_metrics.append(purchase_amount_metric)
        
        # 3. API使用量
        api_usage_metric = {
            "data": {
                "type": "logs_metrics",
                "id": "api_usage_by_endpoint", 
                "attributes": {
                    "compute": {
                        "aggregation_type": "count"
                    },
                    "filter": {
                        "query": "source:api-gateway @http.method:*"
                    },
                    "group_by": [
                        {
                            "facet": "@http.method",
                            "limit": 10
                        },
                        {
                            "facet": "@http.url_details.path",
                            "limit": 50
                        },
                        {
                            "facet": "@client.api_key",
                            "limit": 100
                        }
                    ]
                }
            }
        }
        business_metrics.append(api_usage_metric)
        
        return business_metrics
    
    def create_security_metrics(self):
        """
        セキュリティメトリクスの作成
        """
        security_metrics = []
        
        # 1. 認証失敗回数
        auth_failures_metric = {
            "data": {
                "type": "logs_metrics",
                "id": "authentication_failures",
                "attributes": {
                    "compute": {
                        "aggregation_type": "count"
                    },
                    "filter": {
                        "query": "source:auth @event_type:authentication_failed"
                    },
                    "group_by": [
                        {
                            "facet": "@network.client.ip",
                            "limit": 100
                        },
                        {
                            "facet": "@auth.failure_reason",
                            "limit": 10
                        }
                    ]
                }
            }
        }
        security_metrics.append(auth_failures_metric)
        
        # 2. 異常アクセスパターン
        anomalous_access_metric = {
            "data": {
                "type": "logs_metrics",
                "id": "anomalous_access_patterns",
                "attributes": {
                    "compute": {
                        "aggregation_type": "count"
                    },
                    "filter": {
                        "query": "@http.status_code:[400 TO 499] OR @suspicious_activity:true"
                    },
                    "group_by": [
                        {
                            "facet": "@network.client.ip",
                            "limit": 50
                        },
                        {
                            "facet": "@http.status_code",
                            "limit": 10
                        }
                    ]
                }
            }
        }
        security_metrics.append(anomalous_access_metric)
        
        return security_metrics
    
    def _create_metric(self, metric_config):
        """
        メトリクス作成の内部メソッド
        """
        url = f"{self.base_url}/logs/config/metrics"
        response = requests.post(url, headers=self.headers, data=json.dumps(metric_config))
        return response.json()

# ログベースメトリクスの使用例
metrics_manager = DatadogLogMetricsManager(
    api_key="your-api-key",
    app_key="your-app-key"
)

# 各種メトリクスの作成
error_rate_metric = metrics_manager.create_error_rate_metric()
response_time_metric = metrics_manager.create_response_time_metric()
business_metrics = metrics_manager.create_business_kpi_metrics()
security_metrics = metrics_manager.create_security_metrics()