微服务架构在云环境中的设计模式：构建可扩展、弹性的分布式系统

#!/usr/bin/env python3
"""
微服务拆分分析器
基于业务领域、数据依赖和团队结构分析服务拆分策略
"""

from typing import Dict, List, Set, Tuple
from dataclasses import dataclass
from enum import Enum
import json

class CouplingType(Enum):
    """耦合类型"""
    DATA = "data"
    FUNCTIONAL = "functional"
    TEMPORAL = "temporal"
    SPATIAL = "spatial"

@dataclass
class BusinessCapability:
    """业务能力"""
    name: str
    description: str
    data_entities: List[str]
    operations: List[str]
    dependencies: List[str]
    team_ownership: str
    complexity_score: int

@dataclass
class ServiceBoundary:
    """服务边界"""
    service_name: str
    capabilities: List[BusinessCapability]
    data_ownership: List[str]
    api_contracts: List[str]
    coupling_score: float

class MicroserviceDecomposer:
    """微服务拆分器"""
    
    def __init__(self):
        self.business_capabilities = []
        self.coupling_matrix = {}
        self.team_structure = {}
    
    def add_business_capability(self, capability: BusinessCapability):
        """添加业务能力"""
        self.business_capabilities.append(capability)
    
    def analyze_coupling(self) -> Dict[str, Dict[str, float]]:
        """分析服务间耦合度"""
        coupling_matrix = {}
        
        for cap1 in self.business_capabilities:
            coupling_matrix[cap1.name] = {}
            
            for cap2 in self.business_capabilities:
                if cap1.name == cap2.name:
                    coupling_matrix[cap1.name][cap2.name] = 0.0
                    continue
                
                # 计算数据耦合
                data_coupling = self._calculate_data_coupling(cap1, cap2)
                
                # 计算功能耦合
                functional_coupling = self._calculate_functional_coupling(cap1, cap2)
                
                # 计算团队耦合
                team_coupling = self._calculate_team_coupling(cap1, cap2)
                
                # 综合耦合度
                total_coupling = (data_coupling * 0.4 + 
                                functional_coupling * 0.4 + 
                                team_coupling * 0.2)
                
                coupling_matrix[cap1.name][cap2.name] = total_coupling
        
        return coupling_matrix
    
    def _calculate_data_coupling(self, cap1: BusinessCapability, 
                               cap2: BusinessCapability) -> float:
        """计算数据耦合度"""
        shared_entities = set(cap1.data_entities) & set(cap2.data_entities)
        total_entities = set(cap1.data_entities) | set(cap2.data_entities)
        
        if not total_entities:
            return 0.0
        
        return len(shared_entities) / len(total_entities)
    
    def _calculate_functional_coupling(self, cap1: BusinessCapability, 
                                     cap2: BusinessCapability) -> float:
        """计算功能耦合度"""
        # 检查依赖关系
        if cap2.name in cap1.dependencies or cap1.name in cap2.dependencies:
            return 0.8
        
        # 检查操作相似性
        shared_operations = set(cap1.operations) & set(cap2.operations)
        total_operations = set(cap1.operations) | set(cap2.operations)
        
        if not total_operations:
            return 0.0
        
        return len(shared_operations) / len(total_operations)
    
    def _calculate_team_coupling(self, cap1: BusinessCapability, 
                               cap2: BusinessCapability) -> float:
        """计算团队耦合度"""
        if cap1.team_ownership == cap2.team_ownership:
            return 0.2  # 同一团队，低耦合
        else:
            return 0.8  # 不同团队，高耦合
    
    def suggest_service_boundaries(self, max_coupling_threshold: float = 0.3) -> List[ServiceBoundary]:
        """建议服务边界"""
        coupling_matrix = self.analyze_coupling()
        service_boundaries = []
        processed_capabilities = set()
        
        for capability in self.business_capabilities:
            if capability.name in processed_capabilities:
                continue
            
            # 找到低耦合的相关能力
            related_capabilities = [capability]
            processed_capabilities.add(capability.name)
            
            for other_cap in self.business_capabilities:
                if (other_cap.name not in processed_capabilities and
                    coupling_matrix[capability.name][other_cap.name] <= max_coupling_threshold):
                    
                    related_capabilities.append(other_cap)
                    processed_capabilities.add(other_cap.name)
            
            # 创建服务边界
            service_boundary = self._create_service_boundary(related_capabilities)
            service_boundaries.append(service_boundary)
        
        return service_boundaries
    
    def _create_service_boundary(self, capabilities: List[BusinessCapability]) -> ServiceBoundary:
        """创建服务边界"""
        # 生成服务名称
        service_name = f"{capabilities[0].name.replace('_', '-')}-service"
        
        # 收集数据所有权
        data_ownership = []
        for cap in capabilities:
            data_ownership.extend(cap.data_entities)
        data_ownership = list(set(data_ownership))
        
        # 生成API契约
        api_contracts = []
        for cap in capabilities:
            for operation in cap.operations:
                api_contracts.append(f"/{cap.name.lower()}/{operation}")
        
        # 计算平均耦合度
        coupling_score = sum(cap.complexity_score for cap in capabilities) / len(capabilities)
        
        return ServiceBoundary(
            service_name=service_name,
            capabilities=capabilities,
            data_ownership=data_ownership,
            api_contracts=api_contracts,
            coupling_score=coupling_score
        )
    
    def generate_decomposition_report(self) -> Dict:
        """生成拆分报告"""
        service_boundaries = self.suggest_service_boundaries()
        coupling_matrix = self.analyze_coupling()
        
        report = {
            "decomposition_summary": {
                "total_capabilities": len(self.business_capabilities),
                "suggested_services": len(service_boundaries),
                "average_coupling": self._calculate_average_coupling(coupling_matrix)
            },
            "service_boundaries": [],
            "coupling_analysis": coupling_matrix,
            "recommendations": self._generate_recommendations(service_boundaries)
        }
        
        for boundary in service_boundaries:
            report["service_boundaries"].append({
                "service_name": boundary.service_name,
                "capabilities": [cap.name for cap in boundary.capabilities],
                "data_ownership": boundary.data_ownership,
                "api_contracts": boundary.api_contracts,
                "coupling_score": boundary.coupling_score
            })
        
        return report
    
    def _calculate_average_coupling(self, coupling_matrix: Dict) -> float:
        """计算平均耦合度"""
        total_coupling = 0
        count = 0
        
        for service1, couplings in coupling_matrix.items():
            for service2, coupling in couplings.items():
                if service1 != service2:
                    total_coupling += coupling
                    count += 1
        
        return total_coupling / count if count > 0 else 0.0
    
    def _generate_recommendations(self, service_boundaries: List[ServiceBoundary]) -> List[str]:
        """生成建议"""
        recommendations = []
        
        # 检查服务数量
        if len(service_boundaries) > 15:
            recommendations.append("服务数量较多，考虑进一步合并相关服务")
        elif len(service_boundaries) < 3:
            recommendations.append("服务数量较少，可能存在过度聚合")
        
        # 检查耦合度
        high_coupling_services = [b for b in service_boundaries if b.coupling_score > 0.7]
        if high_coupling_services:
            recommendations.append(f"以下服务耦合度较高，建议重新设计: {[s.service_name for s in high_coupling_services]}")
        
        # 检查数据所有权
        data_conflicts = self._check_data_ownership_conflicts(service_boundaries)
        if data_conflicts:
            recommendations.append(f"存在数据所有权冲突: {data_conflicts}")
        
        return recommendations
    
    def _check_data_ownership_conflicts(self, service_boundaries: List[ServiceBoundary]) -> List[str]:
        """检查数据所有权冲突"""
        data_ownership_map = {}
        conflicts = []
        
        for boundary in service_boundaries:
            for data_entity in boundary.data_ownership:
                if data_entity in data_ownership_map:
                    conflicts.append(f"{data_entity} 被多个服务拥有: {data_ownership_map[data_entity]}, {boundary.service_name}")
                else:
                    data_ownership_map[data_entity] = boundary.service_name
        
        return conflicts

def main():
    """主函数 - 演示微服务拆分分析"""
    decomposer = MicroserviceDecomposer()
    
    # 定义业务能力
    user_management = BusinessCapability(
        name="user_management",
        description="用户管理",
        data_entities=["User", "Profile", "Authentication"],
        operations=["create_user", "update_profile", "authenticate"],
        dependencies=[],
        team_ownership="user_team",
        complexity_score=3
    )
    
    order_processing = BusinessCapability(
        name="order_processing",
        description="订单处理",
        data_entities=["Order", "OrderItem", "OrderStatus"],
        operations=["create_order", "update_order", "cancel_order"],
        dependencies=["user_management", "product_catalog", "payment_processing"],
        team_ownership="order_team",
        complexity_score=5
    )
    
    product_catalog = BusinessCapability(
        name="product_catalog",
        description="商品目录",
        data_entities=["Product", "Category", "Inventory"],
        operations=["add_product", "update_inventory", "search_products"],
        dependencies=[],
        team_ownership="product_team",
        complexity_score=4
    )
    
    payment_processing = BusinessCapability(
        name="payment_processing",
        description="支付处理",
        data_entities=["Payment", "Transaction", "PaymentMethod"],
        operations=["process_payment", "refund", "validate_payment"],
        dependencies=["user_management"],
        team_ownership="payment_team",
        complexity_score=6
    )
    
    notification_service = BusinessCapability(
        name="notification_service",
        description="通知服务",
        data_entities=["Notification", "Template", "DeliveryLog"],
        operations=["send_email", "send_sms", "push_notification"],
        dependencies=["user_management"],
        team_ownership="platform_team",
        complexity_score=2
    )
    
    # 添加业务能力
    for capability in [user_management, order_processing, product_catalog, 
                      payment_processing, notification_service]:
        decomposer.add_business_capability(capability)
    
    # 生成拆分报告
    print("生成微服务拆分报告...")
    report = decomposer.generate_decomposition_report()
    
    print(f"\n=== 微服务拆分分析报告 ===")
    print(f"总业务能力数: {report['decomposition_summary']['total_capabilities']}")
    print(f"建议服务数: {report['decomposition_summary']['suggested_services']}")
    print(f"平均耦合度: {report['decomposition_summary']['average_coupling']:.2f}")
    
    print(f"\n=== 建议的服务边界 ===")
    for service in report['service_boundaries']:
        print(f"\n服务名: {service['service_name']}")
        print(f"  业务能力: {service['capabilities']}")
        print(f"  数据所有权: {service['data_ownership']}")
        print(f"  耦合度评分: {service['coupling_score']:.2f}")
    
    print(f"\n=== 优化建议 ===")
    for recommendation in report['recommendations']:
        print(f"- {recommendation}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
微服务同步通信实现
包括HTTP REST API、gRPC等同步通信模式
"""

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

class CommunicationProtocol(Enum):
    """通信协议"""
    HTTP_REST = "http_rest"
    GRPC = "grpc"
    GRAPHQL = "graphql"

@dataclass
class ServiceEndpoint:
    """服务端点"""
    service_name: str
    host: str
    port: int
    protocol: CommunicationProtocol
    health_check_path: str = "/health"

@dataclass
class RequestContext:
    """请求上下文"""
    correlation_id: str
    user_id: Optional[str] = None
    tenant_id: Optional[str] = None
    timeout: int = 30
    retry_count: int = 3

class CircuitBreakerState(Enum):
    """熔断器状态"""
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

class CircuitBreaker:
    """熔断器实现"""
    
    def __init__(self, failure_threshold: int = 5, 
                 recovery_timeout: int = 60,
                 success_threshold: int = 3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.state = CircuitBreakerState.CLOSED
    
    def can_execute(self) -> bool:
        """检查是否可以执行请求"""
        if self.state == CircuitBreakerState.CLOSED:
            return True
        elif self.state == CircuitBreakerState.OPEN:
            if (time.time() - self.last_failure_time) > self.recovery_timeout:
                self.state = CircuitBreakerState.HALF_OPEN
                self.success_count = 0
                return True
            return False
        else:  # HALF_OPEN
            return True
    
    def record_success(self):
        """记录成功"""
        if self.state == CircuitBreakerState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = CircuitBreakerState.CLOSED
                self.failure_count = 0
        elif self.state == CircuitBreakerState.CLOSED:
            self.failure_count = 0
    
    def record_failure(self):
        """记录失败"""
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN

class ServiceCommunicator:
    """服务通信器"""
    
    def __init__(self):
        self.service_registry = {}
        self.circuit_breakers = {}
        self.session = None
        self.logger = logging.getLogger(__name__)
    
    async def __aenter__(self):
        """异步上下文管理器入口"""
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """异步上下文管理器出口"""
        if self.session:
            await self.session.close()
    
    def register_service(self, endpoint: ServiceEndpoint):
        """注册服务端点"""
        self.service_registry[endpoint.service_name] = endpoint
        self.circuit_breakers[endpoint.service_name] = CircuitBreaker()
    
    async def call_service(self, service_name: str, method: str, 
                          path: str, context: RequestContext,
                          data: Optional[Dict] = None) -> Dict[str, Any]:
        """调用服务"""
        if service_name not in self.service_registry:
            raise ValueError(f"Service {service_name} not registered")
        
        endpoint = self.service_registry[service_name]
        circuit_breaker = self.circuit_breakers[service_name]
        
        # 检查熔断器状态
        if not circuit_breaker.can_execute():
            raise Exception(f"Circuit breaker is OPEN for service {service_name}")
        
        try:
            if endpoint.protocol == CommunicationProtocol.HTTP_REST:
                result = await self._call_http_service(endpoint, method, path, context, data)
            elif endpoint.protocol == CommunicationProtocol.GRPC:
                result = await self._call_grpc_service(endpoint, method, path, context, data)
            else:
                raise ValueError(f"Unsupported protocol: {endpoint.protocol}")
            
            circuit_breaker.record_success()
            return result
            
        except Exception as e:
            circuit_breaker.record_failure()
            self.logger.error(f"Service call failed: {service_name}.{method} - {str(e)}")
            raise
    
    async def _call_http_service(self, endpoint: ServiceEndpoint, method: str,
                               path: str, context: RequestContext,
                               data: Optional[Dict] = None) -> Dict[str, Any]:
        """HTTP服务调用"""
        url = f"http://{endpoint.host}:{endpoint.port}{path}"
        
        headers = {
            'Content-Type': 'application/json',
            'X-Correlation-ID': context.correlation_id,
            'X-Request-Timeout': str(context.timeout)
        }
        
        if context.user_id:
            headers['X-User-ID'] = context.user_id
        if context.tenant_id:
            headers['X-Tenant-ID'] = context.tenant_id
        
        timeout = aiohttp.ClientTimeout(total=context.timeout)
        
        for attempt in range(context.retry_count):
            try:
                async with self.session.request(
                    method=method.upper(),
                    url=url,
                    headers=headers,
                    json=data,
                    timeout=timeout
                ) as response:
                    
                    if response.status == 200:
                        return await response.json()
                    elif response.status >= 500:
                        # 服务器错误，可以重试
                        if attempt < context.retry_count - 1:
                            await asyncio.sleep(2 ** attempt)  # 指数退避
                            continue
                    
                    # 客户端错误或最后一次重试失败
                    error_text = await response.text()
                    raise Exception(f"HTTP {response.status}: {error_text}")
                    
            except asyncio.TimeoutError:
                if attempt < context.retry_count - 1:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise Exception(f"Request timeout after {context.retry_count} attempts")
            except Exception as e:
                if attempt < context.retry_count - 1:
                    await asyncio.sleep(2 ** attempt)
                    continue
                raise
    
    async def _call_grpc_service(self, endpoint: ServiceEndpoint, method: str,
                               path: str, context: RequestContext,
                               data: Optional[Dict] = None) -> Dict[str, Any]:
        """gRPC服务调用"""
        # 这里是gRPC调用的示例实现
        # 实际实现需要根据具体的protobuf定义
        
        channel_address = f"{endpoint.host}:{endpoint.port}"
        
        try:
            async with grpc.aio.insecure_channel(channel_address) as channel:
                # 这里需要根据实际的gRPC服务定义来实现
                # stub = YourServiceStub(channel)
                # response = await stub.YourMethod(request, timeout=context.timeout)
                # return MessageToDict(response)
                
                # 示例返回
                return {"status": "success", "message": "gRPC call completed"}
                
        except grpc.RpcError as e:
            raise Exception(f"gRPC call failed: {e.code()} - {e.details()}")
    
    async def health_check(self, service_name: str) -> bool:
        """健康检查"""
        if service_name not in self.service_registry:
            return False
        
        endpoint = self.service_registry[service_name]
        
        try:
            context = RequestContext(
                correlation_id=f"health-check-{int(time.time())}",
                timeout=5,
                retry_count=1
            )
            
            await self.call_service(
                service_name=service_name,
                method="GET",
                path=endpoint.health_check_path,
                context=context
            )
            
            return True
            
        except Exception:
            return False
    
    async def batch_health_check(self) -> Dict[str, bool]:
        """批量健康检查"""
        health_status = {}
        
        tasks = []
        for service_name in self.service_registry.keys():
            task = asyncio.create_task(self.health_check(service_name))
            tasks.append((service_name, task))
        
        for service_name, task in tasks:
            try:
                health_status[service_name] = await task
            except Exception:
                health_status[service_name] = False
        
        return health_status

async def main():
    """主函数 - 演示服务通信"""
    async with ServiceCommunicator() as communicator:
        
        # 注册服务端点
        user_service = ServiceEndpoint(
            service_name="user-service",
            host="localhost",
            port=8001,
            protocol=CommunicationProtocol.HTTP_REST
        )
        
        order_service = ServiceEndpoint(
            service_name="order-service",
            host="localhost",
            port=8002,
            protocol=CommunicationProtocol.HTTP_REST
        )
        
        communicator.register_service(user_service)
        communicator.register_service(order_service)
        
        # 创建请求上下文
        context = RequestContext(
            correlation_id="req-12345",
            user_id="user-001",
            timeout=30
        )
        
        try:
            # 调用用户服务
            print("调用用户服务...")
            user_response = await communicator.call_service(
                service_name="user-service",
                method="GET",
                path="/users/user-001",
                context=context
            )
            print(f"用户服务响应: {user_response}")
            
            # 调用订单服务
            print("调用订单服务...")
            order_data = {
                "user_id": "user-001",
                "items": [
                    {"product_id": "prod-001", "quantity": 2},
                    {"product_id": "prod-002", "quantity": 1}
                ]
            }
            
            order_response = await communicator.call_service(
                service_name="order-service",
                method="POST",
                path="/orders",
                context=context,
                data=order_data
            )
            print(f"订单服务响应: {order_response}")
            
        except Exception as e:
            print(f"服务调用失败: {str(e)}")
        
        # 批量健康检查
        print("\n执行健康检查...")
        health_status = await communicator.batch_health_check()
        
        for service_name, is_healthy in health_status.items():
            status = "健康" if is_healthy else "不健康"
            print(f"{service_name}: {status}")

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

#!/bin/bash
"""
Istio服务网格部署脚本
包括安装、配置和示例应用部署
"""

set -e

# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color

# 日志函数
log_info() {
    echo -e "${GREEN}[INFO]${NC} $1"
}

log_warn() {
    echo -e "${YELLOW}[WARN]${NC} $1"
}

log_error() {
    echo -e "${RED}[ERROR]${NC} $1"
}

# 检查依赖
check_dependencies() {
    log_info "检查依赖..."
    
    if ! command -v kubectl &> /dev/null; then
        log_error "kubectl 未安装"
        exit 1
    fi
    
    if ! command -v curl &> /dev/null; then
        log_error "curl 未安装"
        exit 1
    fi
    
    # 检查Kubernetes集群连接
    if ! kubectl cluster-info &> /dev/null; then
        log_error "无法连接到Kubernetes集群"
        exit 1
    fi
    
    log_info "依赖检查完成"
}

# 安装Istio
install_istio() {
    log_info "安装Istio..."
    
    # 下载Istio
    ISTIO_VERSION="1.19.0"
    curl -L https://istio.io/downloadIstio | ISTIO_VERSION=$ISTIO_VERSION sh -
    
    # 添加到PATH
    export PATH="$PWD/istio-$ISTIO_VERSION/bin:$PATH"
    
    # 安装Istio
    istioctl install --set values.defaultRevision=default -y
    
    # 启用自动注入
    kubectl label namespace default istio-injection=enabled
    
    log_info "Istio安装完成"
}

# 部署Istio网关
deploy_gateway() {
    log_info "部署Istio网关..."
    
    cat <<EOF | kubectl apply -f -
apiVersion: networking.istio.io/v1beta1
kind: Gateway
metadata:
  name: microservices-gateway
  namespace: default
spec:
  selector:
    istio: ingressgateway
  servers:
  - port:
      number: 80
      name: http
      protocol: HTTP
    hosts:
    - "*"
  - port:
      number: 443
      name: https
      protocol: HTTPS
    tls:
      mode: SIMPLE
      credentialName: microservices-tls
    hosts:
    - "*.example.com"
---
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: microservices-vs
  namespace: default
spec:
  hosts:
  - "*"
  gateways:
  - microservices-gateway
  http:
  - match:
    - uri:
        prefix: /api/users
    route:
    - destination:
        host: user-service
        port:
          number: 8080
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 5s
    retries:
      attempts: 3
      perTryTimeout: 2s
  - match:
    - uri:
        prefix: /api/orders
    route:
    - destination:
        host: order-service
        port:
          number: 8080
      weight: 80
    - destination:
        host: order-service-v2
        port:
          number: 8080
      weight: 20
    timeout: 10s
  - match:
    - uri:
        prefix: /api/payments
    route:
    - destination:
        host: payment-service
        port:
          number: 8080
    headers:
      request:
        add:
          x-request-id: "%REQ(x-request-id)%"
          x-trace-id: "%REQ(x-trace-id)%"
EOF

    log_info "Istio网关部署完成"
}

# 配置流量管理
configure_traffic_management() {
    log_info "配置流量管理..."
    
    # 目标规则
    cat <<EOF | kubectl apply -f -
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: user-service-dr
  namespace: default
spec:
  host: user-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 50
        maxRequestsPerConnection: 10
    loadBalancer:
      simple: LEAST_CONN
    outlierDetection:
      consecutiveErrors: 3
      interval: 30s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
  subsets:
  - name: v1
    labels:
      version: v1
  - name: v2
    labels:
      version: v2
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: order-service-dr
  namespace: default
spec:
  host: order-service
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 50
      http:
        http1MaxPendingRequests: 25
        maxRequestsPerConnection: 5
    loadBalancer:
      simple: ROUND_ROBIN
    circuitBreaker:
      consecutiveErrors: 5
      interval: 10s
      baseEjectionTime: 30s
      maxEjectionPercent: 50
      minHealthPercent: 30
EOF

    log_info "流量管理配置完成"
}

# 配置安全策略
configure_security() {
    log_info "配置安全策略..."
    
    # 认证策略
    cat <<EOF | kubectl apply -f -
apiVersion: security.istio.io/v1beta1
kind: PeerAuthentication
metadata:
  name: default
  namespace: default
spec:
  mtls:
    mode: STRICT
---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: user-service-authz
  namespace: default
spec:
  selector:
    matchLabels:
      app: user-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/default/sa/api-gateway"]
  - to:
    - operation:
        methods: ["GET", "POST"]
        paths: ["/api/users/*"]
  - when:
    - key: request.headers[authorization]
      values: ["Bearer *"]
---
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: order-service-authz
  namespace: default
spec:
  selector:
    matchLabels:
      app: order-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/default/sa/api-gateway"]
    - source:
        principals: ["cluster.local/ns/default/sa/user-service"]
  - to:
    - operation:
        methods: ["GET", "POST", "PUT", "DELETE"]
        paths: ["/api/orders/*"]
  - when:
    - key: request.headers[x-user-id]
      notValues: [""]
EOF

    log_info "安全策略配置完成"
}

# 部署监控组件
deploy_monitoring() {
    log_info "部署监控组件..."
    
    # 安装Prometheus
    kubectl apply -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/prometheus.yaml
    
    # 安装Grafana
    kubectl apply -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/grafana.yaml
    
    # 安装Jaeger
    kubectl apply -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/jaeger.yaml
    
    # 安装Kiali
    kubectl apply -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/kiali.yaml
    
    # 等待部署完成
    kubectl wait --for=condition=available --timeout=300s deployment/prometheus -n istio-system
    kubectl wait --for=condition=available --timeout=300s deployment/grafana -n istio-system
    kubectl wait --for=condition=available --timeout=300s deployment/jaeger -n istio-system
    kubectl wait --for=condition=available --timeout=300s deployment/kiali -n istio-system
    
    log_info "监控组件部署完成"
}

# 部署示例应用
deploy_sample_apps() {
    log_info "部署示例应用..."
    
    # 用户服务
    cat <<EOF | kubectl apply -f -
apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-service
  labels:
    app: user-service
    version: v1
spec:
  replicas: 2
  selector:
    matchLabels:
      app: user-service
      version: v1
  template:
    metadata:
      labels:
        app: user-service
        version: v1
    spec:
      containers:
      - name: user-service
        image: nginx:alpine
        ports:
        - containerPort: 80
        env:
        - name: SERVICE_NAME
          value: "user-service"
        - name: VERSION
          value: "v1"
---
apiVersion: v1
kind: Service
metadata:
  name: user-service
  labels:
    app: user-service
spec:
  ports:
  - port: 8080
    targetPort: 80
    name: http
  selector:
    app: user-service
EOF

    # 订单服务
    cat <<EOF | kubectl apply -f -
apiVersion: apps/v1
kind: Deployment
metadata:
  name: order-service
  labels:
    app: order-service
    version: v1
spec:
  replicas: 2
  selector:
    matchLabels:
      app: order-service
      version: v1
  template:
    metadata:
      labels:
        app: order-service
        version: v1
    spec:
      containers:
      - name: order-service
        image: nginx:alpine
        ports:
        - containerPort: 80
        env:
        - name: SERVICE_NAME
          value: "order-service"
        - name: VERSION
          value: "v1"
---
apiVersion: v1
kind: Service
metadata:
  name: order-service
  labels:
    app: order-service
spec:
  ports:
  - port: 8080
    targetPort: 80
    name: http
  selector:
    app: order-service
EOF

    log_info "示例应用部署完成"
}

# 验证部署
verify_deployment() {
    log_info "验证部署..."
    
    # 检查Istio组件
    kubectl get pods -n istio-system
    
    # 检查应用
    kubectl get pods -n default
    
    # 检查网关
    kubectl get gateway,virtualservice,destinationrule -n default
    
    # 获取Ingress Gateway地址
    INGRESS_HOST=$(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.status.loadBalancer.ingress[0].ip}')
    INGRESS_PORT=$(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.spec.ports[?(@.name=="http2")].port}')
    
    if [ -z "$INGRESS_HOST" ]; then
        INGRESS_HOST=$(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.status.loadBalancer.ingress[0].hostname}')
    fi
    
    if [ -z "$INGRESS_HOST" ]; then
        INGRESS_HOST="localhost"
        INGRESS_PORT=$(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.spec.ports[?(@.name=="http2")].nodePort}')
    fi
    
    GATEWAY_URL="$INGRESS_HOST:$INGRESS_PORT"
    
    log_info "网关地址: http://$GATEWAY_URL"
    log_info "Kiali仪表板: kubectl port-forward -n istio-system svc/kiali 20001:20001"
    log_info "Grafana仪表板: kubectl port-forward -n istio-system svc/grafana 3000:3000"
    log_info "Jaeger仪表板: kubectl port-forward -n istio-system svc/jaeger 16686:16686"
    
    log_info "部署验证完成"
}

# 清理资源
cleanup() {
    log_info "清理资源..."
    
    # 删除应用
    kubectl delete deployment,service user-service order-service 2>/dev/null || true
    
    # 删除Istio配置
    kubectl delete gateway,virtualservice,destinationrule,peerauthentication,authorizationpolicy --all 2>/dev/null || true
    
    # 删除监控组件
    kubectl delete -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/prometheus.yaml 2>/dev/null || true
    kubectl delete -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/grafana.yaml 2>/dev/null || true
    kubectl delete -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/jaeger.yaml 2>/dev/null || true
    kubectl delete -f https://raw.githubusercontent.com/istio/istio/release-1.19/samples/addons/kiali.yaml 2>/dev/null || true
    
    # 卸载Istio
    istioctl uninstall --purge -y 2>/dev/null || true
    
    log_info "清理完成"
}

# 主函数
main() {
    case "${1:-deploy}" in
        "deploy")
            check_dependencies
            install_istio
            deploy_gateway
            configure_traffic_management
            configure_security
            deploy_monitoring
            deploy_sample_apps
            verify_deployment
            ;;
        "cleanup")
            cleanup
            ;;
        "verify")
            verify_deployment
            ;;
        *)
            echo "用法: $0 [deploy|cleanup|verify]"
            echo "  deploy  - 部署Istio服务网格"
            echo "  cleanup - 清理所有资源"
            echo "  verify  - 验证部署状态"
            exit 1
            ;;
    esac
}

main "$@"

#!/usr/bin/env python3
"""
微服务分布式监控系统
包括指标收集、日志聚合、链路追踪和告警
"""

import asyncio
import aiohttp
import json
import time
import logging
import uuid
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
from datetime import datetime, timedelta
from enum import Enum
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
import jaeger_client
from jaeger_client import Config
import structlog
from elasticsearch import AsyncElasticsearch
import redis.asyncio as redis
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

class MetricType(Enum):
    """指标类型"""
    COUNTER = "counter"
    GAUGE = "gauge"
    HISTOGRAM = "histogram"
    SUMMARY = "summary"

class LogLevel(Enum):
    """日志级别"""
    DEBUG = "debug"
    INFO = "info"
    WARNING = "warning"
    ERROR = "error"
    CRITICAL = "critical"

@dataclass
class MetricData:
    """指标数据"""
    name: str
    type: MetricType
    value: float
    labels: Dict[str, str]
    timestamp: datetime
    help_text: str = ""

@dataclass
class LogEntry:
    """日志条目"""
    timestamp: datetime
    level: LogLevel
    service: str
    message: str
    trace_id: Optional[str] = None
    span_id: Optional[str] = None
    user_id: Optional[str] = None
    request_id: Optional[str] = None
    extra_fields: Dict[str, Any] = None

@dataclass
class TraceSpan:
    """追踪跨度"""
    trace_id: str
    span_id: str
    parent_span_id: Optional[str]
    operation_name: str
    service_name: str
    start_time: datetime
    end_time: Optional[datetime]
    duration: Optional[float]
    tags: Dict[str, Any]
    logs: List[Dict[str, Any]]
    status: str = "ok"

class MetricsCollector:
    """指标收集器"""
    
    def __init__(self):
        self.registry = CollectorRegistry()
        self.metrics = {}
        
        # 预定义指标
        self._setup_default_metrics()
    
    def _setup_default_metrics(self):
        """设置默认指标"""
        # HTTP请求指标
        self.http_requests_total = Counter(
            'http_requests_total',
            'Total HTTP requests',
            ['method', 'endpoint', 'status_code', 'service'],
            registry=self.registry
        )
        
        self.http_request_duration = Histogram(
            'http_request_duration_seconds',
            'HTTP request duration',
            ['method', 'endpoint', 'service'],
            registry=self.registry
        )
        
        # 业务指标
        self.active_users = Gauge(
            'active_users_total',
            'Number of active users',
            ['service'],
            registry=self.registry
        )
        
        self.database_connections = Gauge(
            'database_connections_active',
            'Active database connections',
            ['database', 'service'],
            registry=self.registry
        )
        
        # 系统指标
        self.memory_usage = Gauge(
            'memory_usage_bytes',
            'Memory usage in bytes',
            ['service'],
            registry=self.registry
        )
        
        self.cpu_usage = Gauge(
            'cpu_usage_percent',
            'CPU usage percentage',
            ['service'],
            registry=self.registry
        )
    
    def record_http_request(self, method: str, endpoint: str, 
                          status_code: int, service: str, duration: float):
        """记录HTTP请求指标"""
        self.http_requests_total.labels(
            method=method,
            endpoint=endpoint,
            status_code=str(status_code),
            service=service
        ).inc()
        
        self.http_request_duration.labels(
            method=method,
            endpoint=endpoint,
            service=service
        ).observe(duration)
    
    def set_active_users(self, count: int, service: str):
        """设置活跃用户数"""
        self.active_users.labels(service=service).set(count)
    
    def set_database_connections(self, count: int, database: str, service: str):
        """设置数据库连接数"""
        self.database_connections.labels(
            database=database,
            service=service
        ).set(count)
    
    def set_memory_usage(self, bytes_used: int, service: str):
        """设置内存使用量"""
        self.memory_usage.labels(service=service).set(bytes_used)
    
    def set_cpu_usage(self, percentage: float, service: str):
        """设置CPU使用率"""
        self.cpu_usage.labels(service=service).set(percentage)
    
    def get_metrics(self) -> str:
        """获取Prometheus格式的指标"""
        return prometheus_client.generate_latest(self.registry).decode('utf-8')

class DistributedTracer:
    """分布式追踪器"""
    
    def __init__(self, service_name: str, jaeger_endpoint: str = "http://localhost:14268/api/traces"):
        self.service_name = service_name
        
        # 配置Jaeger
        config = Config(
            config={
                'sampler': {
                    'type': 'const',
                    'param': 1,
                },
                'logging': True,
                'reporter_batch_size': 1,
            },
            service_name=service_name,
            validate=True,
        )
        
        self.tracer = config.initialize_tracer()
        
        # 配置OpenTelemetry
        trace.set_tracer_provider(TracerProvider())
        jaeger_exporter = JaegerExporter(
            agent_host_name="localhost",
            agent_port=6831,
        )
        
        span_processor = BatchSpanProcessor(jaeger_exporter)
        trace.get_tracer_provider().add_span_processor(span_processor)
        
        self.otel_tracer = trace.get_tracer(__name__)
    
    def start_span(self, operation_name: str, parent_span=None, tags: Dict[str, Any] = None) -> Any:
        """开始一个新的跨度"""
        span = self.tracer.start_span(
            operation_name=operation_name,
            child_of=parent_span,
            tags=tags or {}
        )
        
        return span
    
    def finish_span(self, span: Any, tags: Dict[str, Any] = None, logs: List[Dict[str, Any]] = None):
        """结束跨度"""
        if tags:
            for key, value in tags.items():
                span.set_tag(key, value)
        
        if logs:
            for log_entry in logs:
                span.log_kv(log_entry)
        
        span.finish()
    
    def extract_span_context(self, headers: Dict[str, str]) -> Any:
        """从HTTP头中提取跨度上下文"""
        try:
            span_ctx = self.tracer.extract(
                format=jaeger_client.Format.HTTP_HEADERS,
                carrier=headers
            )
            return span_ctx
        except Exception:
            return None
    
    def inject_span_context(self, span: Any, headers: Dict[str, str]) -> Dict[str, str]:
        """将跨度上下文注入HTTP头"""
        self.tracer.inject(
            span_context=span.context,
            format=jaeger_client.Format.HTTP_HEADERS,
            carrier=headers
        )
        return headers

class LogAggregator:
    """日志聚合器"""
    
    def __init__(self, elasticsearch_url: str = "http://localhost:9200"):
        self.es_client = AsyncElasticsearch([elasticsearch_url])
        self.logger = structlog.get_logger()
        
        # 配置结构化日志
        structlog.configure(
            processors=[
                structlog.stdlib.filter_by_level,
                structlog.stdlib.add_logger_name,
                structlog.stdlib.add_log_level,
                structlog.stdlib.PositionalArgumentsFormatter(),
                structlog.processors.TimeStamper(fmt="iso"),
                structlog.processors.StackInfoRenderer(),
                structlog.processors.format_exc_info,
                structlog.processors.UnicodeDecoder(),
                structlog.processors.JSONRenderer()
            ],
            context_class=dict,
            logger_factory=structlog.stdlib.LoggerFactory(),
            wrapper_class=structlog.stdlib.BoundLogger,
            cache_logger_on_first_use=True,
        )
    
    async def log(self, entry: LogEntry):
        """记录日志"""
        # 发送到Elasticsearch
        await self._send_to_elasticsearch(entry)
        
        # 本地日志
        self._log_locally(entry)
    
    async def _send_to_elasticsearch(self, entry: LogEntry):
        """发送日志到Elasticsearch"""
        try:
            index_name = f"microservices-logs-{entry.timestamp.strftime('%Y.%m.%d')}"
            
            doc = {
                'timestamp': entry.timestamp.isoformat(),
                'level': entry.level.value,
                'service': entry.service,
                'message': entry.message,
                'trace_id': entry.trace_id,
                'span_id': entry.span_id,
                'user_id': entry.user_id,
                'request_id': entry.request_id,
                **entry.extra_fields or {}
            }
            
            await self.es_client.index(
                index=index_name,
                body=doc
            )
            
        except Exception as e:
            print(f"Failed to send log to Elasticsearch: {e}")
    
    def _log_locally(self, entry: LogEntry):
        """本地日志记录"""
        logger = self.logger.bind(
            service=entry.service,
            trace_id=entry.trace_id,
            span_id=entry.span_id,
            user_id=entry.user_id,
            request_id=entry.request_id,
            **(entry.extra_fields or {})
        )
        
        if entry.level == LogLevel.DEBUG:
            logger.debug(entry.message)
        elif entry.level == LogLevel.INFO:
            logger.info(entry.message)
        elif entry.level == LogLevel.WARNING:
            logger.warning(entry.message)
        elif entry.level == LogLevel.ERROR:
            logger.error(entry.message)
        elif entry.level == LogLevel.CRITICAL:
            logger.critical(entry.message)
    
    async def search_logs(self, query: str, service: str = None, 
                         start_time: datetime = None, end_time: datetime = None,
                         limit: int = 100) -> List[Dict[str, Any]]:
        """搜索日志"""
        try:
            search_body = {
                "query": {
                    "bool": {
                        "must": []
                    }
                },
                "sort": [
                    {"timestamp": {"order": "desc"}}
                ],
                "size": limit
            }
            
            # 添加查询条件
            if query:
                search_body["query"]["bool"]["must"].append({
                    "multi_match": {
                        "query": query,
                        "fields": ["message", "service"]
                    }
                })
            
            if service:
                search_body["query"]["bool"]["must"].append({
                    "term": {"service": service}
                })
            
            if start_time or end_time:
                time_range = {}
                if start_time:
                    time_range["gte"] = start_time.isoformat()
                if end_time:
                    time_range["lte"] = end_time.isoformat()
                
                search_body["query"]["bool"]["must"].append({
                    "range": {"timestamp": time_range}
                })
            
            response = await self.es_client.search(
                index="microservices-logs-*",
                body=search_body
            )
            
            return [hit["_source"] for hit in response["hits"]["hits"]]
            
        except Exception as e:
            print(f"Failed to search logs: {e}")
            return []

class AlertManager:
    """告警管理器"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.redis_client = redis.from_url(redis_url)
        self.alert_rules = []
        self.notification_channels = []
    
    def add_alert_rule(self, name: str, condition: str, threshold: float,
                      duration: int = 300, severity: str = "warning"):
        """添加告警规则"""
        rule = {
            'name': name,
            'condition': condition,
            'threshold': threshold,
            'duration': duration,
            'severity': severity,
            'last_triggered': None
        }
        self.alert_rules.append(rule)
    
    def add_notification_channel(self, channel_type: str, config: Dict[str, Any]):
        """添加通知渠道"""
        channel = {
            'type': channel_type,
            'config': config
        }
        self.notification_channels.append(channel)
    
    async def check_alerts(self, metrics: Dict[str, float]):
        """检查告警条件"""
        current_time = datetime.utcnow()
        
        for rule in self.alert_rules:
            try:
                # 简化的条件检查
                if self._evaluate_condition(rule['condition'], metrics, rule['threshold']):
                    # 检查是否需要触发告警
                    if await self._should_trigger_alert(rule, current_time):
                        await self._trigger_alert(rule, metrics)
                        rule['last_triggered'] = current_time
                        
            except Exception as e:
                print(f"Error checking alert rule {rule['name']}: {e}")
    
    def _evaluate_condition(self, condition: str, metrics: Dict[str, float], threshold: float) -> bool:
        """评估告警条件"""
        # 简化实现，实际应该支持更复杂的表达式
        if condition in metrics:
            value = metrics[condition]
            return value > threshold
        return False
    
    async def _should_trigger_alert(self, rule: Dict[str, Any], current_time: datetime) -> bool:
        """检查是否应该触发告警"""
        if rule['last_triggered'] is None:
            return True
        
        time_since_last = (current_time - rule['last_triggered']).total_seconds()
        return time_since_last >= rule['duration']
    
    async def _trigger_alert(self, rule: Dict[str, Any], metrics: Dict[str, float]):
        """触发告警"""
        alert = {
            'id': str(uuid.uuid4()),
            'rule_name': rule['name'],
            'severity': rule['severity'],
            'message': f"Alert: {rule['name']} triggered",
            'timestamp': datetime.utcnow().isoformat(),
            'metrics': metrics
        }
        
        # 存储告警
        await self.redis_client.lpush('alerts', json.dumps(alert))
        
        # 发送通知
        for channel in self.notification_channels:
            await self._send_notification(channel, alert)
    
    async def _send_notification(self, channel: Dict[str, Any], alert: Dict[str, Any]):
        """发送通知"""
        try:
            if channel['type'] == 'webhook':
                await self._send_webhook_notification(channel['config'], alert)
            elif channel['type'] == 'email':
                await self._send_email_notification(channel['config'], alert)
            elif channel['type'] == 'slack':
                await self._send_slack_notification(channel['config'], alert)
                
        except Exception as e:
            print(f"Failed to send notification via {channel['type']}: {e}")
    
    async def _send_webhook_notification(self, config: Dict[str, Any], alert: Dict[str, Any]):
        """发送Webhook通知"""
        async with aiohttp.ClientSession() as session:
            await session.post(config['url'], json=alert)
    
    async def _send_email_notification(self, config: Dict[str, Any], alert: Dict[str, Any]):
        """发送邮件通知"""
        # 实现邮件发送逻辑
        pass
    
    async def _send_slack_notification(self, config: Dict[str, Any], alert: Dict[str, Any]):
        """发送Slack通知"""
        # 实现Slack通知逻辑
        pass

class MonitoringSystem:
    """监控系统"""
    
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.metrics_collector = MetricsCollector()
        self.tracer = DistributedTracer(service_name)
        self.log_aggregator = LogAggregator()
        self.alert_manager = AlertManager()
        
        # 设置默认告警规则
        self._setup_default_alerts()
    
    def _setup_default_alerts(self):
        """设置默认告警规则"""
        self.alert_manager.add_alert_rule(
            name="high_error_rate",
            condition="error_rate",
            threshold=0.05,  # 5%
            duration=300,    # 5分钟
            severity="critical"
        )
        
        self.alert_manager.add_alert_rule(
            name="high_response_time",
            condition="avg_response_time",
            threshold=2.0,   # 2秒
            duration=600,    # 10分钟
            severity="warning"
        )
        
        self.alert_manager.add_alert_rule(
            name="high_memory_usage",
            condition="memory_usage_percent",
            threshold=80.0,  # 80%
            duration=300,    # 5分钟
            severity="warning"
        )
    
    async def start_monitoring(self):
        """启动监控"""
        # 启动指标收集循环
        asyncio.create_task(self._metrics_collection_loop())
        
        # 启动告警检查循环
        asyncio.create_task(self._alert_check_loop())
    
    async def _metrics_collection_loop(self):
        """指标收集循环"""
        while True:
            try:
                # 收集系统指标
                await self._collect_system_metrics()
                
                # 等待下一次收集
                await asyncio.sleep(30)  # 每30秒收集一次
                
            except Exception as e:
                await self.log_aggregator.log(LogEntry(
                    timestamp=datetime.utcnow(),
                    level=LogLevel.ERROR,
                    service=self.service_name,
                    message=f"Metrics collection failed: {str(e)}"
                ))
                await asyncio.sleep(10)
    
    async def _alert_check_loop(self):
        """告警检查循环"""
        while True:
            try:
                # 获取当前指标
                metrics = await self._get_current_metrics()
                
                # 检查告警
                await self.alert_manager.check_alerts(metrics)
                
                # 等待下一次检查
                await asyncio.sleep(60)  # 每分钟检查一次
                
            except Exception as e:
                await self.log_aggregator.log(LogEntry(
                    timestamp=datetime.utcnow(),
                    level=LogLevel.ERROR,
                    service=self.service_name,
                    message=f"Alert check failed: {str(e)}"
                ))
                await asyncio.sleep(30)
    
    async def _collect_system_metrics(self):
        """收集系统指标"""
        import psutil
        
        # CPU使用率
        cpu_percent = psutil.cpu_percent(interval=1)
        self.metrics_collector.set_cpu_usage(cpu_percent, self.service_name)
        
        # 内存使用量
        memory = psutil.virtual_memory()
        self.metrics_collector.set_memory_usage(memory.used, self.service_name)
    
    async def _get_current_metrics(self) -> Dict[str, float]:
        """获取当前指标"""
        import psutil
        
        return {
            'cpu_usage_percent': psutil.cpu_percent(),
            'memory_usage_percent': psutil.virtual_memory().percent,
            'disk_usage_percent': psutil.disk_usage('/').percent,
        }

# 使用示例
async def main():
    """主函数 - 监控系统示例"""
    monitoring = MonitoringSystem("user-service")
    
    # 启动监控
    await monitoring.start_monitoring()
    
    # 模拟一些指标
    monitoring.metrics_collector.record_http_request(
        method="GET",
        endpoint="/api/users",
        status_code=200,
        service="user-service",
        duration=0.5
    )
    
    # 记录日志
    await monitoring.log_aggregator.log(LogEntry(
        timestamp=datetime.utcnow(),
        level=LogLevel.INFO,
        service="user-service",
        message="User service started successfully",
        extra_fields={"version": "1.0.0"}
    ))
    
    print("监控系统启动成功")
    print("指标端点: http://localhost:8080/metrics")
    print("日志查询: http://localhost:9200/microservices-logs-*/_search")
    
    # 保持运行
    await asyncio.Event().wait()

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