Datadog入門 第8部 - 統合・自動化とAPIによる運用効率化の実践完全ガイド ​

# インテグレーション優先度マトリクス評価
def evaluate_integration_priority(service, business_impact, technical_complexity, maintenance_cost):
    """
    インテグレーション優先度を定量的に評価
    
    Args:
        service: 対象サービス名
        business_impact: ビジネス影響度 (1-10)
        technical_complexity: 技術的複雑さ (1-10)
        maintenance_cost: 保守コスト (1-10)
    
    Returns:
        priority_score: 優先度スコア
        recommendation: 実装推奨度
    """
    
    # 加重スコア計算
    priority_score = (
        business_impact * 0.5 +          # ビジネス価値重視
        (11 - technical_complexity) * 0.3 +  # 実装容易性
        (11 - maintenance_cost) * 0.2        # 運用効率性
    )
    
    # 推奨レベル判定
    if priority_score >= 8.0:
        recommendation = "即座に実装"
        tier = "Tier 1 - Critical"
    elif priority_score >= 6.0:
        recommendation = "短期実装"
        tier = "Tier 2 - Important"
    elif priority_score >= 4.0:
        recommendation = "中期検討"
        tier = "Tier 3 - Beneficial"
    else:
        recommendation = "長期検討"
        tier = "Tier 4 - Optional"
    
    return {
        'service': service,
        'priority_score': round(priority_score, 2),
        'recommendation': recommendation,
        'tier': tier,
        'business_impact': business_impact,
        'technical_complexity': technical_complexity,
        'maintenance_cost': maintenance_cost
    }

# 実装例：代表的なサービス評価
services_evaluation = [
    evaluate_integration_priority("AWS EC2", 9, 2, 2),
    evaluate_integration_priority("PostgreSQL", 8, 3, 3),
    evaluate_integration_priority("Redis", 7, 2, 2),
    evaluate_integration_priority("Kubernetes", 9, 5, 4),
    evaluate_integration_priority("Jenkins", 6, 4, 3),
    evaluate_integration_priority("Kafka", 7, 6, 5)
]

for eval_result in services_evaluation:
    print(f"{eval_result['service']}: {eval_result['tier']} - {eval_result['recommendation']}")

# GitHub Actions メトリクス収集スクリプト
import requests
import time
from datadog import initialize, statsd

# Datadog初期化
options = {
    'api_key': '<DD_API_KEY>',
    'app_key': '<DD_APP_KEY>'
}
initialize(**options)

def collect_github_actions_metrics(repo_owner, repo_name, github_token):
    """
    GitHub Actions ワークフロー実行メトリクスを収集
    """
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github.v3+json'
    }
    
    # ワークフロー実行履歴取得
    url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/actions/runs'
    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        runs = response.json()['workflow_runs']
        
        # 成功率計算
        total_runs = len(runs)
        successful_runs = len([r for r in runs if r['conclusion'] == 'success'])
        success_rate = (successful_runs / total_runs) * 100 if total_runs > 0 else 0
        
        # メトリクス送信
        statsd.gauge('github.actions.success_rate', success_rate, 
                    tags=[f'repo:{repo_owner}/{repo_name}'])
        
        # 実行時間分析
        for run in runs[:10]:  # 最新10件
            if run['conclusion'] and run['created_at'] and run['updated_at']:
                start_time = datetime.fromisoformat(run['created_at'].replace('Z', '+00:00'))
                end_time = datetime.fromisoformat(run['updated_at'].replace('Z', '+00:00'))
                duration_seconds = (end_time - start_time).total_seconds()
                
                statsd.histogram('github.actions.duration', duration_seconds,
                               tags=[
                                   f'repo:{repo_owner}/{repo_name}',
                                   f'workflow:{run["name"]}',
                                   f'status:{run["conclusion"]}'
                               ])

# 定期実行
if __name__ == "__main__":
    repos = [
        ('your-org', 'backend-api'),
        ('your-org', 'frontend-app'),
        ('your-org', 'infrastructure')
    ]
    
    while True:
        for owner, name in repos:
            collect_github_actions_metrics(owner, name, '<GITHUB_TOKEN>')
        time.sleep(300)  # 5分間隔

# 高度なDatadog APIクライアント実装
import asyncio
import aiohttp
import time
from typing import List, Dict, Optional, Any
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.metrics_api import MetricsApi
from datadog_api_client.v1.api.dashboards_api import DashboardsApi
from datadog_api_client.v1.api.monitors_api import MonitorsApi
from datadog_api_client.v1.api.logs_api import LogsApi

class DatadogAPIManager:
    """
    高度なDatadog API管理クラス
    バッチ処理、レート制限、エラーハンドリングを含む
    """
    
    def __init__(self, api_key: str, app_key: str, site: str = "datadoghq.com"):
        self.configuration = Configuration()
        self.configuration.api_key["apiKeyAuth"] = api_key
        self.configuration.api_key["appKeyAuth"] = app_key
        self.configuration.server_variables["site"] = site
        
        self.api_client = ApiClient(self.configuration)
        self.metrics_api = MetricsApi(self.api_client)
        self.dashboards_api = DashboardsApi(self.api_client)
        self.monitors_api = MonitorsApi(self.api_client)
        self.logs_api = LogsApi(self.api_client)
        
        # レート制限管理
        self.rate_limit_window = 3600  # 1時間
        self.api_calls_count = 0
        self.window_start = time.time()
    
    def _check_rate_limit(self, max_calls_per_hour: int = 3000):
        """
        API レート制限チェック
        """
        current_time = time.time()
        if current_time - self.window_start > self.rate_limit_window:
            self.api_calls_count = 0
            self.window_start = current_time
        
        if self.api_calls_count >= max_calls_per_hour:
            sleep_time = self.rate_limit_window - (current_time - self.window_start)
            print(f"Rate limit reached. Sleeping for {sleep_time:.2f} seconds")
            time.sleep(sleep_time)
            self.api_calls_count = 0
            self.window_start = time.time()
        
        self.api_calls_count += 1
    
    async def submit_metrics_batch(self, metrics_data: List[Dict], batch_size: int = 100):
        """
        メトリクス一括送信（非同期）
        """
        for i in range(0, len(metrics_data), batch_size):
            batch = metrics_data[i:i + batch_size]
            self._check_rate_limit()
            
            try:
                series = []
                for metric in batch:
                    series.append({
                        'metric': metric['metric_name'],
                        'points': [[int(time.time()), metric['value']]],
                        'tags': metric.get('tags', []),
                        'host': metric.get('host', ''),
                        'type': metric.get('type', 'gauge')
                    })
                
                response = self.metrics_api.submit_metrics({'series': series})
                print(f"Submitted {len(batch)} metrics successfully")
                
            except Exception as e:
                print(f"Error submitting metrics batch: {str(e)}")
                await asyncio.sleep(5)  # エラー時は少し待機
    
    def create_dynamic_dashboard(self, service_name: str, metrics_config: Dict) -> str:
        """
        サービス固有の動的ダッシュボード作成
        """
        self._check_rate_limit()
        
        dashboard_config = {
            "title": f"{service_name} Service Dashboard",
            "description": f"Auto-generated dashboard for {service_name}",
            "widgets": [],
            "layout_type": "ordered",
            "is_shared": False,
            "tags": [f"service:{service_name}", "auto-generated"]
        }
        
        # メトリクスウィジェット自動生成
        for category, metrics in metrics_config.items():
            if category == "infrastructure":
                # インフラメトリクス用ウィジェット
                widget = {
                    "definition": {
                        "type": "timeseries",
                        "title": f"{service_name} Infrastructure Metrics",
                        "requests": [
                            {
                                "q": f"avg:system.cpu.user{{service:{service_name}}}",
                                "display_type": "line",
                                "style": {"palette": "dog_classic", "line_type": "solid", "line_width": "normal"}
                            },
                            {
                                "q": f"avg:system.mem.pct_usable{{service:{service_name}}}",
                                "display_type": "line",
                                "style": {"palette": "dog_classic", "line_type": "solid", "line_width": "normal"}
                            }
                        ],
                        "yaxis": {"scale": "linear", "min": "auto", "max": "auto"},
                        "time": {"live_span": "1h"}
                    }
                }
                dashboard_config["widgets"].append(widget)
            
            elif category == "application":
                # アプリケーションメトリクス用ウィジェット
                widget = {
                    "definition": {
                        "type": "timeseries", 
                        "title": f"{service_name} Application Performance",
                        "requests": [
                            {
                                "q": f"avg:trace.web.request.duration{{service:{service_name}}}",
                                "display_type": "line"
                            }
                        ],
                        "time": {"live_span": "4h"}
                    }
                }
                dashboard_config["widgets"].append(widget)
        
        try:
            response = self.dashboards_api.create_dashboard(dashboard_config)
            dashboard_id = response.id
            print(f"Created dashboard for {service_name}: {dashboard_id}")
            return dashboard_id
            
        except Exception as e:
            print(f"Error creating dashboard: {str(e)}")
            return None
    
    def create_intelligent_monitor(self, monitor_config: Dict) -> str:
        """
        機械学習ベースのインテリジェントモニター作成
        """
        self._check_rate_limit()
        
        # 異常検知モニター設定
        if monitor_config.get('use_anomaly_detection', False):
            query = f"avg(last_4h):anomalies(avg:trace.web.request.duration{{service:{monitor_config['service']}}}, 'basic', 2, direction='both', alert_window='last_15m', interval=60, count_default_zero='true') >= 1"
        else:
            query = monitor_config['query']
        
        monitor_data = {
            "name": monitor_config['name'],
            "type": "metric alert",
            "query": query,
            "message": monitor_config.get('message', f"Alert for {monitor_config['service']}"),
            "tags": monitor_config.get('tags', []),
            "options": {
                "thresholds": monitor_config.get('thresholds', {"critical": 1}),
                "notify_audit": False,
                "require_full_window": True,
                "notify_no_data": True,
                "no_data_timeframe": 20,
                "include_tags": True
            }
        }
        
        try:
            response = self.monitors_api.create_monitor(monitor_data)
            monitor_id = response.id
            print(f"Created monitor: {monitor_config['name']} (ID: {monitor_id})")
            return monitor_id
            
        except Exception as e:
            print(f"Error creating monitor: {str(e)}")
            return None

# 使用例：サービス監視自動セットアップ
async def setup_service_monitoring(service_name: str, service_config: Dict):
    """
    新しいサービスの監視を自動セットアップ
    """
    api_manager = DatadogAPIManager(
        api_key="<DD_API_KEY>",
        app_key="<DD_APP_KEY>"
    )
    
    # 1. ダッシュボード作成
    dashboard_id = api_manager.create_dynamic_dashboard(
        service_name=service_name,
        metrics_config=service_config['metrics']
    )
    
    # 2. 基本監視モニター作成
    monitors = []
    for monitor_config in service_config['monitors']:
        monitor_config['service'] = service_name
        monitor_id = api_manager.create_intelligent_monitor(monitor_config)
        if monitor_id:
            monitors.append(monitor_id)
    
    # 3. カスタムメトリクス投入
    custom_metrics = [
        {
            'metric_name': f'{service_name}.startup.time',
            'value': service_config.get('startup_time', 0),
            'tags': [f'service:{service_name}', 'environment:production']
        },
        {
            'metric_name': f'{service_name}.health.score',
            'value': 100,  # 初期健全性スコア
            'tags': [f'service:{service_name}', 'environment:production']
        }
    ]
    
    await api_manager.submit_metrics_batch(custom_metrics)
    
    return {
        'dashboard_id': dashboard_id,
        'monitor_ids': monitors,
        'service_name': service_name
    }

# chatops/slack_datadog_integration.py
import asyncio
import json
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.dashboards_api import DashboardsApi
from datadog_api_client.v1.api.monitors_api import MonitorsApi

# Slack アプリ初期化
app = AsyncApp(token="<SLACK_BOT_TOKEN>")

# Datadog API クライアント
config = Configuration()
config.api_key["apiKeyAuth"] = "<DD_API_KEY>"
config.api_key["appKeyAuth"] = "<DD_APP_KEY>"
api_client = ApiClient(config)
dashboards_api = DashboardsApi(api_client)
monitors_api = MonitorsApi(api_client)

@app.command("/datadog-status")
async def handle_datadog_status(ack, respond, command):
    """
    Datadog システム状態をSlackで確認
    """
    await ack()
    
    try:
        # サービス別監視状態取得
        monitors = monitors_api.list_monitors(tags="team:platform")
        
        status_summary = {
            "OK": 0,
            "Alert": 0,
            "Warn": 0,
            "No Data": 0
        }
        
        service_alerts = []
        for monitor in monitors.data:
            status = monitor.overall_state
            status_summary[status] = status_summary.get(status, 0) + 1
            
            if status in ["Alert", "Warn"]:
                service_alerts.append({
                    "name": monitor.name,
                    "status": status,
                    "url": f"https://app.datadoghq.com/monitors/{monitor.id}"
                })
        
        # Slack レスポンス構築
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "🔍 Datadog System Status"
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"✅ *OK:* {status_summary.get('OK', 0)}"
                    },
                    {
                        "type": "mrkdwn", 
                        "text": f"⚠️ *Warning:* {status_summary.get('Warn', 0)}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"🚨 *Alert:* {status_summary.get('Alert', 0)}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"❓ *No Data:* {status_summary.get('No Data', 0)}"
                    }
                ]
            }
        ]
        
        # アラート詳細追加
        if service_alerts:
            blocks.append({
                "type": "divider"
            })
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": "*🚨 Active Alerts:*"
                }
            })
            
            for alert in service_alerts[:5]:  # 最大5件表示
                blocks.append({
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"• <{alert['url']}|{alert['name']}> - {alert['status']}"
                    }
                })
        
        await respond(blocks=blocks)
        
    except Exception as e:
        await respond(f"❌ Error fetching Datadog status: {str(e)}")

@app.command("/datadog-mute")
async def handle_mute_monitor(ack, respond, command):
    """
    監視アラートの一時停止
    """
    await ack()
    
    try:
        # コマンド引数解析
        args = command['text'].split()
        if len(args) < 2:
            await respond("Usage: `/datadog-mute <monitor_id> <duration_minutes> [reason]`")
            return
        
        monitor_id = int(args[0])
        duration_minutes = int(args[1])
        reason = " ".join(args[2:]) if len(args) > 2 else "Temporary maintenance"
        
        # モニター情報取得
        monitor = monitors_api.get_monitor(monitor_id)
        
        # ミュート設定
        mute_config = {
            "scope": "*",
            "end": int(time.time()) + (duration_minutes * 60),
            "message": f"Muted via Slack by {command['user_name']}: {reason}"
        }
        
        monitors_api.mute_monitor(monitor_id, mute_config)
        
        await respond(
            f"✅ Monitor `{monitor.name}` has been muted for {duration_minutes} minutes.\n"
            f"Reason: {reason}"
        )
        
    except ValueError:
        await respond("❌ Invalid monitor ID or duration. Please use numeric values.")
    except Exception as e:
        await respond(f"❌ Error muting monitor: {str(e)}")

@app.command("/datadog-deploy")
async def handle_deployment_notification(ack, respond, command):
    """
    デプロイメント通知とトラッキング
    """
    await ack()
    
    try:
        args = command['text'].split()
        if len(args) < 3:
            await respond("Usage: `/datadog-deploy <service> <version> <environment> [description]`")
            return
        
        service = args[0]
        version = args[1]
        environment = args[2]
        description = " ".join(args[3:]) if len(args) > 3 else f"Deployment of {service} {version}"
        
        # Datadog にデプロイメントイベント送信
        event_data = {
            "title": f"Deployment: {service} {version}",
            "text": description,
            "date_happened": int(time.time()),
            "priority": "normal",
            "tags": [
                f"service:{service}",
                f"version:{version}",
                f"environment:{environment}",
                "event_type:deployment",
                f"deployed_by:{command['user_name']}"
            ],
            "alert_type": "info"
        }
        
        # API経由でイベント送信（実装省略）
        
        # Slack 確認メッセージ
        blocks = [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"🚀 *Deployment Tracked*\n"
                           f"*Service:* {service}\n"
                           f"*Version:* {version}\n"
                           f"*Environment:* {environment}\n"
                           f"*Deployed by:* {command['user_name']}\n"
                           f"*Description:* {description}"
                }
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {
                            "type": "plain_text",
                            "text": "View Dashboard"
                        },
                        "url": f"https://app.datadoghq.com/dashboard/{service}",
                        "action_id": "view_dashboard"
                    },
                    {
                        "type": "button",
                        "text": {
                            "type": "plain_text",
                            "text": "View Monitors"
                        },
                        "url": f"https://app.datadoghq.com/monitors/manage?q=service%3A{service}",
                        "action_id": "view_monitors"
                    }
                ]
            }
        ]
        
        await respond(blocks=blocks)
        
    except Exception as e:
        await respond(f"❌ Error tracking deployment: {str(e)}")

# Datadog アラート Webhook ハンドラー
@app.event("webhook")
async def handle_datadog_webhook(webhook_data):
    """
    Datadog からの Webhook アラートを Slack に転送
    """
    try:
        # Webhook データ解析
        alert_data = json.loads(webhook_data)
        
        # アラート重要度に基づくSlack channel選択
        severity = alert_data.get('priority', 'normal')
        if severity == 'critical':
            channel = "#alerts-critical"
            emoji = "🚨"
        elif severity == 'warning':
            channel = "#alerts-warning" 
            emoji = "⚠️"
        else:
            channel = "#alerts-info"
            emoji = "ℹ️"
        
        # Slack メッセージ構築
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{emoji} Datadog Alert"
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Monitor:* {alert_data.get('alert_title', 'Unknown')}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Status:* {alert_data.get('alert_status', 'Unknown')}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Severity:* {severity.title()}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Time:* <!date^{int(time.time())}^{{date}} {{time}}|{time.strftime('%Y-%m-%d %H:%M:%S')}>"
                    }
                ]
            }
        ]
        
        # アラート詳細追加
        if alert_data.get('body'):
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Details:*\n{alert_data['body']}"
                }
            })
        
        # アクションボタン追加
        if alert_data.get('link'):
            blocks.append({
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {
                            "type": "plain_text",
                            "text": "View in Datadog"
                        },
                        "url": alert_data['link'],
                        "action_id": "view_alert"
                    },
                    {
                        "type": "button",
                        "text": {
                            "type": "plain_text",
                            "text": "Acknowledge"
                        },
                        "action_id": "acknowledge_alert",
                        "style": "primary"
                    }
                ]
            })
        
        await app.client.chat_postMessage(
            channel=channel,
            blocks=blocks
        )
        
    except Exception as e:
        print(f"Error handling Datadog webhook: {str(e)}")

# Slack アプリ起動
async def start_slack_app():
    """
    Slack アプリケーション起動
    """
    handler = AsyncSocketModeHandler(app, "<SLACK_APP_TOKEN>")
    await handler.start_async()

if __name__ == "__main__":
    asyncio.run(start_slack_app())