New Relicデータエクスポート方法 - APIとツール統合による外部連携 ​

import requests
import json
from datetime import datetime, timedelta

class NewRelicDataExporter:
    def __init__(self, api_key, account_id):
        self.api_key = api_key
        self.account_id = account_id
        self.base_url = 'https://api.newrelic.com/v2'
        self.headers = {
            'X-Api-Key': api_key,
            'Content-Type': 'application/json'
        }
    
    def get_application_metrics(self, app_id, metric_names, start_time=None, end_time=None):
        """アプリケーションメトリクスの取得"""
        if not start_time:
            start_time = datetime.now() - timedelta(hours=1)
        if not end_time:
            end_time = datetime.now()
        
        url = f"{self.base_url}/applications/{app_id}/metrics/data.json"
        params = {
            'names[]': metric_names,
            'from': start_time.isoformat(),
            'to': end_time.isoformat(),
            'summarize': 'true'
        }
        
        response = requests.get(url, headers=self.headers, params=params)
        return response.json()
    
    def export_to_csv(self, data, filename):
        """データをCSV形式でエクスポート"""
        import csv
        
        with open(filename, 'w', newline='') as csvfile:
            if data and 'metric_data' in data:
                fieldnames = ['timestamp', 'metric_name', 'value']
                writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
                writer.writeheader()
                
                for metric in data['metric_data']['metrics']:
                    metric_name = metric['name']
                    for datapoint in metric['timeslices']:
                        writer.writerow({
                            'timestamp': datapoint['from'],
                            'metric_name': metric_name,
                            'value': datapoint['values'].get('average_value', 0)
                        })

# 使用例
exporter = NewRelicDataExporter('YOUR_API_KEY', 'YOUR_ACCOUNT_ID')
metrics = exporter.get_application_metrics(
    app_id='123456',
    metric_names=['Apdex', 'HttpDispatcher', 'Errors/all']
)
exporter.export_to_csv(metrics, 'newrelic_metrics.csv')

import pandas as pd
import tableau_server_client as TSC
from newrelic_data_client import NewRelicClient

class TableauNewRelicIntegrator:
    def __init__(self, nr_api_key, tableau_server_url, tableau_token):
        self.nr_client = NewRelicClient(nr_api_key)
        self.tableau_server = tableau_server_url
        self.tableau_token = tableau_token
    
    def create_performance_dashboard_data(self):
        """パフォーマンスダッシュボード用データの準備"""
        
        # New Relic からデータ取得
        performance_data = self.nr_client.query_nrql("""
            SELECT average(duration) as avg_response_time,
                   percentile(duration, 95) as p95_response_time,
                   count(*) as request_count,
                   percentage(count(*), WHERE error IS true) as error_rate
            FROM Transaction
            FACET appName, DATE(timestamp)
            SINCE 30 days ago
        """)
        
        # Pandas DataFrame に変換
        df = pd.DataFrame(performance_data['results'])
        df['date'] = pd.to_datetime(df['facet.date'])
        df['application'] = df['facet.appName']
        
        # Tableau 用にデータ整形
        tableau_data = df[['date', 'application', 'avg_response_time', 
                          'p95_response_time', 'request_count', 'error_rate']]
        
        return tableau_data
    
    def publish_to_tableau(self, data, datasource_name):
        """Tableau Server へのデータ公開"""
        server = TSC.Server(self.tableau_server, use_server_version=True)
        
        with server.auth.sign_in(self.tableau_token):
            # データソースとして公開
            datasource = TSC.DatasourceItem()
            datasource.name = datasource_name
            
            # CSV として一時保存してアップロード
            csv_path = f'/tmp/{datasource_name}.csv'
            data.to_csv(csv_path, index=False)
            
            # Tableau Server に公開
            server.datasources.publish(datasource, csv_path, 'Overwrite')

class PowerBINewRelicConnector:
    def __init__(self, nr_api_key, powerbi_settings):
        self.nr_client = NewRelicClient(nr_api_key)
        self.powerbi_settings = powerbi_settings
    
    def create_business_metrics_report(self):
        """ビジネスメトリクスレポートの生成"""
        
        # カスタムイベントからビジネスデータ取得
        business_data = self.nr_client.query_nrql("""
            SELECT count(*) as transaction_count,
                   sum(orderAmount) as total_revenue,
                   average(orderAmount) as avg_order_value,
                   uniqueCount(customerId) as unique_customers
            FROM PurchaseCompleted
            FACET DATE(timestamp), productCategory
            SINCE 90 days ago
        """)
        
        # アプリケーションパフォーマンスデータ
        performance_data = self.nr_client.query_nrql("""
            SELECT average(duration) as avg_response_time,
                   count(*) as total_requests
            FROM Transaction
            WHERE appName = 'ecommerce-api'
            FACET DATE(timestamp)
            SINCE 90 days ago
        """)
        
        # データ結合とPower BI 用フォーマット
        combined_data = self.merge_business_and_performance_data(
            business_data, performance_data
        )
        
        return combined_data
    
    def schedule_data_refresh(self):
        """定期的なデータ更新スケジュール"""
        # Power BI REST API を使用した自動更新設定
        pass

import asyncio
import aiohttp
from datetime import datetime, timedelta
import json

class BatchDataExporter:
    def __init__(self, api_key, account_id):
        self.api_key = api_key
        self.account_id = account_id
        self.session = None
    
    async def export_historical_data(self, start_date, end_date, chunk_hours=6):
        """大量ヒストリカルデータのバッチエクスポート"""
        
        async with aiohttp.ClientSession() as session:
            self.session = session
            
            current_date = start_date
            all_data = []
            
            while current_date < end_date:
                chunk_end = min(current_date + timedelta(hours=chunk_hours), end_date)
                
                # 並列でデータ取得
                tasks = [
                    self.fetch_application_data(current_date, chunk_end),
                    self.fetch_infrastructure_data(current_date, chunk_end),
                    self.fetch_browser_data(current_date, chunk_end)
                ]
                
                chunk_results = await asyncio.gather(*tasks, return_exceptions=True)
                
                # エラーハンドリング
                for result in chunk_results:
                    if isinstance(result, Exception):
                        print(f"Error processing chunk {current_date}: {result}")
                    else:
                        all_data.extend(result)
                
                current_date = chunk_end
                
                # レート制限回避のための待機
                await asyncio.sleep(1)
            
            return all_data
    
    async def fetch_application_data(self, start_time, end_time):
        """アプリケーションデータの取得"""
        query = f"""
        SELECT timestamp, appName, duration, error
        FROM Transaction
        WHERE timestamp >= '{start_time.isoformat()}'
        AND timestamp < '{end_time.isoformat()}'
        LIMIT MAX
        """
        
        return await self.execute_nrql_query(query)
    
    async def execute_nrql_query(self, nrql_query):
        """NRQL クエリの実行"""
        graphql_query = {
            "query": f"""
            {{
              actor {{
                account(id: {self.account_id}) {{
                  nrql(query: "{nrql_query}") {{
                    results
                  }}
                }}
              }}
            }}
            """
        }
        
        headers = {
            'Content-Type': 'application/json',
            'API-Key': self.api_key
        }
        
        async with self.session.post(
            'https://api.newrelic.com/graphql',
            json=graphql_query,
            headers=headers
        ) as response:
            data = await response.json()
            return data['data']['actor']['account']['nrql']['results']

# 使用例
async def main():
    exporter = BatchDataExporter('YOUR_API_KEY', 'YOUR_ACCOUNT_ID')
    
    start_date = datetime(2024, 1, 1)
    end_date = datetime(2024, 1, 31)
    
    historical_data = await exporter.export_historical_data(start_date, end_date)
    
    # データを JSON ファイルに保存
    with open('historical_data.json', 'w') as f:
        json.dump(historical_data, f, indent=2, default=str)

# 実行
asyncio.run(main())