시니어 SRE 인터뷰 질문과 답변 예시

# 중요한 사용자 여정 식별
user_journeys = {
    'search_product': {
        'steps': ['search_query', 'results_display', 'product_click'],
        'criticality': 'high',
        'expected_latency': '< 500ms'
    },
    'checkout': {
        'steps': ['add_to_cart', 'payment', 'confirmation'],
        'criticality': 'critical',
        'expected_latency': '< 2s'
    },
    'browse_recommendations': {
        'steps': ['load_page', 'fetch_recommendations'],
        'criticality': 'medium',
        'expected_latency': '< 1s'
    }
}

# SLI 사양
slis:
  availability:
    description: "성공적인 요청의 비율"
    measurement: "count(http_status < 500) / count(http_requests)"
    
  latency:
    description: "95번째 백분위수 요청 지연 시간"
    measurement: "histogram_quantile(0.95, http_request_duration_seconds)"
    
  correctness:
    description: "정확한 데이터를 반환하는 요청의 비율"
    measurement: "count(validation_passed) / count(requests)"

def calculate_initial_slo(service_type, criticality):
    """
    서비스 특성을 기반으로 초기 SLO 계산
    """
    base_slos = {
        'critical': {
            'availability': 0.999,  # 99.9%
            'latency_p95': 1.0,     # 1초
            'latency_p99': 2.0      # 2초
        },
        'high': {
            'availability': 0.995,  # 99.5%
            'latency_p95': 2.0,
            'latency_p99': 5.0
        },
        'medium': {
            'availability': 0.99,   # 99%
            'latency_p95': 5.0,
            'latency_p99': 10.0
        }
    }
    
    return base_slos.get(criticality, base_slos['medium'])

# 예시
checkout_slo = calculate_initial_slo('payment', 'critical')
print(f"Checkout SLO: {checkout_slo}")

## SLO 가정 (초기)

### 가용성: 99.9%
- 가정: 표준 클라우드 인프라 안정성
- 오류 예산: 월 43분
- 검토: 3개월 데이터 후

### 지연 시간 (p95): < 1초
- 가정: 데이터베이스 쿼리 < 100ms
- 가정: 복잡한 계산 없음
- 검토: 쿼리 패턴이 변경되는 경우

### 종속성
- 외부 API 가용성: 99.95%
- 데이터베이스 가용성: 99.99%

class SLOTier:
    def __init__(self, name, availability, latency_p95, latency_p99):
        self.name = name
        self.availability = availability
        self.latency_p95 = latency_p95
        self.latency_p99 = latency_p99
        self.error_budget = 1 - availability

# 단계 정의
tiers = {
    'premium': SLOTier(
        name='Premium',
        availability=0.9999,  # 99.99% - 월 4.3분
        latency_p95=0.5,
        latency_p99=1.0
    ),
    'standard': SLOTier(
        name='Standard',
        availability=0.999,   # 99.9% - 월 43분
        latency_p95=1.0,
        latency_p99=2.0
    ),
    'free': SLOTier(
        name='Free',
        availability=0.99,    # 99% - 월 7.2시간
        latency_p95=2.0,
        latency_p99=5.0
    )
}

# 단계를 기반으로 요청 라우팅
def get_user_tier(user_id):
    # 사용자 구독 단계 조회
    return user_subscription_tier(user_id)

def apply_slo_policy(user_id, request):
    tier = get_user_tier(user_id)
    slo = tiers[tier]
    
    # 단계별 정책 적용
    request.timeout = slo.latency_p99
    request.priority = tier  # 큐 우선순위 지정
    request.retry_budget = calculate_retry_budget(slo.error_budget)
    
    return request

# Kubernetes 예시: 단계별 별도 배포
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-premium
spec:
  replicas: 10
  template:
    spec:
      containers:
      - name: api
        resources:
          requests:
            cpu: "2"
            memory: "4Gi"
          limits:
            cpu: "4"
            memory: "8Gi"
      priorityClassName: high-priority
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-standard
spec:
  replicas: 5
  template:
    spec:
      containers:
      - name: api
        resources:
          requests:
            cpu: "1"
            memory: "2Gi"

# 단계별 가용성
sum(rate(http_requests_total{status!~"5.."}[5m])) by (tier)
/
sum(rate(http_requests_total[5m])) by (tier)

# 단계별 지연 시간
histogram_quantile(0.95,
  rate(http_request_duration_seconds_bucket[5m])
) by (tier)

import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression

class CapacityPlanner:
    def __init__(self, metrics_data):
        self.data = pd.DataFrame(metrics_data)
    
    def analyze_trends(self, metric_name, days=30):
        """과거 추세 분석"""
        metric_data = self.data[metric_name].tail(days * 24)  # 시간별 데이터
        
        # 성장률 계산
        start_value = metric_data.iloc[0]
        end_value = metric_data.iloc[-1]
        growth_rate = ((end_value - start_value) / start_value) * 100
        
        # 피크 사용량 식별
        peak_value = metric_data.max()
        peak_time = metric_data.idxmax()
        
        # 백분위수 계산
        p50 = metric_data.quantile(0.50)
        p95 = metric_data.quantile(0.95)
        p99 = metric_data.quantile(0.99)
        
        return {
            'growth_rate': growth_rate,
            'peak_value': peak_value,
            'peak_time': peak_time,
            'p50': p50,
            'p95': p95,
            'p99': p99
        }
    
    def forecast_capacity(self, metric_name, months_ahead=3):
        """미래 용량 요구 사항 예측"""
        # 데이터 준비
        df = self.data[[metric_name]].reset_index()
        df['days'] = (df.index / 24).astype(int)  # 시간을 일로 변환
        
        # 모델 학습
        X = df[['days']].values
        y = df[metric_name].values
        
        model = LinearRegression()
        model.fit(X, y)
        
        # 예측
        future_days = np.array([[df['days'].max() + (30 * months_ahead)]])
        forecast = model.predict(future_days)[0]
        
        # 안전 마진 추가 (20%)
        forecast_with_margin = forecast * 1.2
        
        return {
            'forecast': forecast,
            'with_margin': forecast_with_margin,
            'current': y[-1],
            'growth_factor': forecast / y[-1]
        }
    
    def calculate_resource_needs(self, requests_per_second, 
                                 requests_per_instance=100,
                                 headroom=0.3):
        """필요한 인스턴스 계산"""
        # 기본 용량
        base_instances = np.ceil(requests_per_second / requests_per_instance)
        
        # 스파이크 및 유지 관리를 위한 헤드룸 추가
        total_instances = np.ceil(base_instances * (1 + headroom))
        
        return {
            'base_instances': int(base_instances),
            'total_instances': int(total_instances),
            'headroom_instances': int(total_instances - base_instances)
        }

# 사용 예시
metrics = {
    'requests_per_second': [100, 105, 110, 115, 120, ...],  # 과거 데이터
    'cpu_usage': [45, 48, 50, 52, 55, ...],
    'memory_usage': [60, 62, 65, 67, 70, ...]
}

planner = CapacityPlanner(metrics)

# 추세 분석
trends = planner.analyze_trends('requests_per_second', days=30)
print(f"성장률: {trends['growth_rate']:.2f}%")
print(f"피크 RPS: {trends['peak_value']}")

# 용량 예측
forecast = planner.forecast_capacity('requests_per_second', months_ahead=3)
print(f"3개월 후 예측 RPS: {forecast['forecast']:.0f}")
print(f"안전 마진 포함: {forecast['with_margin']:.0f}")

# 리소스 요구 사항 계산
resources = planner.calculate_resource_needs(
    requests_per_second=forecast['with_margin'],
    requests_per_instance=100,
    headroom=0.3
)
print(f"필요한 인스턴스: {resources['total_instances']}")

def optimize_instance_mix(workload_profile):
    """
    비용에 대한 인스턴스 유형 최적화
    """
    # 인스턴스 유형 혼합
    instance_types = {
        'on_demand': {
            'cost_per_hour': 0.10,
            'reliability': 1.0,
            'percentage': 0.3  # 기준선에 대한 30% 온디맨드
        },
        'spot': {
            'cost_per_hour': 0.03,
            'reliability': 0.95,
            'percentage': 0.5  # 비용 절감을 위한 50% 스팟
        },
        'reserved': {
            'cost_per_hour': 0.06,
            'reliability': 1.0,
            'percentage': 0.2  # 예측 가능한 로드에 대한 20% 예약
        }
    }
    
    total_instances = workload_profile['total_instances']
    
    allocation = {}
    for instance_type, config in instance_types.items():
        count = int(total_instances * config['percentage'])
        allocation[instance_type] = {
            'count': count,
            'monthly_cost': count * config['cost_per_hour'] * 730
        }
    
    return allocation

# 카오스 실험 프레임워크
from dataclasses import dataclass
from enum import Enum
import random
import time

class ExperimentStatus(Enum):
    PLANNED = "planned"
    RUNNING = "running"
    COMPLETED = "completed"
    ABORTED = "aborted"

@dataclass
class ChaosExperiment:
    name: str
    hypothesis: str
    blast_radius: float  # 영향을 받는 트래픽 비율
    duration_seconds: int
    rollback_criteria: dict
    
    def __post_init__(self):
        self.status = ExperimentStatus.PLANNED
        self.metrics_before = {}
        self.metrics_during = {}
        self.metrics_after = {}

class ChaosRunner:
    def __init__(self, monitoring_client):
        self.monitoring = monitoring_client
        self.experiments = []
    
    def run_experiment(self, experiment: ChaosExperiment):
        """안전 점검으로 카오스 실험 실행"""
        print(f"실험 시작: {experiment.name}")
        
        # 1. 기준 측정
        experiment.metrics_before = self.measure_metrics()
        print(f"기준 메트릭: {experiment.metrics_before}")
        
        # 2. 시스템이 정상인지 확인
        if not self.is_system_healthy(experiment.metrics_before):
            print("시스템이 비정상, 실험 중단")
            return False
        
        # 3. 오류 주입
        experiment.status = ExperimentStatus.RUNNING
        failure_injection = self.inject_failure(experiment)
        
        try:
            # 4. 실험 중 모니터링
            start_time = time.time()
            while time.time() - start_time < experiment.duration_seconds:
                experiment.metrics_during = self.measure_metrics()
                
                # 롤백 기준 확인
                if self.should_rollback(experiment):
                    print("롤백 기준 충족, 실험 중단")
                    self.rollback(failure_injection)
                    experiment.status = ExperimentStatus.ABORTED
                    return False
                
                time.sleep(10)  # 10초마다 확인
            
            # 5. 오류 주입 롤백
            self.rollback(failure_injection)
            
            # 6. 복구 측정
            time.sleep(60)  # 시스템이 안정화될 때까지 대기
            experiment.metrics_after = self.measure_metrics()
            
            # 7. 결과 분석
            experiment.status = ExperimentStatus.COMPLETED
            return self.analyze_results(experiment)
            
        except Exception as e:
            print(f"실험 실패: {e}")
            self.rollback(failure_injection)
            experiment.status = ExperimentStatus.ABORTED
            return False
    
    def inject_failure(self, experiment):
        """특정 오류 유형 주입"""
        # 구현은 오류 유형에 따라 다름
        pass
    
    def measure_metrics(self):
        """주요 시스템 메트릭 측정"""
        return {
            'error_rate': self.monitoring.get_error_rate(),
            'latency_p95': self.monitoring.get_latency_p95(),
            'requests_per_second': self.monitoring.get_rps(),
            'availability': self.monitoring.get_availability()
        }
    
    def is_system_healthy(self, metrics):
        """시스템이 SLO를 충족하는지 확인"""
        return (
            metrics['error_rate'] < 0.01 and  # < 1% 오류
            metrics['latency_p95'] < 1.0 and  # < 1초 지연 시간
            metrics['availability'] > 0.999   # > 99.9% 가용성
        )
    
    def should_rollback(self, experiment):
        """실험을 중단해야 하는지 확인"""
        current = experiment.metrics_during
        criteria = experiment.rollback_criteria
        
        return (
            current['error_rate'] > criteria.get('max_error_rate', 0.05) or
            current['latency_p95'] > criteria.get('max_latency', 5.0) or
            current['availability'] < criteria.get('min_availability', 0.99)
        )
    
    def rollback(self, failure_injection):
        """오류 주입 제거"""
        print("오류 주입 롤백")
        # 구현은 오류 유형에 따라 다름
    
    def analyze_results(self, experiment):
        """실험 결과 분석"""
        before = experiment.metrics_before
        during = experiment.metrics_during
        after = experiment.metrics_after
        
        print(f"\n실험 결과: {experiment.name}")
        print(f"가설: {experiment.hypothesis}")
        print(f"\n메트릭:")
        print(f"  이전: {before}")
        print(f"  중간: {during}")
        print(f"  이후: {after}")
        
        # 가설이 검증되었는지 확인
        hypothesis_validated = (
            during['availability'] >= experiment.rollback_criteria['min_availability']
        )
        
        return hypothesis_validated

# 실험 예시
experiment = ChaosExperiment(
    name="데이터베이스 장애 조치 테스트",
    hypothesis="데이터베이스 장애 조치 중 시스템 사용 가능",
    blast_radius=0.1,  # 트래픽의 10%
    duration_seconds=300,  # 5분
    rollback_criteria={
        'max_error_rate': 0.05,
        'max_latency': 5.0,
        'min_availability': 0.99
    }
)

# tc (트래픽 제어)를 사용하여 지연 시간 추가
tc qdisc add dev eth0 root netem delay 100ms 20ms

# 롤백
tc qdisc del dev eth0 root

# 임의의 포드 삭제
kubectl delete pod -l app=myapp --random=1

# Chaos Mesh 사용
kubectl apply -f - <<EOF
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: pod-failure
spec:
  action: pod-failure
  mode: one
  selector:
    namespaces:
      - production
    labelSelectors:
      app: myapp
  duration: "30s"
EOF

# CPU 스트레스
stress-ng --cpu 4 --timeout 60s

# 메모리 스트레스
stress-ng --vm 2 --vm-bytes 2G --timeout 60s

## 사고 지휘관 체크리스트

### 즉각적인 조치
- [ ] 사고 인지
- [ ] 심각도 평가 (SEV-1, SEV-2, SEV-3)
- [ ] 사고 채널 생성 (#incident-YYYY-MM-DD-NNN)
- [ ] 적절한 팀에 페이지
- [ ] 역할 지정:
  - [ ] 기술 리드
  - [ ] 커뮤니케이션 리드
  - [ ] 기록자 (타임라인 문서화)

### 심각도 평가
**SEV-1 (심각):**
- 완전한 서비스 중단
- 데이터 손실
- 보안 침해
- > 50% 사용자에게 영향

**SEV-2 (높음):**
- 부분 중단
- 성능 저하
- 10-50% 사용자에게 영향

**SEV-3 (보통):**
- 경미한 저하
- < 10% 사용자에게 영향
- 해결 방법 사용 가능

class IncidentCommander:
    def __init__(self, incident_id):
        self.incident_id = incident_id
        self.timeline = []
        self.status_updates = []
        self.action_items = []
    
    def coordinate_investigation(self):
        """기술 조사 조정"""
        # 병렬 조사 트랙
        tracks = {
            'recent_changes': self.check_recent_deployments(),
            'infrastructure': self.check_infrastructure_health(),
            'dependencies': self.check_external_dependencies(),
            'metrics': self.analyze_metrics_anomalies()
        }
        
        return tracks
    
    def make_decision(self, options, deadline_minutes=5):
        """시간 제한 결정"""
        print(f"{deadline_minutes}분 이내에 결정 필요")
        print(f"옵션: {options}")
        
        # 기술 리드로부터 입력 수집
        # 다음을 기반으로 결정:
        # - 사용자 영향
        # - 각 옵션의 위험
        # - 구현 시간
        # - 가역성
        
        return selected_option
    
    def communicate_status(self, interval_minutes=15):
        """정기적인 상태 업데이트"""
        update = {
            'timestamp': datetime.now(),
            'status': self.get_current_status(),
            'impact': self.assess_user_impact(),
            'eta': self.estimate_resolution_time(),
            'next_update': datetime.now() + timedelta(minutes=interval_minutes)
        }
        
        # 이해 관계자에게 전송
        self.send_status_update(update)
        self.status_updates.append(update)
    
    def log_timeline_event(self, event, timestamp=None):
        """사고 타임라인 문서화"""
        self.timeline.append({
            'timestamp': timestamp or datetime.now(),
            'event': event,
            'logged_by': self.get_current_user()
        })

def evaluate_mitigation_options():
    """완화 옵션 평가 및 우선순위 지정"""
    options = [
        {
            'action': '배포 롤백',
            'time_to_implement': 5,  # 분
            'risk': '낮음',
            'effectiveness': '높음',
            'reversible': True
        },
        {
            'action': '리소스 확장',
            'time_to_implement': 2,
            'risk': '낮음',
            'effectiveness': '보통',
            'reversible': True
        },
        {
            'action': '기능 플래그 비활성화',
            'time_to_implement': 1,
            'risk': '낮음',
            'effectiveness': '높음',
            'reversible': True
        },
        {
            'action': '데이터베이스 장애 조치',
            'time_to_implement': 10,
            'risk': '보통',
            'effectiveness': '높음',
            'reversible': False
        }
    ]
    
    # 다음 순서로 정렬: 낮은 위험, 높은 효과, 빠른 구현
    sorted_options = sorted(
        options,
        key=lambda x: (
            x['risk'] == '낮음',
            x['effectiveness'] == '높음',
            -x['time_to_implement']
        ),
        reverse=True
    )
    
    return sorted_options

# 사고 사후 분석: API 중단

**날짜:** 2024-11-25  
**기간:** 45분  
**심각도:** SEV-1  
**사고 지휘관:** Alice  
**기술 리드:** Bob

## 요약
데이터베이스 연결 풀 소모로 인해 모든 사용자에게 영향을 미치는 완전한 API 중단.

## 영향
- **영향을 받은 사용자:** 100% (모든 사용자)
- **기간:** 45분
- **수익 영향:** ~ $50,000
- **SLO 영향:** 월별 오류 예산의 75% 소모

## 근본 원인
사고 2시간 전에 배포된 새로운 기능의 연결 누수로 인해 데이터베이스 연결 풀이 소모되었습니다.

## 타임라인
| 시간 | 이벤트 |
|------|-------|
| 14:00 | v2.5.0 배포 |
| 15:45 | 지연 시간 증가에 대한 첫 번째 경고 |
| 15:50 | 완전한 API 중단 |
| 15:52 | 사고 선언 (SEV-1) |
| 15:55 | 데이터베이스를 병목 현상으로 식별 |
| 16:05 | 배포 롤백 결정 |
| 16:15 | 롤백 완료 |
| 16:20 | 서비스 복구 중 |
| 16:35 | 완전한 복구 확인 |

## 잘된 점
- 빠른 탐지 (첫 번째 증상으로부터 5분)
- 명확한 사고 지휘 구조
- 빠른 롤백 결정
- 이해 관계자와의 원활한 커뮤니케이션

## 잘못된 점
- 테스트에서 연결 누수 포착 실패
- 연결 풀 모니터링 없음
- 피크 시간 동안 배포

## 조치 항목
| 조치 | 소유자 | 기한 | 상태 |
|--------|-------|----------|--------|
| 연결 풀 모니터링 추가 | Alice | 2024-12-01 | 열림 |
| 테스트에서 연결 누수 감지 구현 | Bob | 2024-12-05 | 열림 |
| 배포 정책 업데이트 (피크 시간 방지) | Charlie | 2024-11-30 | 열림 |
| 데이터베이스 연결에 대한 회로 차단기 추가 | David | 2024-12-10 | 열림 |

## 배운 점
- 모니터링 격차는 중요한 문제를 숨길 수 있습니다.
- 배포 시기가 중요합니다.
- 리소스 누수에 대한 더 나은 통합 테스트 필요

# Istio 예시: 회로 차단 및 재시도
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: api-service
spec:
  host: api-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        http2MaxRequests: 100
        maxRequestsPerConnection: 2
    outlierDetection:
      consecutiveErrors: 5
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
      minHealthPercent: 40
    loadBalancer:
      simple: LEAST_REQUEST
---
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: api-service
spec:
  hosts:
  - api-service
  http:
  - retries:
      attempts: 3
      perTryTimeout: 2s
      retryOn: 5xx,reset,connect-failure,refused-stream
    timeout: 10s
    route:
    - destination:
        host: api-service

from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# 추적 설정
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",
    agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

# 요청 라이브러리 계측
RequestsInstrumentor().instrument()

# 코드에서 사용
tracer = trace.get_tracer(__name__)

def process_order(order_id):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", order_id)
        
        # 결제 서비스 호출
        with tracer.start_as_current_span("call_payment_service"):
            payment_result = call_payment_service(order_id)
        
        # 재고 서비스 호출
        with tracer.start_as_current_span("call_inventory_service"):
            inventory_result = call_inventory_service(order_id)
        
        return combine_results(payment_result, inventory_result)

import asyncio
from asyncio import Semaphore

class BulkheadExecutor:
    def __init__(self, max_concurrent=10):
        self.semaphore = Semaphore(max_concurrent)
        self.active_requests = 0
    
    async def execute(self, func, *args, **kwargs):
        """동시성 제한으로 함수 실행"""
        async with self.semaphore:
            self.active_requests += 1
            try:
                result = await func(*args, **kwargs)
                return result
            finally:
                self.active_requests -= 1

# 서로 다른 서비스에 대한 별도 벌크헤드
payment_bulkhead = BulkheadExecutor(max_concurrent=20)
inventory_bulkhead = BulkheadExecutor(max_concurrent=10)
notification_bulkhead = Bulkhead