Cell-Based Architecture

1
2
3
4
5
6
7
8
9

# 고객 ID 기반 셀 분할
def get_cell_id(customer_id, total_cells):
    """고객 ID를 기반으로 셀 ID 계산"""
    return customer_id % total_cells

# 지역 기반 셀 분할
def get_cell_by_region(region, cell_mapping):
    """지역을 기반으로 셀 선택"""
    return cell_mapping.get(region, 'default_cell')

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

class CellRouter:
    def __init__(self, cells):
        self.cells = cells
        self.health_checker = HealthChecker()
    
    def route_request(self, request):
        """요청을 적절한 셀로 라우팅"""
        partition_key = self.extract_partition_key(request)
        cell_id = self.calculate_cell_id(partition_key)
        
        # 셀 상태 확인
        if self.health_checker.is_healthy(cell_id):
            return self.forward_to_cell(cell_id, request)
        else:
            # 장애 조치: 다른 셀로 라우팅
            return self.route_to_backup_cell(request)

1
2
3
4
5
6
7
8
9

class DataPartitioner:
    def __init__(self, partition_config):
        self.partition_config = partition_config
        self.hash_ring = ConsistentHashRing()
    
    def partition_data(self, data, partition_key):
        """데이터를 적절한 파티션에 저장"""
        cell_id = self.hash_ring.get_cell(partition_key)
        return self.store_in_cell(cell_id, data)

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

class CellAutoScaler:
    def __init__(self, scaling_policy):
        self.scaling_policy = scaling_policy
        self.metrics_collector = MetricsCollector()
    
    def scale_cells(self):
        """메트릭 기반 셀 자동 확장"""
        current_metrics = self.metrics_collector.get_metrics()
        
        if self.should_scale_out(current_metrics):
            self.create_new_cell()
        elif self.should_scale_in(current_metrics):
            self.remove_cell()

  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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270

import asyncio
import hashlib
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class CellStatus(Enum):
    HEALTHY = "healthy"
    UNHEALTHY = "unhealthy"
    MAINTENANCE = "maintenance"

@dataclass
class Cell:
    """셀 정보를 담는 데이터 클래스"""
    cell_id: str
    region: str
    capacity: int
    current_load: int
    status: CellStatus
    endpoint: str
    
    def get_load_percentage(self) -> float:
        """현재 부하 비율 계산"""
        return (self.current_load / self.capacity) * 100

@dataclass
class User:
    """사용자 정보"""
    user_id: str
    region: str
    subscription_tier: str

@dataclass
class VideoRequest:
    """비디오 요청 정보"""
    user: User
    video_id: str
    quality: str
    timestamp: float

class CellRouter:
    """셀 라우터 - 요청을 적절한 셀로 라우팅"""
    
    def __init__(self):
        self.cells: Dict[str, Cell] = {}
        self.regional_cells: Dict[str, List[str]] = {}
        self.health_check_interval = 30  # 30초마다 헬스체크
        
    def register_cell(self, cell: Cell):
        """셀 등록"""
        self.cells[cell.cell_id] = cell
        
        # 지역별 셀 목록 업데이트
        if cell.region not in self.regional_cells:
            self.regional_cells[cell.region] = []
        self.regional_cells[cell.region].append(cell.cell_id)
        
        print(f"셀 등록 완료: {cell.cell_id} (지역: {cell.region})")
    
    def get_partition_key(self, request: VideoRequest) -> str:
        """분할 키 생성 - 사용자 ID 기반 해시"""
        return hashlib.md5(request.user.user_id.encode()).hexdigest()
    
    def select_cell(self, request: VideoRequest) -> Optional[Cell]:
        """최적 셀 선택 로직"""
        # 1단계: 사용자 지역 기반 셀 후보 선택
        user_region = request.user.region
        candidate_cells = self.regional_cells.get(user_region, [])
        
        if not candidate_cells:
            # 사용자 지역에 셀이 없으면 가장 가까운 지역 선택
            candidate_cells = self._get_nearest_regional_cells(user_region)
        
        # 2단계: 건강한 셀만 필터링
        healthy_cells = [
            cell_id for cell_id in candidate_cells
            if self.cells[cell_id].status == CellStatus.HEALTHY
        ]
        
        if not healthy_cells:
            return None
        
        # 3단계: 부하 기반 셀 선택 (라운드 로빈 + 부하 고려)
        partition_key = self.get_partition_key(request)
        selected_cell_id = self._select_least_loaded_cell(healthy_cells, partition_key)
        
        return self.cells[selected_cell_id]
    
    def _get_nearest_regional_cells(self, user_region: str) -> List[str]:
        """가장 가까운 지역의 셀 목록 반환"""
        # 지역별 우선순위 매핑 (실제로는 지리적 거리 기반)
        region_priority = {
            'us-west': ['us-east', 'eu-west', 'ap-east'],
            'us-east': ['us-west', 'eu-west', 'ap-east'],
            'eu-west': ['us-east', 'us-west', 'ap-east'],
            'ap-east': ['us-west', 'us-east', 'eu-west']
        }
        
        fallback_regions = region_priority.get(user_region, list(self.regional_cells.keys()))
        
        for region in fallback_regions:
            if region in self.regional_cells:
                return self.regional_cells[region]
        
        return []
    
    def _select_least_loaded_cell(self, cell_ids: List[str], partition_key: str) -> str:
        """부하가 가장 낮은 셀 선택"""
        # 파티션 키 기반 일관성 해시 + 부하 고려
        cell_loads = [(cell_id, self.cells[cell_id].get_load_percentage()) 
                     for cell_id in cell_ids]
        
        # 부하가 80% 미만인 셀 중에서 선택
        available_cells = [cell_id for cell_id, load in cell_loads if load < 80]
        
        if not available_cells:
            # 모든 셀이 고부하 상태면 가장 낮은 부하의 셀 선택
            available_cells = [min(cell_loads, key=lambda x: x[1])[0]]
        
        # 파티션 키 기반 일관성 있는 선택
        hash_value = int(hashlib.md5(partition_key.encode()).hexdigest(), 16)
        selected_index = hash_value % len(available_cells)
        
        return available_cells[selected_index]
    
    async def route_request(self, request: VideoRequest) -> Dict:
        """요청 라우팅 및 처리"""
        # 셀 선택
        selected_cell = self.select_cell(request)
        
        if not selected_cell:
            return {
                'status': 'error',
                'message': '사용 가능한 셀이 없습니다',
                'timestamp': request.timestamp
            }
        
        # 선택된 셀로 요청 전달
        try:
            response = await self._forward_to_cell(selected_cell, request)
            
            # 셀 부하 업데이트
            self.cells[selected_cell.cell_id].current_load += 1
            
            return {
                'status': 'success',
                'cell_id': selected_cell.cell_id,
                'region': selected_cell.region,
                'response': response,
                'timestamp': request.timestamp
            }
            
        except Exception as e:
            # 셀 오류 발생 시 다른 셀로 재시도
            print(f"셀 {selected_cell.cell_id} 오류: {e}")
            return await self._retry_with_different_cell(request)
    
    async def _forward_to_cell(self, cell: Cell, request: VideoRequest) -> Dict:
        """실제 셀로 요청 전달 (시뮬레이션)"""
        # 실제 구현에서는 HTTP 요청을 보냄
        await asyncio.sleep(0.1)  # 네트워크 지연 시뮬레이션
        
        return {
            'video_url': f"{cell.endpoint}/video/{request.video_id}",
            'quality': request.quality,
            'cdn_endpoint': f"cdn-{cell.region}.example.com"
        }
    
    async def _retry_with_different_cell(self, request: VideoRequest) -> Dict:
        """다른 셀로 재시도"""
        # 재시도 로직 구현
        return {
            'status': 'error',
            'message': '셀 장애로 인한 요청 실패',
            'timestamp': request.timestamp
        }
    
    async def health_check(self):
        """주기적 헬스체크"""
        while True:
            for cell in self.cells.values():
                # 실제로는 HTTP 헬스체크 엔드포인트 호출
                if cell.get_load_percentage() > 95:
                    cell.status = CellStatus.UNHEALTHY
                    print(f"셀 {cell.cell_id} 과부하로 인한 비활성화")
                else:
                    cell.status = CellStatus.HEALTHY
            
            await asyncio.sleep(self.health_check_interval)

class VideoStreamingService:
    """비디오 스트리밍 서비스 메인 클래스"""
    
    def __init__(self):
        self.router = CellRouter()
        self.setup_cells()
    
    def setup_cells(self):
        """셀 초기화 및 등록"""
        # 다양한 지역의 셀 생성
        cells = [
            Cell("cell-us-west-1", "us-west", 1000, 0, CellStatus.HEALTHY, "https://us-west-1.streaming.example.com"),
            Cell("cell-us-west-2", "us-west", 1000, 0, CellStatus.HEALTHY, "https://us-west-2.streaming.example.com"),
            Cell("cell-us-east-1", "us-east", 1000, 0, CellStatus.HEALTHY, "https://us-east-1.streaming.example.com"),
            Cell("cell-eu-west-1", "eu-west", 800, 0, CellStatus.HEALTHY, "https://eu-west-1.streaming.example.com"),
            Cell("cell-ap-east-1", "ap-east", 600, 0, CellStatus.HEALTHY, "https://ap-east-1.streaming.example.com"),
        ]
        
        for cell in cells:
            self.router.register_cell(cell)
    
    async def handle_video_request(self, user_id: str, region: str, video_id: str, quality: str = "HD"):
        """비디오 스트리밍 요청 처리"""
        # 사용자 정보 생성
        user = User(user_id, region, "premium")
        
        # 비디오 요청 생성
        request = VideoRequest(user, video_id, quality, asyncio.get_event_loop().time())
        
        # 라우터를 통해 요청 처리
        response = await self.router.route_request(request)
        
        return response
    
    async def start_health_monitoring(self):
        """헬스 모니터링 시작"""
        await self.router.health_check()

# 사용 예시
async def main():
    """메인 실행 함수"""
    service = VideoStreamingService()
    
    # 헬스 모니터링 백그라운드 실행
    health_task = asyncio.create_task(service.start_health_monitoring())
    
    # 다양한 지역의 사용자 요청 시뮬레이션
    test_requests = [
        ("user_123", "us-west", "video_001", "4K"),
        ("user_456", "us-east", "video_002", "HD"),
        ("user_789", "eu-west", "video_003", "HD"),
        ("user_321", "ap-east", "video_001", "SD"),
        ("user_654", "us-west", "video_004", "4K"),
    ]
    
    print("=== 비디오 스트리밍 서비스 시작 ===")
    
    # 동시 요청 처리 테스트
    tasks = []
    for user_id, region, video_id, quality in test_requests:
        task = service.handle_video_request(user_id, region, video_id, quality)
        tasks.append(task)
    
    # 모든 요청 동시 처리
    responses = await asyncio.gather(*tasks)
    
    print("\n=== 요청 처리 결과 ===")
    for i, response in enumerate(responses):
        request_info = test_requests[i]
        print(f"사용자 {request_info[0]} ({request_info[1]}): {response['status']}")
        if response['status'] == 'success':
            print(f"  - 할당된 셀: {response['cell_id']}")
            print(f"  - 지역: {response['region']}")
            print(f"  - 비디오 URL: {response['response']['video_url']}")
    
    # 헬스 모니터링 종료
    health_task.cancel()

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# Cell 라우팅 시뮬레이션 예제 (Partition Key 기반)

# 각 사용자 그룹이 어떤 셀에 매핑되는지 정의
partition_map = {
    "us-east": ["cust_001", "cust_002"],
    "eu-west": ["cust_003", "cust_004"]
}

def get_cell_for_customer(customer_id: str) -> str:
    """
    고객 ID에 따른 셀 결정 함수
    """
    for cell, customers in partition_map.items():
        if customer_id in customers:
            return cell
    return "default"

# 예시 실행
customer_id = "cust_003"
target_cell = get_cell_for_customer(customer_id)
print(f"{customer_id}은/는 '{target_cell}' 셀에서 처리됩니다.")