Monitoring

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# API 가용성 계산 예시
def calculate_availability(total_requests, failed_requests):
    if total_requests == 0:
        return 100.0  # 요청이 없으면 100% 가용성으로 간주
    
    availability = ((total_requests - failed_requests) / total_requests) * 100
    return round(availability, 2)  # 소수점 둘째 자리까지 반올림

# 9s 가용성 수준 정의
availability_levels = {
    "two_nines": 99.0,    # 가동 중지 시간: 월간 약 7.2시간
    "three_nines": 99.9,  # 가동 중지 시간: 월간 약 43.2분
    "four_nines": 99.99,  # 가동 중지 시간: 월간 약 4.32분
    "five_nines": 99.999  # 가동 중지 시간: 월간 약 26초
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// Express.js에서 응답 시간 모니터링 미들웨어 예시
const responseTimeMiddleware = (req, res, next) => {
  const start = process.hrtime();
  
  res.on('finish', () => {
    const end = process.hrtime(start);
    const duration = (end[0] * 1e9 + end[1]) / 1e6; // 밀리초로 변환
    
    // 메트릭 기록 (예: Prometheus, StatsD 등에 전송)
    recordMetric('api.response_time', {
      method: req.method,
      path: req.route ? req.route.path : req.path,
      status: res.statusCode,
      duration: duration
    });
    
    // 응답 시간이 임계값을 초과하면 알림
    if (duration > 500) {  // 500ms = 0.5초
      notifySlowResponse({
        method: req.method,
        path: req.path,
        duration: duration,
        timestamp: new Date()
      });
    }
  });
  
  next();
};

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# FastAPI에서 오류 모니터링 예시
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
import time
import logging
from collections import defaultdict

app = FastAPI()
error_counts = defaultdict(int)
error_logger = logging.getLogger("api_errors")

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    error_type = "validation_error"
    error_counts[error_type] += 1
    
    error_logger.error(
        f"Validation error: endpoint={request.url.path}, client_ip={request.client.host}, "
        f"error_detail={exc.errors()}"
    )
    
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"detail": exc.errors()}
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    error_type = type(exc).__name__
    error_counts[error_type] += 1
    
    error_logger.error(
        f"Unhandled exception: endpoint={request.url.path}, client_ip={request.client.host}, "
        f"error_type={error_type}, error_message={str(exc)}"
    )
    
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "Internal server error"}
    )

# 정기적으로 오류 통계 보고
@app.on_event("startup")
async def start_error_reporting():
    async def report_errors():
        while True:
            await asyncio.sleep(300)  # 5분마다 보고
            
            if error_counts:
                error_logger.info(f"Error summary (last 5 minutes): {dict(error_counts)}")
                # 메트릭 시스템에 데이터 전송
                for error_type, count in error_counts.items():
                    send_metric(f"api.error.{error_type}", count)
                
                # 카운터 초기화
                error_counts.clear()
    
    asyncio.create_task(report_errors())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

// Puppeteer를 사용한 합성 모니터링 예시
const puppeteer = require('puppeteer');
const axios = require('axios');

async function runSyntheticTest() {
  const startTime = Date.now();
  let success = false;
  let responseTime = 0;
  let errorMessage = null;
  
  try {
    // API 호출 시뮬레이션
    const response = await axios.get('https://api.example.com/health', {
      timeout: 5000  // 5초 타임아웃
    });
    
    responseTime = Date.now() - startTime;
    success = response.status === 200;
    
    // JWT 인증이 필요한 API 흐름 검증
    const token = await getAuthToken();
    const userResponse = await axios.get('https://api.example.com/users/me', {
      headers: { 'Authorization': `Bearer ${token}` }
    });
    
    // 응답 데이터 유효성 검사
    if (!userResponse.data.id) {
      throw new Error('User data is invalid');
    }
    
  } catch (error) {
    success = false;
    errorMessage = error.message;
    responseTime = Date.now() - startTime;
  }
  
  // 결과 보고
  reportTestResult({
    testName: 'API Health Check',
    success,
    responseTime,
    errorMessage,
    timestamp: new Date()
  });
}

// 스케줄링: 5분마다 테스트 실행
setInterval(runSyntheticTest, 5 * 60 * 1000);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

// 클라이언트 측 RUM 구현 예시 (브라우저)
class APIMonitor {
  constructor() {
    this.metrics = {
      calls: 0,
      errors: 0,
      totalResponseTime: 0
    };
    
    this.initialize();
  }
  
  initialize() {
    // 원본 fetch 메서드를 가로채기
    const originalFetch = window.fetch;
    
    window.fetch = async (url, options) => {
      // API 호출만 모니터링 (자체 도메인의 /api 경로)
      if (url.includes('/api')) {
        const start = performance.now();
        let error = false;
        
        try {
          const response = await originalFetch(url, options);
          if (!response.ok) {
            error = true;
          }
          
          const end = performance.now();
          this.recordMetrics(url, end - start, error);
          
          return response;
        } catch (err) {
          const end = performance.now();
          this.recordMetrics(url, end - start, true);
          throw err;
        }
      } else {
        // API가 아닌 경우 원본 fetch 실행
        return originalFetch(url, options);
      }
    };
    
    // 주기적으로 메트릭 전송
    setInterval(() => this.sendMetrics(), 60000);
  }
  
  recordMetrics(url, responseTime, isError) {
    this.metrics.calls++;
    this.metrics.totalResponseTime += responseTime;
    
    if (isError) {
      this.metrics.errors++;
    }
    
    // 개별 엔드포인트별 메트릭 기록도 가능
  }
  
  sendMetrics() {
    if (this.metrics.calls > 0) {
      const avgResponseTime = this.metrics.totalResponseTime / this.metrics.calls;
      const errorRate = (this.metrics.errors / this.metrics.calls) * 100;
      
      // 백엔드 애널리틱스 시스템으로 메트릭 전송
      navigator.sendBeacon('/analytics/api-metrics', JSON.stringify({
        calls: this.metrics.calls,
        avgResponseTime,
        errorRate,
        userAgent: navigator.userAgent,
        timestamp: Date.now()
      }));
      
      // 메트릭 초기화
      this.metrics = {
        calls: 0,
        errors: 0,
        totalResponseTime: 0
      };
    }
  }
}

// 모니터링 시작
const apiMonitor = new APIMonitor();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

# Python/Flask에서의 구조화된 로깅 예시
import logging
import json
import time
import uuid
from flask import Flask, request, g

app = Flask(__name__)

# JSON 형식 로거 설정
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_record = {
            "timestamp": self.formatTime(record, self.datefmt),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name
        }
        
        # 추가 컨텍스트 정보가 있으면 포함
        if hasattr(record, 'request_id'):
            log_record["request_id"] = record.request_id
        
        if hasattr(record, 'path'):
            log_record["path"] = record.path
        
        if hasattr(record, 'method'):
            log_record["method"] = record.method
        
        if hasattr(record, 'duration_ms'):
            log_record["duration_ms"] = record.duration_ms
        
        if hasattr(record, 'status_code'):
            log_record["status_code"] = record.status_code
        
        if hasattr(record, 'user_id'):
            log_record["user_id"] = record.user_id
        
        return json.dumps(log_record)

# 로거 설정
logger = logging.getLogger("api")
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# 요청/응답 로깅 미들웨어
@app.before_request
def before_request():
    g.start_time = time.time()
    g.request_id = str(uuid.uuid4())

@app.after_request
def after_request(response):
    duration = time.time() - g.start_time
    
    # 구조화된 로그 레코드 생성
    logger.info(
        f"API Request Completed",
        extra={
            "request_id": g.request_id,
            "method": request.method,
            "path": request.path,
            "status_code": response.status_code,
            "duration_ms": round(duration * 1000, 2),
            "user_id": g.get('user_id', 'anonymous')
        }
    )
    
    return response

# 오류 로깅
@app.errorhandler(Exception)
def handle_exception(e):
    logger.error(
        f"API Error: {str(e)}",
        extra={
            "request_id": g.get('request_id', 'unknown'),
            "method": request.method,
            "path": request.path,
            "error_type": type(e).__name__
        },
        exc_info=True
    )
    
    return {"error": "Internal Server Error"}, 500

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

# Prometheus 클라이언트를 사용한 메트릭 모니터링 예시
from flask import Flask, request
from prometheus_client import Counter, Histogram, start_http_server
import time

app = Flask(__name__)

# 메트릭 정의
REQUEST_COUNT = Counter(
    'api_requests_total',
    'Total API Request Count',
    ['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'api_request_latency_seconds',
    'API Request Latency',
    ['method', 'endpoint']
)

# 메트릭 수집 미들웨어
@app.before_request
def before_request():
    request.start_time = time.time()

@app.after_request
def after_request(response):
    request_latency = time.time() - request.start_time
    REQUEST_LATENCY.labels(
        method=request.method,
        endpoint=request.path
    ).observe(request_latency)
    
    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.path,
        status=response.status_code
    ).inc()
    
    return response

# Prometheus 메트릭 서버 시작
@app.before_first_request
def start_metrics_server():
    start_http_server(8000)  # 메트릭을 8000 포트에서 제공

@app.route('/api/user/<user_id>')
def get_user(user_id):
    # API 로직...
    return {"user": {"id": user_id, "name": "Test User"}}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# OpenTelemetry를 사용한 분산 추적 예시
from flask import Flask
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
import requests

# 트레이싱 설정
resource = Resource(attributes={SERVICE_NAME: "user-api"})
tracer_provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="localhost:4317"))
tracer_provider.add_span_processor(processor)
trace.set_tracer_provider(tracer_provider)
tracer = trace.get_tracer(__name__)

app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)
RequestsInstrumentor().instrument()

@app.route('/api/user-activity/<user_id>')
def get_user_activity(user_id):
    with tracer.start_as_current_span("get_user_details") as span:
        span.set_attribute("user.id", user_id)
        
        # 사용자 세부 정보 가져오기
        user_response = requests.get(f"http://user-service/users/{user_id}")
        user_data = user_response.json()
        
        with tracer.start_as_current_span("get_activity_data") as activity_span:
            # 사용자 활동 데이터 가져오기
            activity_response = requests.get(f"http://activity-service/activities?user_id={user_id}")
            activity_data = activity_response.json()
            activity_span.set_attribute("activity.count", len(activity_data))
        
        # 응답 데이터 조합
        result = {
            "user": user_data,
            "recent_activities": activity_data[:5]
        }
        
        return result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# Prometheus 알림 규칙 예시 (다층적 접근)
groups:
  - name: api_alerts
    rules:
      # 인프라 계층 알림
      - alert: HighCpuUsage
        expr: avg(rate(node_cpu_seconds_total{mode!="idle"}[5m])) by (instance) > 0.8
        for: 5m
        labels:
          severity: warning
          layer: infrastructure
        annotations:
          summary: "High CPU usage on {{ $labels.instance }}"
          description: "CPU usage is above 80% for 5 minutes"

      # 애플리케이션 계층 알림
      - alert: ApiHighErrorRate
        expr: sum(rate(api_requests_total{status=~"5.."}[5m])) / sum(rate(api_requests_total[5m])) > 0.05
        for: 2m
        labels:
          severity: critical
          layer: application
        annotations:
          summary: "API error rate is high"
          description: "Error rate is above 5% for 2 minutes"

      # 비즈니스 계층 알림
      - alert: LowTransactionSuccessRate
        expr: sum(rate(business_transactions_success_total[5m])) / sum(rate(business_transactions_total[5m])) < 0.95
        for: 5m
        labels:
          severity: critical
          layer: business
        annotations:
          summary: "Business transaction success rate is low"
          description: "Transaction success rate is below 95% for 5 minutes"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

# 알림 임계값 설정 및 에스컬레이션 예시
class AlertManager:
    def __init__(self):
        self.thresholds = {
            'response_time': {
                'warning': 500,    # 500ms
                'critical': 1000   # 1초
            },
            'error_rate': {
                'warning': 0.01,   # 1%
                'critical': 0.05   # 5%
            },
            'request_rate': {
                'warning': 1000,   # 초당 1000 요청
                'critical': 1500   # 초당 1500 요청
            }
        }
        
        self.escalation_policies = {
            'warning': ['slack-channel'],
            'critical': ['slack-channel', 'pagerduty', 'sms']
        }
        
        # 알림 발생 이력 (알림 폭풍 방지)
        self.alert_history = {}
        
    def check_and_alert(self, metric_name, value, context=None):
        if metric_name not in self.thresholds:
            return
        
        severity = None
        if value >= self.thresholds[metric_name]['critical']:
            severity = 'critical'
        elif value >= self.thresholds[metric_name]['warning']:
            severity = 'warning'
        
        if severity:
            # 알림 폭풍 방지: 최근에 비슷한 알림이 발생했는지 확인
            alert_key = f"{metric_name}:{severity}"
            current_time = time.time()
            
            # 지난 10분 이내에 알림이 발생했는지 확인
            if (alert_key in self.alert_history and 
                current_time - self.alert_history[alert_key] < 600):
                # 이미 알림이 발생했으므로 무시
                return
            
            # 알림 발생 시간 기록
            self.alert_history[alert_key] = current_time
            
            # 알림 생성 및 전송
            alert = {
                'metric': metric_name,
                'value': value,
                'threshold': self.thresholds[metric_name][severity],
                'severity': severity,
                'timestamp': current_time,
                'context': context or {}
            }
            
            self.send_alert(alert, self.escalation_policies[severity])
    
    def send_alert(self, alert, channels):
        for channel in channels:
            if channel == 'slack-channel':
                self.send_to_slack(alert)
            elif channel == 'pagerduty':
                self.send_to_pagerduty(alert)
            elif channel == 'sms':
                self.send_sms(alert)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

# API 버전별 사용 현황 모니터링 예시
def track_api_version_usage(request, response):
    """API 버전 사용 현황을 모니터링하는 미들웨어"""
    version = get_api_version(request)
    
    # 버전별 사용량 메트릭 기록
    increment_counter('api.version.usage', tags={'version': version})
    
    # 폐기 예정 버전에 대한 경고 로깅
    if version in DEPRECATED_VERSIONS:
        log.warning(
            "Deprecated API version used",
            extra={
                'version': version,
                'client_id': get_client_id(request),
                'endpoint': request.path,
                'sunset_date': SUNSET_DATES.get(version, 'unknown')
            }
        )
        
        # 응답 헤더에 폐기 관련 정보 추가
        sunset_date = SUNSET_DATES.get(version)
        if sunset_date:
            response.headers['Sunset'] = sunset_date
            response.headers['Deprecation'] = 'true'
            response.headers['Link'] = f'<{MIGRATION_GUIDES[version]}>; rel="deprecation"; type="text/html"'
    
    return response

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

# Flask에서의 종합적인 건강 검사 API 구현
from flask import Flask, jsonify
import redis
import psycopg2
import requests
import time

app = Flask(__name__)

def check_database():
    """데이터베이스 연결 및 기본 쿼리 건강 검사"""
    try:
        conn = psycopg2.connect(DATABASE_URL)
        cursor = conn.cursor()
        cursor.execute("SELECT 1")  # 간단한 쿼리로 연결 확인
        cursor.close()
        conn.close()
        return True, "Database connection is healthy"
    except Exception as e:
        return False, f"Database connection failed: {str(e)}"

def check_redis():
    """Redis 연결 건강 검사"""
    try:
        redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT)
        redis_client.ping()  # PING 명령으로 연결 확인
        return True, "Redis connection is healthy"
    except Exception as e:
        return False, f"Redis connection failed: {str(e)}"

def check_external_api():
    """외부 의존성 API 건강 검사"""
    try:
        response = requests.get(
            "https://external-api.example.com/health",
            timeout=3  # 3초 타임아웃 설정
        )
        if response.status_code == 200:
            return True, "External API is reachable"
        else:
            return False, f"External API returned status code {response.status_code}"
    except Exception as e:
        return False, f"External API check failed: {str(e)}"

@app.route('/health')
def basic_health_check():
    """기본 가용성 확인을 위한 간단한 건강 검사"""
    return jsonify({"status": "UP"})

@app.route('/health/detailed')
def detailed_health_check():
    """상세한, 종합적인 건강 검사"""
    start_time = time.time()
    
    checks = {
        "database": check_database(),
        "redis": check_redis(),
        "external_api": check_external_api()
    }
    
    # 전체 상태 계산
    all_healthy = all(result[0] for result in checks.values())
    overall_status = "UP" if all_healthy else "DOWN"
    
    # 응답 형식화
    response = {
        "status": overall_status,
        "checks": {
            name: {
                "status": "UP" if result[0] else "DOWN",
                "message": result[1]
            }
            for name, result in checks.items()
        },
        "time": round((time.time() - start_time) * 1000, 2)  # ms 단위 응답 시간
    }
    
    status_code = 200 if all_healthy else 503  # 서비스 사용 불가 상태 코드
    return jsonify(response), status_code

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

// 샘플링 구현 예시 (Node.js)
function shouldSample(requestId) {
  // 현재 샘플링 비율 (환경 변수나 설정에서 가져옴)
  const samplingRate = parseFloat(process.env.TRACING_SAMPLING_RATE || "0.1");  // 기본값 10%
  
  // requestId의 해시값을 계산하여 0-1 사이의 값으로 정규화
  const hash = calculateHash(requestId);
  const normalizedHash = hash / Number.MAX_SAFE_INTEGER;
  
  // 정규화된 해시값이 샘플링 비율보다 작으면 샘플링
  return normalizedHash < samplingRate;
}

// 미들웨어에서 사용
app.use((req, res, next) => {
  const requestId = req.headers['x-request-id'] || uuidv4();
  req.requestId = requestId;
  
  // 샘플링 여부 결정
  req.shouldTrace = shouldSample(requestId);
  
  // 모든 요청에 대해 기본 메트릭은 항상 수집
  collectBasicMetrics(req, res);
  
  // 샘플링된 요청에 대해서만 상세 추적 활성화
  if (req.shouldTrace) {
    enableDetailedTracing(req, res);
  }
  
  next();
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# API 보안 모니터링 예시
def monitor_security_events(request, response):
    # 인증 실패 모니터링
    if response.status_code == 401:
        increment_counter('api.auth.failure', tags={
            'ip': request.remote_addr,
            'user_agent': request.user_agent.string,
            'endpoint': request.path
        })
        
        # 동일 IP에서의 인증 실패 횟수 확인
        failure_count = get_recent_failures(request.remote_addr)
        if failure_count > THRESHOLD_SUSPICIOUS_AUTH_FAILURES:
            log_security_event('excessive_auth_failures', {
                'ip': request.remote_addr,
                'count': failure_count,
                'time_window': 'last 1 hour'
            })
    
    # API 속도 제한 모니터링
    if response.status_code == 429:
        increment_counter('api.rate_limit.exceeded', tags={
            'client_id': get_client_id(request),
            'endpoint': request.path
        })
        
        # 속도 제한 초과 패턴 분석
        if is_abnormal_rate_pattern(get_client_id(request)):
            log_security_event('suspicious_rate_limit_pattern', {
                'client_id': get_client_id(request),
                'pattern': 'burst_traffic'
            })
    
    # 데이터 유출 의심 모니터링 (비정상적으로 큰 응답)
    if response.content_length > THRESHOLD_LARGE_RESPONSE:
        log_security_event('large_data_transfer', {
            'client_id': get_client_id(request),
            'endpoint': request.path,
            'size': response.content_length
        })

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

// 상관 ID 전파 예시 (Java/Spring)
@Component
public class CorrelationIdFilter extends OncePerRequestFilter {

    private static final String CORRELATION_ID_HEADER = "X-Correlation-ID";

    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, 
                                   FilterChain filterChain) throws ServletException, IOException {
        String correlationId = request.getHeader(CORRELATION_ID_HEADER);
        
        // 상관 ID가 없으면 새로 생성
        if (correlationId == null || correlationId.isEmpty()) {
            correlationId = UUID.randomUUID().toString();
        }
        
        // MDC에 상관 ID 설정 (로깅용)
        MDC.put("correlationId", correlationId);
        
        // 응답 헤더에 상관 ID 추가
        response.addHeader(CORRELATION_ID_HEADER, correlationId);
        
        try {
            // 다운스트림 서비스 호출 시 상관 ID 전달을 위한 설정
            RequestContextHolder.currentRequestAttributes()
                .setAttribute("correlationId", correlationId, RequestAttributes.SCOPE_REQUEST);
            
            filterChain.doFilter(request, response);
        } finally {
            MDC.remove("correlationId");
        }
    }
}

// RestTemplate에 상관 ID 추가
@Configuration
public class RestTemplateConfig {

    @Bean
    public RestTemplate restTemplate() {
        RestTemplate restTemplate = new RestTemplate();
        restTemplate.setInterceptors(
            Collections.singletonList((request, body, execution) -> {
                String correlationId = (String) RequestContextHolder.currentRequestAttributes()
                    .getAttribute("correlationId", RequestAttributes.SCOPE_REQUEST);
                
                if (correlationId != null) {
                    request.getHeaders().add("X-Correlation-ID", correlationId);
                }
                
                return execution.execute(request, body);
            })
        );
        return restTemplate;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

# 이상 탐지 예시 (Python/SciPy)
from scipy import stats
import numpy as np

class AnomalyDetector:
    def __init__(self, window_size=60, threshold=3.0):
        self.window_size = window_size  # 분석할 이전 데이터 포인트 수
        self.threshold = threshold      # 이상 감지 z-score 임계값
        self.history = []               # 이전 데이터 포인트 저장
    
    def is_anomaly(self, value):
        """현재 값이 이상치인지 확인"""
        self.history.append(value)
        
        # 충분한 데이터가 없으면 이상으로 간주하지 않음
        if len(self.history) < self.window_size:
            return False
        
        # 윈도우 크기 유지
        if len(self.history) > self.window_size:
            self.history.pop(0)
        
        # z-score 계산
        mean = np.mean(self.history[:-1])  # 현재 값을 제외한 평균
        std = np.std(self.history[:-1])    # 현재 값을 제외한 표준편차
        
        # 표준편차가 0에 가까우면 일정한 값이므로 이상이 아님
        if std < 1e-10:
            return False
        
        z_score = abs((value - mean) / std)
        
        return z_score > self.threshold

# 사용 예시
response_time_detector = AnomalyDetector(window_size=100, threshold=3.5)

def monitor_api_response_time(endpoint, response_time):
    if response_time_detector.is_anomaly(response_time):
        log.warning(
            f"Anomalous response time detected for {endpoint}: {response_time}ms "
            f"(threshold: {response_time_detector.threshold} standard deviations)"
        )
        
        # 알림 전송
        send_alert(
            title=f"Anomalous API Response Time: {endpoint}",
            message=f"Response time of {response_time}ms is abnormally high.",
            severity="warning"
        )

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

# 지능형 알림 시스템 예시
class SmartAlertSystem:
    def __init__(self):
        # 알림 상태 저장
        self.active_alerts = {}
        # 알림 에스컬레이션 상태
        self.escalations = {}
        # 자동 해결 동작
        self.auto_remediation_actions = {
            'high_error_rate': self.restart_service,
            'memory_leak': self.increase_memory_limit,
            'database_connection_issues': self.check_db_connections
        }
    
    def process_alert(self, alert_type, severity, context):
        """알림 처리 및 지능형 라우팅"""
        alert_key = f"{alert_type}:{context.get('service', 'unknown')}"
        
        # 이미 활성화된 알림인지 확인
        if alert_key in self.active_alerts:
            # 기존 알림 상태 업데이트
            self.active_alerts[alert_key]['count'] += 1
            self.active_alerts[alert_key]['last_seen'] = time.time()
            
            # 알림이 지속되면 에스컬레이션
            duration = time.time() - self.active_alerts[alert_key]['first_seen']
            if duration > 900 and alert_key not in self.escalations:  # 15분 이상 지속
                self.escalate_alert(alert_key, context)
        else:
            # 새로운 알림 생성
            self.active_alerts[alert_key] = {
                'type': alert_type,
                'severity': severity,
                'context': context,
                'count': 1,
                'first_seen': time.time(),
                'last_seen': time.time()
            }
            
            # 즉시 알림
            self.send_initial_alert(alert_key)
            
            # 자동 해결 시도
            self.attempt_auto_remediation(alert_type, context)
    
    def escalate_alert(self, alert_key, context):
        """알림 에스컬레이션 - 더 높은 우선순위로 라우팅"""
        alert = self.active_alerts[alert_key]
        
        # 에스컬레이션 상태 저장
        self.escalations[alert_key] = {
            'time': time.time(),
            'previous_severity': alert['severity']
        }
        
        # 심각도 상향 조정
        new_severity = 'critical' if alert['severity'] == 'warning' else 'emergency'
        alert['severity'] = new_severity
        
        # 상향된 심각도로 알림 전송
        message = (
            f"ESCALATED ALERT: {alert['type']} issue has been ongoing for "
            f"{int((time.time() - alert['first_seen']) / 60)} minutes. "
            f"Occurred {alert['count']} times."
        )
        
        # 에스컬레이션 경로에 따라 알림 전송
        self.send_escalated_alert(alert_key, message, new_severity)
    
    def attempt_auto_remediation(self, alert_type, context):
        """자동 문제 해결 시도"""
        if alert_type in self.auto_remediation_actions:
            try:
                result = self.auto_remediation_actions[alert_type](context)
                
                if result['success']:
                    log.info(
                        f"Auto-remediation for {alert_type} successful: {result['message']}"
                    )
                else:
                    log.warning(
                        f"Auto-remediation for {alert_type} failed: {result['message']}"
                    )
            except Exception as e:
                log.error(f"Error during auto-remediation for {alert_type}: {str(e)}")
    
    # 자동 해결 동작 구현
    def restart_service(self, context):
        service_name = context.get('service')
        # 서비스 재시작 로직…
        return {'success': True, 'message': f"Service {service_name} restarted successfully"}
    
    def increase_memory_limit(self, context):
        service_name = context.get('service')
        # 메모리 한도 증가 로직…
        return {'success': True, 'message': f"Memory limit for {service_name} increased"}
    
    def check_db_connections(self, context):
        # 데이터베이스 연결 풀 확인 및 리셋 로직…
        return {'success': True, 'message': "Database connection pool reset"}