Ambassador

1
2

Computer Science and Engineering
└─ Software Engineering > Design and Architecture > Architecture Patterns > Deployment and Scalability Patterns

 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

# 앰배서더 역할을 하는 프록시 클래스
class DatabaseAmbassador:
    def __init__(self, db_host, db_port):
        self.db_host = db_host
        self.db_port = db_port
        self.connection_pool = []  # 예시: 연결 풀 구현

    def connect(self):
        # 연결 풀에서 연결 가져오기, 없으면 새로 생성
        if not self.connection_pool:
            # 실제 DB 연결 코드 (예시)
            print(f"Connecting to {self.db_host}:{self.db_port}")
        return "connection_object"

    def execute_query(self, query, retries=3):
        for i in range(retries):
            try:
                conn = self.connect()
                # 실제 쿼리 실행 코드 (예시)
                print(f"Executing query: {query}")
                return "query_result"
            except Exception as e:
                print(f"Retry {i+1}: {e}")
        raise Exception("Query failed after retries")

# 서비스 클래스
class Service:
    def __init__(self, ambassador):
        self.ambassador = ambassador

    def get_data(self, query):
        return self.ambassador.execute_query(query)

# 사용 예시
ambassador = DatabaseAmbassador("localhost", 5432)
service = Service(ambassador)
result = service.get_data("SELECT * FROM users")
print(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
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

# sidecar.yaml: Ambassador Sidecar Deployment (Kubernetes Pod)
apiVersion: v1
kind: Pod
metadata:
  name: app-with-ambassador
spec:
  containers:
  - name: app
    image: myorg/myapp:latest
    ports:
    - containerPort: 50051
    env:
    - name: AMBASSADOR_ADDR
      value: "localhost:9900"
  - name: ambassador
    image: envoyproxy/envoy:v1.27
    ports:
    - containerPort: 9900
    volumeMounts:
    - name: envoy-config
      mountPath: /etc/envoy
  volumes:
  - name: envoy-config
    configMap:
      name: envoy-config

---
apiVersion: v1
kind: ConfigMap
metadata:
  name: envoy-config
data:
  envoy.yaml: |
    static_resources:
      listeners:
      - name: listener_9900
        address:
          socket_address: { address: 0.0.0.0, port_value: 9900 }
        filter_chains:
        - filters:
          - name: envoy.filters.network.http_connection_manager
            typed_config:
              "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
              stat_prefix: grpc_proxy
              route_config:
                name: local_route
                virtual_hosts:
                - name: backend
                  domains: ["*"]
                  routes:
                  - match: { prefix: "/" }
                    route: { cluster: "grpc_service" }
              http_filters:
              - name: envoy.filters.http.router
      clusters:
      - name: grpc_service
        connect_timeout: 0.25s
        type: STRICT_DNS
        lb_policy: ROUND_ROBIN
        load_assignment:
          cluster_name: grpc_service
          endpoints:
          - lb_endpoints:
            - endpoint: { address: { socket_address: { address: grpc-backend.default.svc.cluster.local, port_value: 50051 } } }

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

# client.py: Python gRPC Client
import grpc
import myapp_pb2, myapp_pb2_grpc

def run():
    channel = grpc.insecure_channel("localhost:9900")
    stub = myapp_pb2_grpc.MyServiceStub(channel)
    resp = stub.MyRPC(myapp_pb2.MyRequest(value="hello"))
    print("Response:", resp.value)

if __name__ == "__main__":
    run()

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

# Kubernetes Pod 설정
apiVersion: v1
kind: Pod
metadata:
  name: app-with-ambassador
spec:
  containers:
  - name: main-app
    image: myapp:latest
    ports:
    - containerPort: 8080
  - name: ambassador
    image: envoy:latest
    ports:
    - containerPort: 9901
    volumeMounts:
    - name: envoy-config
      mountPath: /etc/envoy
  volumes:
  - name: envoy-config
    configMap:
      name: envoy-config

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

# Nginx 기반 Ambassador 설정
upstream backend_service {
    server service1.example.com:8080 max_fails=3 fail_timeout=30s;
    server service2.example.com:8080 max_fails=3 fail_timeout=30s;
}

server {
    listen 80;
    location / {
        proxy_pass http://backend_service;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_connect_timeout 30s;
        proxy_send_timeout 30s;
        proxy_read_timeout 30s;
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

class AmbassadorClient:
    def __init__(self, config):
        self.config = config
        self.circuit_breaker = CircuitBreaker()
        self.metrics = MetricsCollector()
    
    def request(self, method, url, **kwargs):
        with self.metrics.timer('request_duration'):
            return self.circuit_breaker.call(
                self._make_request, method, url, **kwargs
            )

  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
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307

import asyncio
import aiohttp
import json
import time
import logging
from typing import Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum

# 설정 및 상수
class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"  
    HALF_OPEN = "half_open"

@dataclass
class AmbassadorConfig:
    """Ambassador 설정 클래스"""
    target_host: str
    target_port: int
    timeout: float = 30.0
    max_retries: int = 3
    circuit_breaker_threshold: int = 5
    circuit_breaker_timeout: int = 60
    enable_logging: bool = True
    enable_metrics: bool = True

class CircuitBreaker:
    """회로 차단기 구현"""
    
    def __init__(self, threshold: int, timeout: int):
        self.threshold = threshold
        self.timeout = timeout
        self.failure_count = 0
        self.state = CircuitState.CLOSED
        self.last_failure_time = 0
    
    def can_execute(self) -> bool:
        """요청 실행 가능 여부 확인"""
        if self.state == CircuitState.CLOSED:
            return True
        elif self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time >= self.timeout:
                self.state = CircuitState.HALF_OPEN
                return True
            return False
        else:  # HALF_OPEN
            return True
    
    def on_success(self):
        """성공 시 호출"""
        self.failure_count = 0
        self.state = CircuitState.CLOSED
    
    def on_failure(self):
        """실패 시 호출"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.threshold:
            self.state = CircuitState.OPEN

class MetricsCollector:
    """메트릭 수집기"""
    
    def __init__(self):
        self.metrics = {
            'requests_total': 0,
            'requests_success': 0,
            'requests_failed': 0,
            'response_times': [],
            'circuit_breaker_trips': 0
        }
    
    def record_request(self, success: bool, response_time: float):
        """요청 메트릭 기록"""
        self.metrics['requests_total'] += 1
        if success:
            self.metrics['requests_success'] += 1
        else:
            self.metrics['requests_failed'] += 1
        self.metrics['response_times'].append(response_time)
    
    def record_circuit_breaker_trip(self):
        """회로 차단기 작동 기록"""
        self.metrics['circuit_breaker_trips'] += 1
    
    def get_stats(self) -> Dict[str, Any]:
        """통계 정보 반환"""
        total_requests = self.metrics['requests_total']
        if total_requests == 0:
            return self.metrics
            
        avg_response_time = sum(self.metrics['response_times']) / len(self.metrics['response_times'])
        success_rate = (self.metrics['requests_success'] / total_requests) * 100
        
        return {
            **self.metrics,
            'success_rate_percent': round(success_rate, 2),
            'average_response_time': round(avg_response_time, 3)
        }

class PaymentAmbassador:
    """결제 서비스를 위한 Ambassador 구현"""
    
    def __init__(self, config: AmbassadorConfig):
        self.config = config
        self.circuit_breaker = CircuitBreaker(
            config.circuit_breaker_threshold,
            config.circuit_breaker_timeout
        )
        self.metrics = MetricsCollector()
        self.session: Optional[aiohttp.ClientSession] = None
        
        # 로깅 설정
        if config.enable_logging:
            logging.basicConfig(level=logging.INFO)
            self.logger = logging.getLogger(__name__)
        else:
            self.logger = logging.getLogger(__name__)
            self.logger.disabled = True
    
    async def __aenter__(self):
        """비동기 컨텍스트 매니저 진입"""
        self.session = aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=self.config.timeout)
        )
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """비동기 컨텍스트 매니저 종료"""
        if self.session:
            await self.session.close()
    
    async def process_payment(self, payment_data: Dict[str, Any]) -> Dict[str, Any]:
        """결제 요청 처리 (REST → SOAP 변환)"""
        
        # 회로 차단기 확인
        if not self.circuit_breaker.can_execute():
            self.metrics.record_circuit_breaker_trip()
            self.logger.warning("Circuit breaker is OPEN, rejecting request")
            raise Exception("Service temporarily unavailable")
        
        start_time = time.time()
        
        try:
            # REST 요청을 SOAP 형태로 변환
            soap_payload = self._convert_rest_to_soap(payment_data)
            
            # 레거시 시스템에 요청 전송
            response = await self._send_to_legacy_system(soap_payload)
            
            # SOAP 응답을 REST 형태로 변환
            rest_response = self._convert_soap_to_rest(response)
            
            # 성공 처리
            response_time = time.time() - start_time
            self.circuit_breaker.on_success()
            
            if self.config.enable_metrics:
                self.metrics.record_request(True, response_time)
            
            self.logger.info(f"Payment processed successfully in {response_time:.3f}s")
            return rest_response
            
        except Exception as e:
            # 실패 처리
            response_time = time.time() - start_time
            self.circuit_breaker.on_failure()
            
            if self.config.enable_metrics:
                self.metrics.record_request(False, response_time)
            
            self.logger.error(f"Payment processing failed: {str(e)}")
            raise
    
    def _convert_rest_to_soap(self, rest_data: Dict[str, Any]) -> str:
        """REST 데이터를 SOAP 형태로 변환"""
        soap_envelope = f"""
        <?xml version="1.0" encoding="UTF-8"?>
        <soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
                       xmlns:pay="http://legacy.payment.system/payment">
            <soap:Header>
                <pay:Authentication>
                    <pay:ApiKey>{self._get_legacy_api_key()}</pay:ApiKey>
                    <pay:Timestamp>{int(time.time())}</pay:Timestamp>
                </pay:Authentication>
            </soap:Header>
            <soap:Body>
                <pay:ProcessPayment>
                    <pay:Amount>{rest_data.get('amount', 0)}</pay:Amount>
                    <pay:Currency>{rest_data.get('currency', 'USD')}</pay:Currency>
                    <pay:CardNumber>{rest_data.get('card_number', '')}</pay:CardNumber>
                    <pay:ExpiryDate>{rest_data.get('expiry_date', '')}</pay:ExpiryDate>
                    <pay:CVV>{rest_data.get('cvv', '')}</pay:CVV>
                    <pay:MerchantId>{rest_data.get('merchant_id', '')}</pay:MerchantId>
                </pay:ProcessPayment>
            </soap:Body>
        </soap:Envelope>
        """
        return soap_envelope.strip()
    
    def _convert_soap_to_rest(self, soap_response: str) -> Dict[str, Any]:
        """SOAP 응답을 REST 형태로 변환"""
        # 실제 구현에서는 XML 파싱 라이브러리 사용
        # 여기서는 시뮬레이션을 위한 간단한 변환
        if "SUCCESS" in soap_response:
            return {
                "status": "success",
                "transaction_id": "TXN_" + str(int(time.time())),
                "amount": 100.0,
                "currency": "USD",
                "timestamp": time.time()
            }
        else:
            return {
                "status": "failed",
                "error_code": "PAYMENT_DECLINED",
                "error_message": "Payment was declined",
                "timestamp": time.time()
            }
    
    async def _send_to_legacy_system(self, soap_payload: str) -> str:
        """레거시 시스템에 SOAP 요청 전송"""
        url = f"http://{self.config.target_host}:{self.config.target_port}/payment/soap"
        headers = {
            'Content-Type': 'text/xml; charset=utf-8',
            'SOAPAction': '"http://legacy.payment.system/payment/ProcessPayment"'
        }
        
        # 재시도 로직
        for attempt in range(self.config.max_retries + 1):
            try:
                async with self.session.post(url, data=soap_payload, headers=headers) as response:
                    if response.status == 200:
                        return await response.text()
                    else:
                        raise aiohttp.ClientResponseError(
                            request_info=response.request_info,
                            history=response.history,
                            status=response.status
                        )
            except Exception as e:
                if attempt == self.config.max_retries:
                    raise
                # 지수 백오프로 재시도
                await asyncio.sleep(2 ** attempt)
                self.logger.warning(f"Retry attempt {attempt + 1} for payment request")
    
    def _get_legacy_api_key(self) -> str:
        """레거시 시스템 API 키 반환"""
        # 실제 구현에서는 보안 저장소에서 가져옴
        return "LEGACY_API_KEY_12345"
    
    def get_health_status(self) -> Dict[str, Any]:
        """Ambassador 상태 정보 반환"""
        return {
            "status": "healthy" if self.circuit_breaker.state == CircuitState.CLOSED else "degraded",
            "circuit_breaker_state": self.circuit_breaker.state.value,
            "failure_count": self.circuit_breaker.failure_count,
            "metrics": self.metrics.get_stats() if self.config.enable_metrics else {}
        }

# 사용 예시
async def main():
    """Ambassador 사용 예시"""
    
    # Ambassador 설정
    config = AmbassadorConfig(
        target_host="legacy-payment.internal",
        target_port=8080,
        timeout=30.0,
        max_retries=3,
        circuit_breaker_threshold=5,
        circuit_breaker_timeout=60,
        enable_logging=True,
        enable_metrics=True
    )
    
    # Ambassador 사용
    async with PaymentAmbassador(config) as ambassador:
        
        # 결제 요청 데이터
        payment_request = {
            "amount": 100.0,
            "currency": "USD",
            "card_number": "4111111111111111",
            "expiry_date": "12/25",
            "cvv": "123",
            "merchant_id": "MERCHANT_001"
        }
        
        try:
            # 결제 처리
            result = await ambassador.process_payment(payment_request)
            print(f"Payment result: {json.dumps(result, indent=2)}")
            
            # 상태 확인
            health = ambassador.get_health_status()
            print(f"Ambassador health: {json.dumps(health, indent=2)}")
            
        except Exception as e:
            print(f"Payment failed: {str(e)}")

# 실행
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
 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

import requests
import time
import logging
from typing import Any, Dict, Optional
from functools import wraps

class ServiceAmbassador:
    """
    Ambassador 패턴을 구현한 클래스
    외부 서비스와의 통신을 관리하고 다양한 cross-cutting concerns를 처리
    """
    def __init__(self, service_url: str, timeout: int = 5):
        self.service_url = service_url
        self.timeout = timeout
        self.retry_count = 3
        self.circuit_breaker_threshold = 5
        self.failure_count = 0
        self.last_failure_time = 0
        self.circuit_open = False
        
        # 로깅 설정
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)

    def circuit_breaker(func):
        """서킷 브레이커 패턴을 구현한 데코레이터"""
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            if self.circuit_open:
                # 서킷이 열려있는 경우 복구 시도
                if time.time() - self.last_failure_time > 60:  # 1분 후 복구 시도
                    self.logger.info("Attempting circuit recovery")
                    self.circuit_open = False
                    self.failure_count = 0
                else:
                    raise Exception("Circuit is open - service unavailable")

            try:
                result = func(self, *args, **kwargs)
                # 성공 시 실패 카운트 리셋
                self.failure_count = 0
                return result
            except Exception as e:
                self.failure_count += 1
                self.last_failure_time = time.time()
                
                # 실패 임계값 초과 시 서킷 오픈
                if self.failure_count >= self.circuit_breaker_threshold:
                    self.circuit_open = True
                    self.logger.warning("Circuit breaker opened due to multiple failures")
                
                raise e
        return wrapper

    def retry_mechanism(func):
        """재시도 메커니즘을 구현한 데코레이터"""
        @wraps(func)
        def wrapper(self, *args, **kwargs):
            for attempt in range(self.retry_count):
                try:
                    return func(self, *args, **kwargs)
                except Exception as e:
                    if attempt == self.retry_count - 1:
                        raise e
                    self.logger.warning(f"Request failed, retrying... (attempt {attempt + 1})")
                    time.sleep(2 ** attempt)  # 지수 백오프
            return None
        return wrapper

    @circuit_breaker
    @retry_mechanism
    def make_request(self, endpoint: str, method: str = 'GET', 
                    data: Optional[Dict] = None) -> Any:
        """외부 서비스에 요청을 보내는 메서드"""
        start_time = time.time()
        
        try:
            url = f"{self.service_url}/{endpoint}"
            response = requests.request(
                method=method,
                url=url,
                json=data,
                timeout=self.timeout
            )
            
            # 응답 시간 측정 및 로깅
            elapsed_time = time.time() - start_time
            self.logger.info(f"Request to {url} completed in {elapsed_time:.2f}s")
            
            # 응답 상태 확인
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.Timeout:
            self.logger.error(f"Request to {endpoint} timed out")
            raise
        except requests.exceptions.RequestException as e:
            self.logger.error(f"Request failed: {str(e)}")
            raise

# 사용 예시
class PaymentService:
    def __init__(self):
        self.ambassador = ServiceAmbassador("https://payment-api.example.com")
    
    def process_payment(self, payment_data: Dict) -> Dict:
        """결제 처리 메서드"""
        try:
            result = self.ambassador.make_request(
                endpoint="process",
                method="POST",
                data=payment_data
            )
            return result
        except Exception as e:
            logging.error(f"Payment processing failed: {str(e)}")
            raise

# 메트릭 수집을 위한 확장
class MetricsCollector:
    def __init__(self):
        self.metrics = {
            'request_count': 0,
            'error_count': 0,
            'total_response_time': 0
        }
    
    def record_request(self, response_time: float, success: bool):
        """요청 메트릭 기록"""
        self.metrics['request_count'] += 1
        self.metrics['total_response_time'] += response_time
        if not success:
            self.metrics['error_count'] += 1

    def get_metrics(self) -> Dict:
        """현재 메트릭 반환"""
        avg_response_time = (
            self.metrics['total_response_time'] / self.metrics['request_count']
            if self.metrics['request_count'] > 0 else 0
        )
        
        return {
            'request_count': self.metrics['request_count'],
            'error_rate': (
                self.metrics['error_count'] / self.metrics['request_count']
                if self.metrics['request_count'] > 0 else 0
            ),
            'average_response_time': avg_response_time
        }