New Relic APM入門 第5.2章 - 分散トレーシングとサービスマップ ​

# スパンの詳細構造例
checkout_spans = [
    {
        "span_id": "span-001",
        "parent_id": None,  # ルートスパン
        "service": "api-gateway",
        "operation": "POST /checkout",
        "start_time": "2025-07-21T10:00:00.000Z",
        "end_time": "2025-07-21T10:00:03.200Z",
        "duration": "3200ms",
        "tags": {
            "http.method": "POST",
            "http.url": "/api/checkout",
            "user.id": "user123",
            "request.size": "1.2KB"
        }
    },
    {
        "span_id": "span-002", 
        "parent_id": "span-001",  # 親スパン
        "service": "user-service",
        "operation": "authenticate_user",
        "start_time": "2025-07-21T10:00:00.050Z",
        "end_time": "2025-07-21T10:00:00.170Z",
        "duration": "120ms",
        "tags": {
            "auth.method": "JWT",
            "user.role": "premium",
            "cache.hit": True
        }
    },
    {
        "span_id": "span-003",
        "parent_id": "span-001", 
        "service": "inventory-service",
        "operation": "check_stock",
        "start_time": "2025-07-21T10:00:00.200Z", 
        "end_time": "2025-07-21T10:00:02.700Z",
        "duration": "2500ms",  # 問題のスパン！
        "tags": {
            "inventory.items": "3",
            "db.query.slow": True,  # 遅いクエリ検出
            "warehouse.api.timeout": True
        },
        "logs": [
            {
                "timestamp": "2025-07-21T10:00:02.600Z",
                "level": "WARNING", 
                "message": "Warehouse API response time > 2s"
            }
        ]
    }
]

# Python gRPC: サーバー側
import grpc
import newrelic.agent

class InventoryService(inventory_pb2_grpc.InventoryServiceServicer):
    
    @newrelic.agent.function_trace('inventory.check_stock')
    def CheckStock(self, request, context):
        # gRPCメタデータからトレーシング情報抽出
        metadata = dict(context.invocation_metadata())
        
        # New Relicトレースコンテキスト復元
        newrelic.agent.add_custom_attribute('grpc.method', 'CheckStock')
        newrelic.agent.add_custom_attribute('request.items', len(request.items))
        
        try:
            # データベース処理（自動監視）
            availability = self.check_database_inventory(request.items)
            
            # 外部API呼び出し（ウェアハウス確認）
            with newrelic.agent.FunctionTrace('warehouse.api_call'):
                warehouse_stock = self.call_warehouse_api(request.items)
            
            result = inventory_pb2.StockResponse(
                available=availability and warehouse_stock,
                message="Stock check completed"
            )
            
            newrelic.agent.record_custom_metric('Custom/Inventory/CheckSuccess', 1)
            return result
            
        except Exception as e:
            newrelic.agent.notice_error()
            newrelic.agent.record_custom_metric('Custom/Inventory/CheckError', 1)
            
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f'Inventory check failed: {str(e)}')
            return inventory_pb2.StockResponse()

# Python gRPC: クライアント側  
def call_inventory_service(items):
    
    with grpc.insecure_channel('inventory-service:50051') as channel:
        stub = inventory_pb2_grpc.InventoryServiceStub(channel)
        
        # New Relicトレーシングメタデータ作成
        metadata = []
        
        # gRPC呼び出し（トレーシング情報自動伝播）
        with newrelic.agent.FunctionTrace('grpc.inventory_check'):
            response = stub.CheckStock(
                inventory_pb2.StockRequest(items=items),
                metadata=metadata
            )
        
        return response.available

# Django settings.py
import newrelic.agent

# New Relic設定
NEW_RELIC_CONFIG = {
    'distributed_tracing.enabled': True,
    'transaction_tracer.enabled': True,
    'transaction_tracer.record_sql': 'obfuscated',
    'span_events.enabled': True,
    'span_events.max_samples_stored': 2000
}

# wsgi.py
import newrelic.agent
newrelic.agent.initialize('/path/to/newrelic.ini')

# views.py - チェックアウトAPI
import requests
import newrelic.agent
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
@newrelic.agent.function_trace('checkout_api')
def checkout_view(request):
    """チェックアウト処理（分散トレーシング統合）"""
    
    # ビジネスコンテキスト追加
    newrelic.agent.add_custom_attribute('user.id', request.POST.get('user_id'))
    newrelic.agent.add_custom_attribute('checkout.amount', request.POST.get('amount'))
    
    try:
        # ステップ1：認証サービス呼び出し
        with newrelic.agent.FunctionTrace('auth_service_call'):
            auth_response = call_auth_service(request.META.get('HTTP_AUTHORIZATION'))
            
            if not auth_response['valid']:
                raise AuthenticationError("Invalid token")
        
        # ステップ2：在庫サービス呼び出し  
        with newrelic.agent.FunctionTrace('inventory_service_call'):
            inventory_response = call_inventory_service(request.POST.get('items'))
            
            if not inventory_response['available']:
                raise InventoryError("Items not available")
        
        # ステップ3：決済サービス呼び出し
        with newrelic.agent.FunctionTrace('payment_service_call'):
            payment_response = call_payment_service({
                'amount': request.POST.get('amount'),
                'method': request.POST.get('payment_method'),
                'user_id': auth_response['user_id']
            })
        
        # ステップ4：注文作成
        with newrelic.agent.FunctionTrace('order_creation'):
            order = create_order_record(
                user_id=auth_response['user_id'],
                items=request.POST.get('items'),
                payment_id=payment_response['transaction_id']
            )
        
        # 成功メトリクス
        newrelic.agent.record_custom_metric('Custom/Checkout/Success', 1)
        
        return JsonResponse({
            'success': True,
            'order_id': order.id,
            'trace_id': newrelic.agent.current_trace_id()  # デバッグ用
        })
        
    except Exception as e:
        # エラー詳細（トレース自動関連付け）
        newrelic.agent.notice_error(attributes={
            'error.step': getattr(e, 'step', 'unknown'),
            'error.service': getattr(e, 'service', 'unknown')
        })
        
        return JsonResponse({
            'error': str(e),
            'trace_id': newrelic.agent.current_trace_id()
        }, status=500)

def call_auth_service(auth_header):
    """認証サービス呼び出し（詳細監視）"""
    
    newrelic.agent.add_custom_attribute('auth.method', 'jwt_validation')
    
    try:
        response = requests.post('http://auth-service/validate', 
            headers={'Authorization': auth_header},
            timeout=5
        )
        
        result = response.json()
        
        # 認証結果属性
        newrelic.agent.add_custom_attribute('auth.user_id', result.get('user_id'))
        newrelic.agent.add_custom_attribute('auth.user_role', result.get('role'))
        
        return result
        
    except requests.exceptions.Timeout:
        error = AuthenticationError("Auth service timeout")
        error.service = 'auth-service'
        raise error
    except Exception as e:
        error = AuthenticationError(f"Auth service error: {str(e)}")
        error.service = 'auth-service' 
        raise error

def call_inventory_service(items_json):
    """在庫サービス呼び出し（詳細監視）"""
    
    items = json.loads(items_json)
    newrelic.agent.add_custom_attribute('inventory.items_count', len(items))
    
    try:
        response = requests.post('http://inventory-service/check', 
            json={'items': items},
            timeout=10
        )
        
        result = response.json()
        
        # 在庫確認結果
        newrelic.agent.add_custom_attribute('inventory.available', result.get('available'))
        newrelic.agent.add_custom_attribute('inventory.reserved_count', result.get('reserved_count'))
        
        return result
        
    except requests.exceptions.Timeout:
        error = InventoryError("Inventory service timeout")
        error.service = 'inventory-service'
        raise error

# requirements.txt
opentelemetry-api==1.21.0
opentelemetry-sdk==1.21.0
opentelemetry-instrumentation-flask==0.42b0
opentelemetry-instrumentation-requests==0.42b0
opentelemetry-instrumentation-sqlalchemy==0.42b0
opentelemetry-exporter-otlp==1.21.0

# otel_config.py - OpenTelemetry設定
import os
from opentelemetry import trace, metrics
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

def initialize_otel():
    """OpenTelemetry + New Relic初期化"""
    
    # リソース設定（サービス識別情報）
    resource = Resource.create({
        "service.name": "ecommerce-checkout",
        "service.version": "1.2.3", 
        "deployment.environment": os.getenv("ENVIRONMENT", "development"),
        "service.namespace": "ecommerce"
    })
    
    # New Relic エクスポーター設定
    otlp_span_exporter = OTLPSpanExporter(
        endpoint="https://otlp.nr-data.net:4317",
        headers={"api-key": os.getenv("NEW_RELIC_LICENSE_KEY")}
    )
    
    otlp_metric_exporter = OTLPMetricExporter(
        endpoint="https://otlp.nr-data.net:4317", 
        headers={"api-key": os.getenv("NEW_RELIC_LICENSE_KEY")}
    )
    
    # トレース設定
    trace.set_tracer_provider(TracerProvider(resource=resource))
    span_processor = BatchSpanProcessor(otlp_span_exporter)
    trace.get_tracer_provider().add_span_processor(span_processor)
    
    # メトリクス設定
    metric_reader = PeriodicExportingMetricReader(otlp_metric_exporter, export_interval_millis=30000)
    metrics.set_meter_provider(MeterProvider(resource=resource, metric_readers=[metric_reader]))
    
    # 自動計装有効化
    FlaskInstrumentor().instrument()           # Flask自動計装
    RequestsInstrumentor().instrument()        # Requests自動計装  
    SQLAlchemyInstrumentor().instrument()      # SQLAlchemy自動計装
    
    print("OpenTelemetry initialized with New Relic")

# app.py - Flask アプリケーション
from flask import Flask, request, jsonify
import requests
from opentelemetry import trace, metrics
from opentelemetry.trace import Status, StatusCode
from otel_config import initialize_otel

# OpenTelemetry初期化
initialize_otel()

app = Flask(__name__)

# トレーサー取得
tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)

# カスタムメトリクス定義
checkout_counter = meter.create_counter(
    "checkout_requests_total",
    description="Total checkout requests"
)
checkout_duration = meter.create_histogram(
    "checkout_duration_seconds", 
    description="Checkout processing duration"
)

@app.route('/api/checkout', methods=['POST'])
def checkout():
    """チェックアウトAPI（OpenTelemetry統合）"""
    
    # メインスパン開始
    with tracer.start_as_current_span("checkout_process") as span:
        
        # ビジネス属性追加
        span.set_attributes({
            "checkout.user_id": request.json.get('user_id'),
            "checkout.amount": request.json.get('amount'),
            "checkout.items_count": len(request.json.get('items', [])),
            "http.method": request.method,
            "http.route": request.endpoint
        })
        
        try:
            # 認証処理
            with tracer.start_as_current_span("user_authentication") as auth_span:
                auth_result = authenticate_user(request.headers.get('Authorization'))
                
                auth_span.set_attributes({
                    "auth.user_id": auth_result.get('user_id'),
                    "auth.user_role": auth_result.get('role'),
                    "auth.success": True
                })
            
            # 在庫確認処理
            with tracer.start_as_current_span("inventory_check") as inventory_span:
                inventory_result = check_inventory(request.json.get('items'))
                
                inventory_span.set_attributes({
                    "inventory.items_checked": len(request.json.get('items', [])),
                    "inventory.all_available": inventory_result.get('available'),
                    "inventory.processing_time_ms": inventory_result.get('processing_time')
                })
                
                if not inventory_result.get('available'):
                    raise InventoryNotAvailableError("Items not available")
            
            # 決済処理 
            with tracer.start_as_current_span("payment_processing") as payment_span:
                payment_result = process_payment({
                    'amount': request.json.get('amount'),
                    'method': request.json.get('payment_method'),
                    'user_id': auth_result.get('user_id')
                })
                
                payment_span.set_attributes({
                    "payment.amount": request.json.get('amount'),
                    "payment.method": request.json.get('payment_method'),
                    "payment.transaction_id": payment_result.get('transaction_id'),
                    "payment.success": True
                })
            
            # 注文作成
            with tracer.start_as_current_span("order_creation") as order_span:
                order = create_order({
                    'user_id': auth_result.get('user_id'),
                    'items': request.json.get('items'),
                    'payment_id': payment_result.get('transaction_id')
                })
                
                order_span.set_attributes({
                    "order.id": order.get('id'),
                    "order.creation_success": True
                })
            
            # 成功メトリクス
            checkout_counter.add(1, {
                "status": "success",
                "payment_method": request.json.get('payment_method')
            })
            
            span.set_attributes({
                "checkout.success": True,
                "checkout.order_id": order.get('id')
            })
            
            span.add_event("checkout_completed", {
                "order_id": order.get('id'),
                "completion_time": str(datetime.now())
            })
            
            span.set_status(Status(StatusCode.OK))
            
            return jsonify({
                "success": True,
                "order_id": order.get('id'),
                "trace_id": format(span.get_span_context().trace_id, '032x')
            })
            
        except Exception as e:
            # エラー処理
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.set_attributes({
                "checkout.success": False,
                "error.type": type(e).__name__
            })
            
            # エラーメトリクス
            checkout_counter.add(1, {
                "status": "error", 
                "error_type": type(e).__name__
            })
            
            return jsonify({
                "error": str(e),
                "trace_id": format(span.get_span_context().trace_id, '032x')
            }), 500

def authenticate_user(auth_header):
    """認証サービス呼び出し（自動計装）"""
    response = requests.post('http://auth-service/validate', 
        headers={'Authorization': auth_header},
        timeout=5
    )
    
    if not response.ok:
        raise AuthenticationError("Authentication failed")
    
    return response.json()

def check_inventory(items):
    """在庫確認サービス呼び出し（自動計装）"""
    response = requests.post('http://inventory-service/check',
        json={'items': items},
        timeout=10
    )
    
    if not response.ok:
        raise InventoryServiceError("Inventory check failed")
    
    return response.json()

# サービスマップを活用した障害影響分析
def analyze_external_service_outage(external_service_name):
    """外部サービス障害の影響分析"""
    
    impact_analysis = {
        "affected_external_service": external_service_name,
        "direct_dependencies": [],      # 直接影響を受けるサービス
        "indirect_dependencies": [],    # 間接的に影響を受けるサービス 
        "business_impact": {},         # ビジネス影響
        "mitigation_actions": []       # 対処法
    }
    
    # Stripeの例
    if external_service_name == "Stripe Payment API":
        impact_analysis.update({
            "direct_dependencies": [
                {
                    "service": "payment-service",
                    "impact": "critical",
                    "error_rate": "100%",           # 全決済が失敗
                    "fallback_available": False
                }
            ],
            "indirect_dependencies": [
                {
                    "service": "checkout-service", 
                    "impact": "critical",
                    "error_rate": "85%",            # チェックアウトの85%が失敗
                    "degraded_functions": ["payment_processing"]
                },
                {
                    "service": "order-service",
                    "impact": "medium",
                    "error_rate": "25%",            # 注文の25%が未完了状態
                    "degraded_functions": ["order_completion"]
                },
                {
                    "service": "user-service",
                    "impact": "low", 
                    "error_rate": "5%",             # ユーザー体験の軽微な悪化
                    "degraded_functions": ["purchase_history_update"]
                }
            ],
            "business_impact": {
                "revenue_loss_per_hour": "$12,000",
                "affected_customers": 450,
                "conversion_rate_drop": "78%"
            },
            "mitigation_actions": [
                "backup_payment_provider を有効化（PayPal）",
                "checkout プロセスでの明確なエラーメッセージ表示", 
                "顧客サポートチームへの状況連絡",
                "ソーシャルメディアでの障害告知"
            ]
        })
    
    return impact_analysis

# 使用例：リアルタイム監視
def monitor_service_map_health():
    """サービスマップの健全性リアルタイム監視"""
    
    service_health = get_current_service_map_status()
    
    for service in service_health:
        # 異常検出
        if service['error_rate'] > 5.0:  # 5%以上のエラー率
            alert_data = {
                "severity": "critical",
                "service": service['name'],
                "issue": f"High error rate: {service['error_rate']}%",
                "affected_dependencies": get_downstream_services(service['name']),
                "business_impact": calculate_business_impact(service['name'])
            }
            
            send_alert_to_oncall(alert_data)
            trigger_auto_remediation(service['name'])
        
        elif service['response_time'] > service['sla_threshold']:
            alert_data = {
                "severity": "warning", 
                "service": service['name'],
                "issue": f"SLA violation: {service['response_time']}ms > {service['sla_threshold']}ms",
                "trend": get_performance_trend(service['name'], hours=6)
            }
            
            send_alert_to_team(alert_data)