无服务器函数

无服务器函数技能

概述

无服务器函数是现代云原生应用的核心组件。不当的函数设计会导致冷启动延迟、资源浪费和性能问题。在设计无服务器架构前需要仔细分析业务需求。

核心原则: 好的无服务器函数应该快速启动、高效执行、自动扩展，同时保证成本可控。坏的函数设计会导致高延迟、高成本和维护困难。

何时使用

始终:

• 设计事件驱动架构时
• 实现微服务后端逻辑时
• 处理异步任务和工作流时
• 构建API网关和端点时
• 优化云原生应用架构时

触发短语:

• "无服务器函数设计"
• "Lambda函数优化"
• "冷启动问题解决"
• "函数扩展策略"
• "事件驱动架构"
• "Serverless最佳实践"

无服务器函数功能

函数架构设计

• 函数拆分策略
• 事件源配置
• 触发器设置
• 错误处理机制
• 状态管理方案

性能优化

• 冷启动优化
• 内存配置调优
• 并发控制策略
• 超时时间设置
• 资源利用率分析

监控和调试

• 函数执行监控
• 性能指标收集
• 错误日志分析
• 调试工具配置
• 告警规则设置

部署和管理

• 函数版本管理
• 环境配置管理
• 自动化部署流程
• 测试策略制定
• 安全配置检查

常见无服务器函数问题

冷启动延迟

问题:
函数冷启动时间过长，影响用户体验

错误示例:
- 函数包体积过大
- 依赖库过多
- 初始化逻辑复杂
- 运行时选择不当

解决方案:
1. 优化函数包大小，移除不必要依赖
2. 使用预热策略保持函数活跃
3. 选择合适的运行时和内存配置
4. 简化初始化逻辑

内存和并发限制

问题:
函数内存不足或并发达到上限

错误示例:
- 内存配置过低
- 并发数设置不当
- 资源泄漏未处理
- 长时间运行任务

解决方案:
1. 根据实际需求调整内存配置
2. 设置合理的并发限制
3. 优化代码避免资源泄漏
4. 拆分长时间任务为多个函数

错误处理不当

问题:
函数错误处理机制不完善，导致任务失败

错误示例:
- 缺少异常捕获
- 重试策略不当
- 死信队列未配置
- 监控告警缺失

解决方案:
1. 完善异常处理和错误捕获
2. 配置合适的重试策略
3. 设置死信队列处理失败消息
4. 建立完善的监控告警

代码实现示例

无服务器函数分析器

import json
import time
import boto3
import zipfile
import os
from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from collections import defaultdict

@dataclass
class FunctionMetrics:
    """函数指标"""
    name: str
    runtime: str
    memory_size: int
    timeout: int
    code_size: int
    cold_start_time: float
    average_duration: float
    error_rate: float
    invocation_count: int
    cost_per_month: float

@dataclass
class FunctionIssue:
    """函数问题"""
    severity: str  # critical, high, medium, low
    type: str
    function_name: str
    message: str
    suggestion: str
    affected_metrics: List[str]

class ServerlessFunctionAnalyzer:
    def __init__(self, region: str = 'us-east-1'):
        self.region = region
        self.lambda_client = boto3.client('lambda', region_name=region)
        self.cloudwatch_client = boto3.client('cloudwatch', region_name=region)
        self.issues: List[FunctionIssue] = []
        
    def analyze_functions(self, function_names: List[str] = None) -> Dict[str, Any]:
        """分析Lambda函数"""
        try:
            if function_names is None:
                # 获取所有函数
                functions = self.lambda_client.list_functions()
                function_names = [f['FunctionName'] for f in functions['Functions']]
            
            function_metrics = []
            
            for function_name in function_names:
                metrics = self.analyze_function(function_name)
                if metrics:
                    function_metrics.append(metrics)
            
            # 生成分析报告
            report = {
                'total_functions': len(function_metrics),
                'functions': [self.metrics_to_dict(m) for m in function_metrics],
                'issues': self.issues,
                'cost_analysis': self.analyze_costs(function_metrics),
                'performance_analysis': self.analyze_performance(function_metrics),
                'recommendations': self.generate_recommendations(function_metrics)
            }
            
            return report
            
        except Exception as e:
            return {'error': f'分析失败: {e}'}
    
    def analyze_function(self, function_name: str) -> Optional[FunctionMetrics]:
        """分析单个函数"""
        try:
            # 获取函数配置
            function_config = self.lambda_client.get_function_configuration(
                FunctionName=function_name
            )
            
            # 获取函数统计信息
            function_stats = self.get_function_statistics(function_name)
            
            # 计算冷启动时间
            cold_start_time = self.measure_cold_start_time(function_name)
            
            # 分析函数代码
            code_analysis = self.analyze_function_code(function_name)
            
            # 计算月度成本
            monthly_cost = self.calculate_monthly_cost(function_config, function_stats)
            
            metrics = FunctionMetrics(
                name=function_name,
                runtime=function_config['Runtime'],
                memory_size=function_config['MemorySize'],
                timeout=function_config['Timeout'],
                code_size=function_config['CodeSize'],
                cold_start_time=cold_start_time,
                average_duration=function_stats.get('average_duration', 0),
                error_rate=function_stats.get('error_rate', 0),
                invocation_count=function_stats.get('invocation_count', 0),
                cost_per_month=monthly_cost
            )
            
            # 检查函数问题
            self.check_function_issues(metrics, code_analysis)
            
            return metrics
            
        except Exception as e:
            print(f'分析函数 {function_name} 失败: {e}')
            return None
    
    def get_function_statistics(self, function_name: str) -> Dict[str, float]:
        """获取函数统计信息"""
        try:
            # 获取过去30天的CloudWatch指标
            end_time = time.time()
            start_time = end_time - 30 * 24 * 60 * 60  # 30天前
            
            # 获取平均执行时间
            duration_response = self.cloudwatch_client.get_metric_statistics(
                Namespace='AWS/Lambda',
                MetricName='Duration',
                Dimensions=[{'Name': 'FunctionName', 'Value': function_name}],
                StartTime=start_time,
                EndTime=end_time,
                Period=24 * 60 * 60,  # 1天
                Statistics=['Average']
            )
            
            # 获取错误率
            errors_response = self.cloudwatch_client.get_metric_statistics(
                Namespace='AWS/Lambda',
                MetricName='Errors',
                Dimensions=[{'Name': 'FunctionName', 'Value': function_name}],
                StartTime=start_time,
                EndTime=end_time,
                Period=24 * 60 * 60,
                Statistics=['Sum']
            )
            
            # 获取调用次数
            invocations_response = self.cloudwatch_client.get_metric_statistics(
                Namespace='AWS/Lambda',
                MetricName='Invocations',
                Dimensions=[{'Name': 'FunctionName', 'Value': function_name}],
                StartTime=start_time,
                EndTime=end_time,
                Period=24 * 60 * 60,
                Statistics=['Sum']
            )
            
            # 计算统计数据
            avg_duration = 0
            if duration_response['Datapoints']:
                avg_duration = sum(d['Average'] for d in duration_response['Datapoints']) / len(duration_response['Datapoints'])
            
            total_errors = sum(d['Sum'] for d in errors_response['Datapoints'])
            total_invocations = sum(d['Sum'] for d in invocations_response['Datapoints'])
            
            error_rate = (total_errors / total_invocations * 100) if total_invocations > 0 else 0
            
            return {
                'average_duration': avg_duration,
                'error_rate': error_rate,
                'invocation_count': int(total_invocations),
                'total_errors': int(total_errors)
            }
            
        except Exception as e:
            print(f'获取函数统计信息失败: {e}')
            return {}
    
    def measure_cold_start_time(self, function_name: str) -> float:
        """测量冷启动时间"""
        try:
            # 通过多次调用测量冷启动时间
            cold_start_times = []
            
            for i in range(3):
                start_time = time.time()
                
                # 调用函数
                response = self.lambda_client.invoke(
                    FunctionName=function_name,
                    Payload=json.dumps({'test': 'cold_start'})
                )
                
                end_time = time.time()
                cold_start_times.append(end_time - start_time)
                
                # 等待一段时间确保函数冷却
                time.sleep(60)
            
            # 返回平均冷启动时间
            return sum(cold_start_times) / len(cold_start_times) if cold_start_times else 0
            
        except Exception as e:
            print(f'测量冷启动时间失败: {e}')
            return 0
    
    def analyze_function_code(self, function_name: str) -> Dict[str, Any]:
        """分析函数代码"""
        try:
            # 获取函数代码
            response = self.lambda_client.get_function(FunctionName=function_name)
            code_location = response['Code']['Location']
            
            # 下载代码包
            import urllib.request
            with urllib.request.urlopen(code_location) as response:
                code_zip = response.read()
            
            # 分析ZIP包
            analysis = self.analyze_zip_package(code_zip)
            
            return analysis
            
        except Exception as e:
            print(f'分析函数代码失败: {e}')
            return {}
    
    def analyze_zip_package(self, zip_data: bytes) -> Dict[str, Any]:
        """分析ZIP包"""
        try:
            with zipfile.ZipFile(io.BytesIO(zip_data)) as zip_file:
                files = zip_file.namelist()
                
                analysis = {
                    'file_count': len(files),
                    'total_size': len(zip_data),
                    'file_types': defaultdict(int),
                    'large_files': [],
                    'dependencies': []
                }
                
                for file in files:
                    # 统计文件类型
                    ext = os.path.splitext(file)[1].lower()
                    analysis['file_types'][ext] += 1
                    
                    # 检查大文件
                    file_info = zip_file.getinfo(file)
                    if file_info.file_size > 1024 * 1024:  # 1MB
                        analysis['large_files'].append({
                            'name': file,
                            'size': file_info.file_size
                        })
                    
                    # 检查依赖文件
                    if file.endswith('requirements.txt') or file.endswith('package.json'):
                        analysis['dependencies'].append(file)
            
            return analysis
            
        except Exception as e:
            print(f'分析ZIP包失败: {e}')
            return {}
    
    def calculate_monthly_cost(self, function_config: Dict, function_stats: Dict) -> float:
        """计算月度成本"""
        try:
            # AWS Lambda定价 (示例价格，实际价格可能变化)
            # 免费额度：100万次请求，400,000 GB-秒计算时间
            # 超出免费额度：$0.20 per 1M requests, $0.0000166667 per GB-second
            
            memory_mb = function_config['MemorySize']
            avg_duration_ms = function_stats.get('average_duration', 0) * 1000
            invocations = function_stats.get('invocation_count', 0)
            
            # 计算计算时间 (GB-秒)
            compute_time_gb_sec = (memory_mb / 1024) * (avg_duration_ms / 1000) * invocations
            
            # 计算请求成本
            request_cost = max(0, invocations - 1000000) * 0.20 / 1000000
            
            # 计算计算时间成本
            compute_cost = max(0, compute_time_gb_sec - 400000) * 0.0000166667
            
            total_cost = request_cost + compute_cost
            
            return total_cost
            
        except Exception as e:
            print(f'计算成本失败: {e}')
            return 0
    
    def check_function_issues(self, metrics: FunctionMetrics, code_analysis: Dict) -> None:
        """检查函数问题"""
        # 检查冷启动时间
        if metrics.cold_start_time > 5.0:
            self.issues.append(FunctionIssue(
                severity='high',
                type='performance',
                function_name=metrics.name,
                message=f'冷启动时间过长: {metrics.cold_start_time:.2f}秒',
                suggestion='优化函数包大小，使用预热策略或选择更小的运行时',
                affected_metrics=['cold_start_time', 'performance']
            ))
        
        # 检查内存配置
        if metrics.memory_size > 1024:
            self.issues.append(FunctionIssue(
                severity='medium',
                type='cost',
                function_name=metrics.name,
                message=f'内存配置过高: {metrics.memory_size}MB',
                suggestion='根据实际使用情况调整内存配置以降低成本',
                affected_metrics=['memory_size', 'cost_per_month']
            ))
        
        # 检查错误率
        if metrics.error_rate > 5.0:
            self.issues.append(FunctionIssue(
                severity='high',
                type='reliability',
                function_name=metrics.name,
                message=f'错误率过高: {metrics.error_rate:.2f}%',
                suggestion='检查代码逻辑，增加错误处理和重试机制',
                affected_metrics=['error_rate', 'reliability']
            ))
        
        # 检查超时配置
        if metrics.timeout > 300:
            self.issues.append(FunctionIssue(
                severity='medium',
                type='configuration',
                function_name=metrics.name,
                message=f'超时时间过长: {metrics.timeout}秒',
                suggestion'检查函数逻辑，考虑拆分长时间运行的任务',
                affected_metrics=['timeout', 'performance']
            ))
        
        # 检查代码包大小
        if code_analysis.get('total_size', 0) > 50 * 1024 * 1024:  # 50MB
            self.issues.append(FunctionIssue(
                severity='medium',
                type='performance',
                function_name=metrics.name,
                message=f'代码包过大: {code_analysis["total_size"] / 1024 / 1024:.2f}MB',
                suggestion='移除不必要的依赖，使用Lambda Layers',
                affected_metrics=['code_size', 'cold_start_time']
            ))
        
        # 检查调用频率
        if metrics.invocation_count == 0:
            self.issues.append(FunctionIssue(
                severity='low',
                type='usage',
                function_name=metrics.name,
                message='函数未被调用',
                suggestion='检查触发器配置或考虑删除未使用的函数',
                affected_metrics=['invocation_count', 'cost_per_month']
            ))
    
    def analyze_costs(self, function_metrics: List[FunctionMetrics]) -> Dict[str, Any]:
        """分析成本"""
        total_cost = sum(m.cost_per_month for m in function_metrics)
        most_expensive = max(function_metrics, key=lambda m: m.cost_per_month)
        least_expensive = min(function_metrics, key=lambda m: m.cost_per_month)
        
        cost_distribution = defaultdict(int)
        for metric in function_metrics:
            cost_range = self.get_cost_range(metric.cost_per_month)
            cost_distribution[cost_range] += 1
        
        return {
            'total_monthly_cost': total_cost,
            'average_cost_per_function': total_cost / len(function_metrics) if function_metrics else 0,
            'most_expensive_function': most_expensive.name,
            'most_expensive_cost': most_expensive.cost_per_month,
            'least_expensive_function': least_expensive.name,
            'least_expensive_cost': least_expensive.cost_per_month,
            'cost_distribution': dict(cost_distribution)
        }
    
    def get_cost_range(self, cost: float) -> str:
        """获取成本范围"""
        if cost < 1:
            return '< $1'
        elif cost < 10:
            return '$1-$10'
        elif cost < 50:
            return '$10-$50'
        elif cost < 100:
            return '$50-$100'
        else:
            return '> $100'
    
    def analyze_performance(self, function_metrics: List[FunctionMetrics]) -> Dict[str, Any]:
        """分析性能"""
        if not function_metrics:
            return {}
        
        avg_cold_start = sum(m.cold_start_time for m in function_metrics) / len(function_metrics)
        avg_duration = sum(m.average_duration for m in function_metrics) / len(function_metrics)
        avg_memory = sum(m.memory_size for m in function_metrics) / len(function_metrics)
        
        # 找出性能最差的函数
        slowest_cold_start = max(function_metrics, key=lambda m: m.cold_start_time)
        longest_duration = max(function_metrics, key=lambda m: m.average_duration)
        
        return {
            'average_cold_start_time': avg_cold_start,
            'average_duration': avg_duration,
            'average_memory_size': avg_memory,
            'slowest_cold_start_function': slowest_cold_start.name,
            'slowest_cold_start_time': slowest_cold_start.cold_start_time,
            'longest_duration_function': longest_duration.name,
            'longest_duration': longest_duration.average_duration
        }
    
    def generate_recommendations(self, function_metrics: List[FunctionMetrics]) -> List[Dict[str, str]]:
        """生成优化建议"""
        recommendations = []
        
        # 基于问题生成建议
        issue_counts = defaultdict(int)
        for issue in self.issues:
            issue_counts[issue.type] += 1
        
        if issue_counts['performance'] > 0:
            recommendations.append({
                'priority': 'high',
                'type': 'performance',
                'message': f'{issue_counts["performance"]}个函数存在性能问题',
                'suggestion': '优化函数代码，减少冷启动时间，调整资源配置'
            })
        
        if issue_counts['cost'] > 0:
            recommendations.append({
                'priority': 'medium',
                'type': 'cost',
                'message': f'{issue_counts["cost"]}个函数成本过高',
                'suggestion': '调整内存配置，优化代码以减少计算时间'
            })
        
        if issue_counts['reliability'] > 0:
            recommendations.append({
                'priority': 'critical',
                'type': 'reliability',
                'message': f'{issue_counts["reliability"]}个函数可靠性问题',
                'suggestion': '增加错误处理，配置重试策略和死信队列'
            })
        
        # 通用建议
        total_cost = sum(m.cost_per_month for m in function_metrics)
        if total_cost > 100:
            recommendations.append({
                'priority': 'high',
                'type': 'cost',
                'message': '月度成本超过$100',
                'suggestion': '审查所有函数的使用情况，优化高成本函数'
            })
        
        avg_cold_start = sum(m.cold_start_time for m in function_metrics) / len(function_metrics) if function_metrics else 0
        if avg_cold_start > 3.0:
            recommendations.append({
                'priority': 'medium',
                'type': 'performance',
                'message': '平均冷启动时间过长',
                'suggestion': '使用预热策略，优化函数包大小'
            })
        
        return recommendations
    
    def metrics_to_dict(self, metrics: FunctionMetrics) -> Dict[str, Any]:
        """将指标转换为字典"""
        return {
            'name': metrics.name,
            'runtime': metrics.runtime,
            'memory_size': metrics.memory_size,
            'timeout': metrics.timeout,
            'code_size': metrics.code_size,
            'cold_start_time': metrics.cold_start_time,
            'average_duration': metrics.average_duration,
            'error_rate': metrics.error_rate,
            'invocation_count': metrics.invocation_count,
            'cost_per_month': metrics.cost_per_month
        }

# 无服务器函数优化器
class ServerlessFunctionOptimizer:
    def __init__(self, region: str = 'us-east-1'):
        self.region = region
        self.lambda_client = boto3.client('lambda', region_name=region)
        
    def optimize_function(self, function_name: str, optimization_type: str = 'auto') -> Dict[str, Any]:
        """优化函数配置"""
        try:
            # 获取当前配置
            current_config = self.lambda_client.get_function_configuration(
                FunctionName=function_name
            )
            
            # 获取函数统计信息
            analyzer = ServerlessFunctionAnalyzer(self.region)
            stats = analyzer.get_function_statistics(function_name)
            
            # 生成优化建议
            optimizations = self.generate_optimizations(current_config, stats, optimization_type)
            
            # 应用优化配置
            applied_optimizations = []
            for optimization in optimizations:
                if optimization.get('auto_apply', False):
                    result = self.apply_optimization(function_name, optimization)
                    applied_optimizations.append(result)
            
            return {
                'function_name': function_name,
                'current_config': current_config,
                'recommended_optimizations': optimizations,
                'applied_optimizations': applied_optimizations,
                'estimated_improvements': self.estimate_improvements(current_config, optimizations)
            }
            
        except Exception as e:
            return {'error': f'优化失败: {e}'}
    
    def generate_optimizations(self, current_config: Dict, stats: Dict, optimization_type: str) -> List[Dict[str, Any]]:
        """生成优化建议"""
        optimizations = []
        
        # 内存优化
        current_memory = current_config['MemorySize']
        avg_duration = stats.get('average_duration', 0)
        
        if optimization_type in ['auto', 'memory']:
            # 基于执行时间建议内存配置
            if avg_duration > 0:
                if current_memory > 512 and avg_duration < 1000:  # 执行时间短但内存高
                    optimizations.append({
                        'type': 'memory',
                        'priority': 'medium',
                        'message': '内存配置可能过高',
                        'suggestion': f'建议将内存从{current_memory}MB降低到256MB',
                        'current_value': current_memory,
                        'recommended_value': 256,
                        'auto_apply': False
                    })
                elif current_memory < 512 and avg_duration > 5000:  # 执行时间长但内存低
                    optimizations.append({
                        'type': 'memory',
                        'priority': 'high',
                        'message': '内存配置可能不足',
                        'suggestion': f'建议将内存从{current_memory}MB增加到1024MB',
                        'current_value': current_memory,
                        'recommended_value': 1024,
                        'auto_apply': False
                    })
        
        # 超时优化
        current_timeout = current_config['Timeout']
        if optimization_type in ['auto', 'timeout']:
            if current_timeout > 300 and avg_duration < current_timeout * 0.5:
                optimizations.append({
                    'type': 'timeout',
                    'priority': 'medium',
                    'message': '超时时间可能过长',
                    'suggestion': f'建议将超时时间从{current_timeout}秒减少到{int(avg_duration * 2)}秒',
                    'current_value': current_timeout,
                    'recommended_value': int(avg_duration * 2),
                    'auto_apply': False
                })
        
        # 环境变量优化
        env_vars = current_config.get('Environment', {}).get('Variables', {})
        if optimization_type in ['auto', 'environment']:
            # 检查是否有敏感信息
            sensitive_keys = ['password', 'secret', 'key', 'token']
            for key, value in env_vars.items():
                if any(sensitive in key.lower() for sensitive in sensitive_keys):
                    optimizations.append({
                        'type': 'security',
                        'priority': 'high',
                        'message': f'环境变量{key}包含敏感信息',
                        'suggestion': '使用AWS Secrets Manager或Parameter Store存储敏感信息',
                        'current_value': key,
                        'recommended_value': 'Use Secrets Manager',
                        'auto_apply': False
                    })
        
        # 并发优化
        if optimization_type in ['auto', 'concurrency']:
            reserved_concurrency = current_config.get('ReservedConcurrentExecutions')
            if not reserved_concurrency and stats.get('invocation_count', 0) > 10000:
                optimizations.append({
                    'type': 'concurrency',
                    'priority': 'medium',
                    'message': '高并发函数建议设置预留并发',
                    'suggestion': '设置ReservedConcurrentExecutions防止函数过载',
                    'current_value': 'Unlimited',
                    'recommended_value': 'Set appropriate limit',
                    'auto_apply': False
                })
        
        return optimizations
    
    def apply_optimization(self, function_name: str, optimization: Dict[str, Any]) -> Dict[str, Any]:
        """应用优化配置"""
        try:
            update_params = {}
            
            if optimization['type'] == 'memory':
                update_params['MemorySize'] = optimization['recommended_value']
            elif optimization['type'] == 'timeout':
                update_params['Timeout'] = optimization['recommended_value']
            else:
                return {'status': 'skipped', 'message': '此优化类型需要手动应用'}
            
            # 应用更新
            response = self.lambda_client.update_function_configuration(
                FunctionName=function_name,
                **update_params
            )
            
            return {
                'status': 'applied',
                'optimization_type': optimization['type'],
                'old_value': optimization['current_value'],
                'new_value': optimization['recommended_value'],
                'last_modified': response['LastModified']
            }
            
        except Exception as e:
            return {
                'status': 'failed',
                'optimization_type': optimization['type'],
                'error': str(e)
            }
    
    def estimate_improvements(self, current_config: Dict, optimizations: List[Dict[str, Any]]) -> Dict[str, Any]:
        """估算改进效果"""
        improvements = {
            'cost_reduction': 0,
            'performance_improvement': 0,
            'reliability_improvement': 0
        }
        
        for opt in optimizations:
            if opt['type'] == 'memory':
                old_memory = opt['current_value']
                new_memory = opt['recommended_value']
                if new_memory < old_memory:
                    # 估算成本节省
                    cost_reduction = (old_memory - new_memory) / old_memory * 20  # 最多20%节省
                    improvements['cost_reduction'] += cost_reduction
            
            elif opt['type'] == 'timeout':
                improvements['reliability_improvement'] += 10  # 超时优化提升可靠性
        
        return improvements

# 使用示例
def main():
    # 分析函数
    analyzer = ServerlessFunctionAnalyzer('us-east-1')
    analysis_result = analyzer.analyze_functions(['my-function', 'another-function'])
    
    print("无服务器函数分析结果:")
    print(f"总函数数: {analysis_result['total_functions']}")
    print(f"总月度成本: ${analysis_result['cost_analysis']['total_monthly_cost']:.2f}")
    print(f"平均冷启动时间: {analysis_result['performance_analysis']['average_cold_start_time']:.2f}秒")
    
    # 优化函数
    optimizer = ServerlessFunctionOptimizer('us-east-1')
    optimization_result = optimizer.optimize_function('my-function', 'auto')
    
    print(f"\n函数优化建议:")
    for opt in optimization_result['recommended_optimizations']:
        print(f"- {opt['message']}: {opt['suggestion']}")

if __name__ == '__main__':
    main()

无服务器函数部署器

import json
import zipfile
import boto3
import os
from typing import List, Dict, Any, Optional
from pathlib import Path

class ServerlessFunctionDeployer:
    def __init__(self, region: str = 'us-east-1'):
        self.region = region
        self.lambda_client = boto3.client('lambda', region_name=region)
        self.iam_client = boto3.client('iam', region_name=region)
        self.apigateway_client = boto3.client('apigateway', region_name=region)
        
    def deploy_function(self, function_config: Dict[str, Any]) -> Dict[str, Any]:
        """部署Lambda函数"""
        try:
            function_name = function_config['name']
            code_path = function_config['code_path']
            runtime = function_config['runtime']
            handler = function_config['handler']
            role_arn = function_config['role_arn']
            
            # 创建部署包
            deployment_package = self.create_deployment_package(code_path)
            
            # 检查函数是否存在
            try:
                existing_function = self.lambda_client.get_function(FunctionName=function_name)
                function_exists = True
            except self.lambda_client.exceptions.ResourceNotFoundException:
                function_exists = False
            
            # 创建或更新函数
            if function_exists:
                response = self.lambda_client.update_function_code(
                    FunctionName=function_name,
                    ZipFile=deployment_package
                )
                
                response = self.lambda_client.update_function_configuration(
                    FunctionName=function_name,
                    Runtime=runtime,
                    Role=role_arn,
                    Handler=handler,
                    **function_config.get('configuration', {})
                )
                
                action = 'updated'
            else:
                response = self.lambda_client.create_function(
                    FunctionName=function_name,
                    Runtime=runtime,
                    Role=role_arn,
                    Handler=handler,
                    Code={'ZipFile': deployment_package},
                    **function_config.get('configuration', {})
                )
                
                action = 'created'
            
            # 配置触发器
            if 'triggers' in function_config:
                self.configure_triggers(function_name, function_config['triggers'])
            
            return {
                'function_name': function_name,
                'action': action,
                'function_arn': response['FunctionArn'],
                'status': 'success'
            }
            
        except Exception as e:
            return {
                'function_name': function_config.get('name', 'unknown'),
                'action': 'failed',
                'error': str(e),
                'status': 'error'
            }
    
    def create_deployment_package(self, code_path: str) -> bytes:
        """创建部署包"""
        try:
            code_dir = Path(code_path)
            
            # 创建内存中的ZIP文件
            zip_buffer = io.BytesIO()
            
            with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
                # 添加所有Python文件
                for file_path in code_dir.rglob('*.py'):
                    arcname = file_path.relative_to(code_dir)
                    zip_file.write(file_path, arcname)
                
                # 添加requirements.txt中的依赖
                requirements_file = code_dir / 'requirements.txt'
                if requirements_file.exists():
                    zip_file.write(requirements_file, 'requirements.txt')
                
                # 添加其他必要文件
                for pattern in ['*.json', '*.yaml', '*.yml', '*.txt']:
                    for file_path in code_dir.rglob(pattern):
                        if file_path.name != 'requirements.txt':
                            arcname = file_path.relative_to(code_dir)
                            zip_file.write(file_path, arcname)
            
            zip_buffer.seek(0)
            return zip_buffer.getvalue()
            
        except Exception as e:
            raise Exception(f'创建部署包失败: {e}')
    
    def configure_triggers(self, function_name: str, triggers: List[Dict[str, Any]]) -> None:
        """配置触发器"""
        for trigger in triggers:
            trigger_type = trigger['type']
            
            if trigger_type == 'api_gateway':
                self.configure_api_gateway_trigger(function_name, trigger)
            elif trigger_type == 's3':
                self.configure_s3_trigger(function_name, trigger)
            elif trigger_type == 'dynamodb':
                self.configure_dynamodb_trigger(function_name, trigger)
            elif trigger_type == 'sns':
                self.configure_sns_trigger(function_name, trigger)
            elif trigger_type == 'schedule':
                self.configure_schedule_trigger(function_name, trigger)
    
    def configure_api_gateway_trigger(self, function_name: str, trigger: Dict[str, Any]) -> None:
        """配置API Gateway触发器"""
        try:
            api_name = trigger.get('api_name', f'{function_name}-api')
            stage = trigger.get('stage', 'prod')
            http_method = trigger.get('method', 'GET')
            path = trigger.get('path', '/')
            
            # 检查API是否存在
            apis = self.apigateway_client.get_rest_apis()
            api = next((api for api in apis['items'] if api['name'] == api_name), None)
            
            if not api:
                # 创建新的API
                api = self.apigateway_client.create_rest_api(name=api_name)
                api_id = api['id']
                
                # 获取根资源ID
                resources = self.apigateway_client.get_resources(restApiId=api_id)
                root_resource_id = resources['items'][0]['id']
                
                # 创建资源
                if path != '/':
                    resource = self.apigateway_client.create_resource(
                        restApiId=api_id,
                        parentId=root_resource_id,
                        pathPart=path.strip('/')
                    )
                    resource_id = resource['id']
                else:
                    resource_id = root_resource_id
                
                # 创建方法
                self.apigateway_client.put_method(
                    restApiId=api_id,
                    resourceId=resource_id,
                    httpMethod=http_method,
                    authorizationType='NONE'
                )
                
                # 设置集成
                lambda_arn = f'arn:aws:lambda:{self.region}:{boto3.client("sts").get_caller_identity()["Account"]}:function:{function_name}'
                
                self.apigateway_client.put_integration(
                    restApiId=api_id,
                    resourceId=resource_id,
                    httpMethod=http_method,
                    type='AWS_PROXY',
                    integrationHttpMethod=http_method,
                    uri=f'arn:aws:apigateway:{self.region}:lambda:path/2015-03-31/functions/{lambda_arn}/invocations'
                )
                
                # 部署API
                self.apigateway_client.create_deployment(
                    restApiId=api_id,
                    stageName=stage
                )
                
                print(f'创建API Gateway触发器: {api_name}')
            
        except Exception as e:
            print(f'配置API Gateway触发器失败: {e}')
    
    def configure_s3_trigger(self, function_name: str, trigger: Dict[str, Any]) -> None:
        """配置S3触发器"""
        try:
            bucket_name = trigger['bucket_name']
            events = trigger.get('events', ['s3:ObjectCreated:*'])
            filter_prefix = trigger.get('filter_prefix', '')
            filter_suffix = trigger.get('filter_suffix', '')
            
            # 添加S3 bucket通知配置
            s3_client = boto3.client('s3')
            lambda_arn = f'arn:aws:lambda:{self.region}:{boto3.client("sts").get_caller_identity()["Account"]}:function:{function_name}'
            
            # 获取当前通知配置
            try:
                notification_config = s3_client.get_bucket_notification_configuration(Bucket=bucket_name)
            except:
                notification_config = {'ResponseMetadata': {}, 'LambdaFunctionConfigurations': []}
            
            # 添加新的Lambda函数配置
            lambda_config = {
                'LambdaFunctionArn': lambda_arn,
                'Events': events
            }
            
            if filter_prefix:
                lambda_config['Filter'] = {
                    'Key': {
                        'FilterRules': [
                            {'Name': 'prefix', 'Value': filter_prefix}
                        ]
                    }
                }
            
            if filter_suffix:
                if 'Filter' not in lambda_config:
                    lambda_config['Filter'] = {'Key': {'FilterRules': []}}
                lambda_config['Filter']['Key']['FilterRules'].append(
                    {'Name': 'suffix', 'Value': filter_suffix}
                )
            
            notification_config['LambdaFunctionConfigurations'].append(lambda_config)
            
            # 更新bucket通知配置
            s3_client.put_bucket_notification_configuration(
                Bucket=bucket_name,
                NotificationConfiguration=notification_config
            )
            
            # 添加Lambda权限
            self.lambda_client.add_permission(
                FunctionName=function_name,
                StatementId=f's3-trigger-{bucket_name}',
                Action='lambda:InvokeFunction',
                Principal='s3.amazonaws.com',
                SourceArn=f'arn:aws:s3:::{bucket_name}'
            )
            
            print(f'配置S3触发器: {bucket_name}')
            
        except Exception as e:
            print(f'配置S3触发器失败: {e}')
    
    def configure_schedule_trigger(self, function_name: str, trigger: Dict[str, Any]) -> None:
        """配置定时触发器"""
        try:
            schedule_expression = trigger['schedule_expression']
            rule_name = trigger.get('rule_name', f'{function_name}-schedule')
            
            events_client = boto3.client('events')
            lambda_arn = f'arn:aws:lambda:{self.region}:{boto3.client("sts").get_caller_identity()["Account"]}:function:{function_name}'
            
            # 创建EventBridge规则
            events_client.put_rule(
                Name=rule_name,
                ScheduleExpression=schedule_expression,
                State='ENABLED'
            )
            
            # 添加Lambda目标
            events_client.put_targets(
                Rule=rule_name,
                Targets=[
                    {
                        'Id': '1',
                        'Arn': lambda_arn,
                        'RetryPolicy': {
                            'MaximumRetryAttempts': 0
                        }
                    }
                ]
            )
            
            # 添加Lambda权限
            self.lambda_client.add_permission(
                FunctionName=function_name,
                StatementId=f'events-trigger-{rule_name}',
                Action='lambda:InvokeFunction',
                Principal='events.amazonaws.com',
                SourceArn=f'arn:aws:events:{self.region}:{boto3.client("sts").get_caller_identity()["Account"]}:rule/{rule_name}'
            )
            
            print(f'配置定时触发器: {rule_name} ({schedule_expression})')
            
        except Exception as e:
            print(f'配置定时触发器失败: {e}')
    
    def deploy_functions_from_config(self, config_file: str) -> Dict[str, Any]:
        """从配置文件部署多个函数"""
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
            
            deployment_results = []
            
            for function_config in config['functions']:
                result = self.deploy_function(function_config)
                deployment_results.append(result)
            
            successful_deployments = [r for r in deployment_results if r['status'] == 'success']
            failed_deployments = [r for r in deployment_results if r['status'] == 'error']
            
            return {
                'total_functions': len(deployment_results),
                'successful_deployments': len(successful_deployments),
                'failed_deployments': len(failed_deployments),
                'results': deployment_results,
                'summary': {
                    'success_rate': len(successful_deployments) / len(de

…(truncated)