New Relic APM入門 第5章 - APM高度活用とコードレベル分析 ​

# インシデント分析支援ツール
import requests
import json
from datetime import datetime, timedelta

class IncidentAnalyzer:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.newrelic.com"
    
    def analyze_incident(self, incident_id, service_name):
        """包括的なインシデント分析を実行"""
        analysis = {
            "timeline": self.get_incident_timeline(incident_id),
            "related_events": self.get_related_events(service_name),
            "performance_impact": self.calculate_impact(service_name),
            "root_cause_hints": self.identify_root_cause_patterns(service_name)
        }
        
        return self.generate_analysis_report(analysis)
    
    def get_incident_timeline(self, incident_id):
        """インシデントのタイムライン構築"""
        end_time = datetime.now()
        start_time = end_time - timedelta(hours=2)
        
        nrql_query = f"""
        SELECT timestamp, eventType, message, severity
        FROM Log, Transaction, PageView, SystemSample
        WHERE timestamp > {start_time.timestamp()}
        AND timestamp < {end_time.timestamp()}
        ORDER BY timestamp
        """
        
        return self.execute_nrql(nrql_query)
    
    def identify_root_cause_patterns(self, service_name):
        """ルートコーズパターンの特定"""
        patterns = []
        
        # デプロイメント相関チェック
        deployment_query = f"""
        SELECT timestamp, version, releaseTag
        FROM Deployment
        WHERE appName = '{service_name}'
        SINCE 1 day ago
        """
        
        deployments = self.execute_nrql(deployment_query)
        if deployments:
            patterns.append({
                "type": "deployment_correlation",
                "confidence": 0.85,
                "description": "Recent deployment may be related to the incident"
            })
        
        # 外部依存関係チェック
        dependency_query = f"""
        SELECT average(duration), count(*)
        FROM Transaction
        WHERE appName = '{service_name}'
        AND name LIKE '%external%'
        FACET name
        SINCE 1 hour ago
        """
        
        # リソース使用率チェック
        resource_query = f"""
        SELECT max(cpuPercent), max(memoryUsedPercent)
        FROM SystemSample
        WHERE hostname IN (
          SELECT uniques(host)
          FROM Transaction
          WHERE appName = '{service_name}'
        )
        SINCE 1 hour ago
        """
        
        return patterns

# PIR（Post-Incident Review）レポート自動生成
class PIRGenerator:
    def __init__(self, api_key):
        self.nr_client = NewRelicClient(api_key)
    
    def generate_pir_report(self, incident_id, start_time, end_time):
        """包括的なPIRレポート生成"""
        report = {
            "incident_summary": self.get_incident_summary(incident_id),
            "timeline": self.create_incident_timeline(start_time, end_time),
            "impact_analysis": self.analyze_business_impact(start_time, end_time),
            "root_cause": self.determine_root_cause(incident_id),
            "lessons_learned": self.extract_lessons_learned(incident_id),
            "action_items": self.generate_action_items(incident_id)
        }
        
        return self.format_pir_report(report)
    
    def analyze_business_impact(self, start_time, end_time):
        """ビジネスインパクト分析"""
        impact_query = f"""
        SELECT 
          count(*) as total_requests,
          filter(count(*), WHERE httpResponseCode >= 500) as error_count,
          average(duration) as avg_response_time,
          percentile(duration, 95) as p95_response_time
        FROM Transaction
        WHERE timestamp >= '{start_time}'
        AND timestamp <= '{end_time}'
        TIMESERIES 5 minutes
        """
        
        metrics = self.nr_client.execute_nrql(impact_query)
        
        # ビジネスメトリクス計算
        revenue_impact = self.calculate_revenue_impact(metrics)
        user_impact = self.calculate_user_impact(metrics)
        
        return {
            "revenue_impact": revenue_impact,
            "user_impact": user_impact,
            "performance_degradation": metrics
        }

# 適応的サンプリング戦略
class AdaptiveSampling:
    def __init__(self, nr_client):
        self.nr_client = nr_client
        self.sampling_rules = {}
    
    def calculate_optimal_sampling(self, service_name):
        """サービス特性に基づく最適サンプリング率計算"""
        
        # 過去のデータボリューム分析
        volume_query = f"""
        SELECT count(*)
        FROM Transaction
        WHERE appName = '{service_name}'
        SINCE 1 day ago
        FACET hour(timestamp)
        """
        
        volume_data = self.nr_client.execute_nrql(volume_query)
        
        # エラー率分析
        error_query = f"""
        SELECT filter(count(*), WHERE error IS true) / count(*)
        FROM Transaction
        WHERE appName = '{service_name}'
        SINCE 1 day ago
        """
        
        error_rate = self.nr_client.execute_nrql(error_query)
        
        # 動的サンプリング率決定
        if error_rate > 0.05:  # 5%以上のエラー率
            return 1.0  # 100%サンプリング
        elif volume_data['peak_volume'] > 10000:  # 高ボリューム
            return 0.1  # 10%サンプリング
        else:
            return 0.5  # 50%サンプリング
    
    def implement_sampling_rules(self, service_name, sampling_rate):
        """サンプリングルールの実装"""
        config = {
            "application_settings": {
                "transaction_tracer": {
                    "record_sql": "raw" if sampling_rate >= 0.5 else "off",
                    "explain_enabled": sampling_rate >= 0.8
                },
                "error_collector": {
                    "capture_events": True,
                    "max_samples_stored": int(1000 * sampling_rate)
                },
                "attributes": {
                    "enabled": sampling_rate >= 0.3
                }
            }
        }
        
        return config

# 自動コスト最適化システム
class CostOptimizer:
    def __init__(self, nr_client, budget_limit):
        self.nr_client = nr_client
        self.budget_limit = budget_limit
        
    def optimize_monthly_spending(self):
        """月次支出最適化"""
        current_usage = self.get_current_usage()
        projected_cost = self.project_monthly_cost(current_usage)
        
        if projected_cost > self.budget_limit:
            optimizations = self.generate_optimizations(current_usage)
            return self.implement_optimizations(optimizations)
    
    def generate_optimizations(self, usage_data):
        """最適化提案生成"""
        optimizations = []
        
        for source, usage in usage_data.items():
            # 低価値データソースの特定
            if usage['value_score'] < 0.1:
                optimizations.append({
                    "type": "reduce_sampling",
                    "source": source,
                    "current_rate": usage['sampling_rate'],
                    "recommended_rate": usage['sampling_rate'] * 0.5,
                    "estimated_savings": usage['cost'] * 0.5
                })
            
            # 重複データの除去
            if usage['redundancy_score'] > 0.3:
                optimizations.append({
                    "type": "eliminate_redundancy",
                    "source": source,
                    "estimated_savings": usage['cost'] * usage['redundancy_score']
                })
        
        return optimizations
    
    def implement_optimizations(self, optimizations):
        """最適化の実装"""
        results = []
        
        for optimization in optimizations:
            if optimization['type'] == 'reduce_sampling':
                result = self.update_sampling_rate(
                    optimization['source'],
                    optimization['recommended_rate']
                )
                results.append(result)
                
        return results

# ROI計算システム
class ROICalculator:
    def __init__(self, nr_client):
        self.nr_client = nr_client
        
    def calculate_annual_roi(self, period_months=12):
        """年次ROI計算"""
        
        # コスト計算
        costs = self.calculate_total_costs(period_months)
        
        # 節約効果計算
        savings = self.calculate_savings(period_months)
        
        # 生産性向上効果
        productivity_gains = self.calculate_productivity_gains(period_months)
        
        # ROI計算
        total_benefits = savings + productivity_gains
        roi_percentage = ((total_benefits - costs) / costs) * 100
        
        return {
            "costs": costs,
            "savings": savings,
            "productivity_gains": productivity_gains,
            "total_benefits": total_benefits,
            "roi_percentage": roi_percentage,
            "payback_period_months": costs / (total_benefits / 12)
        }
    
    def calculate_savings(self, period_months):
        """コスト削減効果計算"""
        
        # インフラコスト削減
        infra_savings = self.calculate_infrastructure_savings(period_months)
        
        # インシデント対応コスト削減
        incident_savings = self.calculate_incident_cost_reduction(period_months)
        
        # パフォーマンス最適化によるコスト削減
        performance_savings = self.calculate_performance_savings(period_months)
        
        return {
            "infrastructure": infra_savings,
            "incident_response": incident_savings,
            "performance_optimization": performance_savings,
            "total": infra_savings + incident_savings + performance_savings
        }
    
    def calculate_productivity_gains(self, period_months):
        """生産性向上効果計算"""
        
        # MTTR短縮による効果
        mttr_query = f"""
        SELECT 
          average(closeDuration) as avg_resolution_time
        FROM AlertViolation
        WHERE timestamp >= {period_months} months ago
        """
        
        mttr_data = self.nr_client.execute_nrql(mttr_query)
        
        # デプロイメント頻度向上効果
        deployment_query = f"""
        SELECT count(*)
        FROM Deployment
        WHERE timestamp >= {period_months} months ago
        TIMESERIES 1 month
        """
        
        deployment_data = self.nr_client.execute_nrql(deployment_query)
        
        # エンジニア工数削減効果計算
        time_saved_hours = self.calculate_time_savings(mttr_data, deployment_data)
        engineer_hourly_rate = 100  # $100/hour
        
        return time_saved_hours * engineer_hourly_rate

# 改善提案エンジン
class ImprovementEngine:
    def __init__(self, nr_client):
        self.nr_client = nr_client
        
    def generate_monthly_recommendations(self):
        """月次改善提案生成"""
        
        recommendations = []
        
        # 使用量分析
        usage_analysis = self.analyze_usage_patterns()
        
        # コスト効率分析
        cost_efficiency = self.analyze_cost_efficiency()
        
        # パフォーマンス分析
        performance_analysis = self.analyze_performance_opportunities()
        
        recommendations.extend(self.generate_usage_recommendations(usage_analysis))
        recommendations.extend(self.generate_cost_recommendations(cost_efficiency))
        recommendations.extend(self.generate_performance_recommendations(performance_analysis))
        
        return self.prioritize_recommendations(recommendations)
    
    def prioritize_recommendations(self, recommendations):
        """推奨事項の優先順位付け"""
        
        for rec in recommendations:
            # インパクトスコア計算
            rec['impact_score'] = (
                rec['cost_savings'] * 0.4 +
                rec['performance_improvement'] * 0.3 +
                rec['risk_reduction'] * 0.2 +
                rec['implementation_ease'] * 0.1
            )
        
        return sorted(recommendations, key=lambda x: x['impact_score'], reverse=True)