FinOps视角下基于代码剖析的性能瓶颈定位与成本优化

perf-cost-opt/
├── README.md
├── requirements.txt
├── config/
│   └── cost_config.yaml      # 云资源配置与单价
├── src/
│   ├── __init__.py
│   ├── agent/
│   │   ├── __init__.py
│   │   └── profiler.py       # 性能数据收集核心逻辑
│   ├── core/
│   │   ├── __init__.py
│   │   ├── cost_mapper.py    # 成本计算与映射
│   │   └── models.py         # 数据模型（Pydantic）
│   ├── analyzer/
│   │   ├── __init__.py
│   │   └── hotspot.py        # 热点分析与报告生成
│   ├── service/
│   │   ├── __init__.py
│   │   └── order_service.py  # 模拟的订单服务（含"问题"代码）
│   └── api/
│       ├── __init__.py
│       └── app.py            # Flask API 与数据看板
├── data/
│   └── profiler_data.db      # SQLite 存储性能数据（运行时生成）
├── scripts/
│   └── simulate_load.py      # 模拟服务负载的脚本
└── run.py                    # 应用主入口

# 云资源配置与成本单价（示例为AWS us-east-1按需实例价格）
instance_types:
  c5.large:
    vcpus: 2
    memory_gib: 4
    cost_per_hour_usd: 0.085
  m5.xlarge:
    vcpus: 4
    memory_gib: 16
    cost_per_hour_usd: 0.192

# 当前应用运行的实例类型
current_instance: c5.large

# 成本计算基准：将CPU时间（秒）转换为成本
# 公式：成本 = (函数总CPU秒数 / 3600) * 实例每小时成本
cost_calculation_base: hourly

# 可接受的性能阈值（毫秒），用于报告生成
performance_thresholds:
  critical: 1000
  warning: 500
  info: 200

from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional, Dict, Any

class FunctionProfileData(BaseModel):
    """单次函数调用的性能数据点"""
    func_name: str
    start_time: datetime
    end_time: datetime
    duration_ms: float = Field(..., gt=0)  # 持续时间（毫秒）
    module_name: str
    extra_tags: Dict[str, Any] = Field(default_factory=dict) # 用于携带业务标签，如order_id

class AggregatedProfile(BaseModel):
    """聚合后的函数性能数据"""
    func_name: str
    module_name: str
    total_calls: int = 0
    total_duration_ms: float = 0.0
    avg_duration_ms: float = 0.0
    p95_duration_ms: Optional[float] = None
    # 成本相关字段（由成本映射引擎填充）
    estimated_cost_usd: float = 0.0
    cost_percentage: float = 0.0  # 该函数成本占总监控成本的百分比

class CostReport(BaseModel):
    """成本优化报告条目"""
    func_name: str
    module_name: str
    avg_duration_ms: float
    total_cost_usd: float
    cost_percentage: float
    severity: str  # 'critical', 'warning', 'info'
    suggestion: str

import time
import functools
import sqlite3
from contextlib import contextmanager
from datetime import datetime
from typing import Callable, Any, Dict
from threading import Lock
from pathlib import Path
import json

from ..core.models import FunctionProfileData

class ProfilerAgent:
    """
    性能数据收集代理。使用单例模式管理数据库连接和写入。
    采用同步写入SQLite简化设计，生产环境应考虑异步或批量写入。
    """
    _instance = None
    _lock = Lock()

    def __new__(cls, db_path: str = "data/profiler_data.db"):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super(ProfilerAgent, cls).__new__(cls)
                cls._instance._initialized = False
                cls._instance.db_path = Path(db_path)
            return cls._instance

    def __init__(self, db_path: str = "data/profiler_data.db"):
        if self._initialized:
            return
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._init_database()
        self._initialized = True

    def _init_database(self):
        """初始化数据库表"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS function_profile (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                func_name TEXT NOT NULL,
                module_name TEXT NOT NULL,
                start_time TIMESTAMP NOT NULL,
                end_time TIMESTAMP NOT NULL,
                duration_ms REAL NOT NULL,
                extra_tags TEXT DEFAULT '{}'
            )
        ''')
        # 创建索引以加速查询
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_func_name ON function_profile(func_name)')
        cursor.execute('CREATE INDEX IF NOT EXISTS idx_start_time ON function_profile(start_time)')
        conn.commit()
        conn.close()

    def _save_profile_data(self, data: FunctionProfileData):
        """保存单条性能数据到数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            INSERT INTO function_profile (func_name, module_name, start_time, end_time, duration_ms, extra_tags)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (
            data.func_name,
            data.module_name,
            data.start_time.isoformat(),
            data.end_time.isoformat(),
            data.duration_ms,
            json.dumps(data.extra_tags)
        ))
        conn.commit()
        conn.close()

    def profile_function(self, func: Callable) -> Callable:
        """
        装饰器：用于监控普通函数的执行时间。
        """
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = datetime.now()
            start_perf = time.perf_counter()
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                end_perf = time.perf_counter()
                end_time = datetime.now()
                duration_ms = (end_perf - start_perf) * 1000  # 转换为毫秒

                profile_data = FunctionProfileData(
                    func_name=func.__name__,
                    module_name=func.__module__,
                    start_time=start_time,
                    end_time=end_time,
                    duration_ms=duration_ms,
                    extra_tags=getattr(func, '_profile_tags', {})  # 允许通过函数属性传递标签
                )
                # 异步保存可能更好，这里简化为同步
                self._save_profile_data(profile_data)
        return wrapper

    @contextmanager
    def profile_block(self, block_name: str, module_name: str = None, **tags):
        """
        上下文管理器：用于监控代码块的执行时间。
        用法：
            with profiler.profile_block("expensive_loop", module=__name__, order_id=order.id):
                # 被监控的代码
        """
        start_time = datetime.now()
        start_perf = time.perf_counter()
        caller_module = module_name or __name__
        try:
            yield
        finally:
            end_perf = time.perf_counter()
            end_time = datetime.now()
            duration_ms = (end_perf - start_perf) * 1000

            profile_data = FunctionProfileData(
                func_name=block_name,
                module_name=caller_module,
                start_time=start_time,
                end_time=end_time,
                duration_ms=duration_ms,
                extra_tags=tags
            )
            self._save_profile_data(profile_data)

# 创建全局代理实例
global_profiler_agent = ProfilerAgent()

# 便捷装饰器
profile = global_profiler_agent.profile_function

import time
import random
from decimal import Decimal
from typing import List, Dict
from ..agent.profiler import profile, global_profiler_agent as profiler

class OrderService:
    """
    模拟的订单处理服务，内含故意设计的性能问题用于演示。
    """
    def __init__(self):
        self._inventory = {"widget_a": 100, "widget_b": 50, "widget_c": 200}

    @profile
    def process_order(self, order_id: str, items: List[Dict]) -> Dict:
        """处理订单的主函数"""
        # 模拟验证
        self._validate_order(items)
        # 模拟计算（包含问题）
        total = self._calculate_total_with_issues(items)
        # 模拟库存更新
        self._update_inventory(items)
        # 模拟记录日志（模拟慢I/O）
        self._log_order(order_id, total)
        return {"order_id": order_id, "status": "processed", "total": float(total)}

    def _validate_order(self, items):
        """验证订单项 - 快速操作"""
        time.sleep(0.001 * len(items))  # 模拟轻量级工作
        if not items:
            raise ValueError("订单项不能为空")

    def _calculate_total_with_issues(self, items):
        """
        计算订单总额 - 包含性能问题:

        1. 冗余的数据库/缓存查询模拟（嵌套循环内重复调用）
        2. 低效的算法（模拟复杂计算）
        """
        total = Decimal('0.0')
        # 问题1：在循环内模拟重复的"重操作"（如查询基础数据）
        for item in items:
            # 每次循环都"查询"价格和折扣，实际应缓存
            base_price = self._fetch_price_from_mock_db(item['sku'])
            discount = self._fetch_discount_from_mock_db(item['sku'])
            # 问题2：模拟一个不必要的复杂计算（例如，重复的数学运算）
            for _ in range(50):  # 放大问题
                item_total = Decimal(base_price) * Decimal(item['quantity']) * Decimal(1 - discount)
            total += item_total
        return total

    def _fetch_price_from_mock_db(self, sku: str):
        """模拟一个相对耗时的数据获取操作（如数据库/缓存查询）"""
        time.sleep(0.005)  # 5毫秒延迟
        prices = {"widget_a": 29.99, "widget_b": 45.50, "widget_c": 12.75}
        return prices.get(sku, 10.00)

    def _fetch_discount_from_mock_db(self, sku: str):
        """模拟另一个耗时的数据获取操作"""
        time.sleep(0.003)  # 3毫秒延迟
        discounts = {"widget_a": 0.1, "widget_b": 0.05, "widget_c": 0.0}
        return discounts.get(sku, 0.0)

    def _update_inventory(self, items):
        """更新库存 - 使用上下文管理器监控一个块"""
        with profiler.profile_block("update_inventory_block", module_name=__name__):
            time.sleep(0.002 * len(items))
            for item in items:
                if item['sku'] in self._inventory:
                    self._inventory[item['sku']] -= item['quantity']

    def _log_order(self, order_id, total):
        """模拟日志写入（慢I/O）"""
        time.sleep(0.010)  # 10毫秒延迟

# 辅助函数：生成模拟订单
def generate_mock_order(item_count=5):
    skus = ["widget_a", "widget_b", "widget_c"]
    items = []
    for _ in range(item_count):
        sku = random.choice(skus)
        items.append({
            "sku": sku,
            "quantity": random.randint(1, 5)
        })
    return items

import yaml
from typing import List
from pathlib import Path
from .models import AggregatedProfile

class CostMapper:
    """成本映射引擎：将性能数据转换为成本估算"""
    def __init__(self, config_path: str = "config/cost_config.yaml"):
        self.config_path = Path(config_path)
        self._load_config()

    def _load_config(self):
        with open(self.config_path, 'r') as f:
            self.config = yaml.safe_load(f)
        instance_type = self.config['current_instance']
        self.instance_spec = self.config['instance_types'][instance_type]
        self.cost_per_hour = self.instance_spec['cost_per_hour_usd']
        self.vcpus = self.instance_spec['vcpus']

    def calculate_cost(self, aggregated_profiles: List[AggregatedProfile]) -> List[AggregatedProfile]:
        """
        为聚合的性能数据计算预估成本。
        假设：函数耗时均匀占用一个vCPU。
        成本 = (函数总CPU时间 / 3600秒) * 实例每小时成本
        总CPU时间 = 总耗时(秒) / vCPU数量 (简化模型)
        """
        # 首先计算所有被监控函数的总CPU时间（秒）
        total_cpu_seconds = 0.0
        for profile in aggregated_profiles:
            # 将毫秒转换为秒，并简单假设平均使用一个vCPU
            total_seconds = profile.total_duration_ms / 1000.0
            cpu_seconds = total_seconds / self.vcpus
            total_cpu_seconds += cpu_seconds

        if total_cpu_seconds <= 0:
            return aggregated_profiles

        # 计算每个函数的成本及占比
        for profile in aggregated_profiles:
            func_total_seconds = profile.total_duration_ms / 1000.0
            func_cpu_seconds = func_total_seconds / self.vcpus
            profile.estimated_cost_usd = (func_cpu_seconds / 3600.0) * self.cost_per_hour
            if total_cpu_seconds > 0:
                profile.cost_percentage = (func_cpu_seconds / total_cpu_seconds) * 100.0
            else:
                profile.cost_percentage = 0.0

        return aggregated_profiles

import sqlite3
from typing import List, Tuple
from pathlib import Path
import statistics

from ..core.models import AggregatedProfile, CostReport
from ..core.cost_mapper import CostMapper

class HotspotAnalyzer:
    """
    热点分析引擎：从数据库读取原始数据，聚合分析，并生成优化报告。
    """
    def __init__(self, db_path: str = "data/profiler_data.db"):
        self.db_path = Path(db_path)
        self.cost_mapper = CostMapper()

    def _fetch_raw_data(self, limit: int = 10000) -> List[Tuple]:
        """从数据库获取原始性能数据"""
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute('''
            SELECT func_name, module_name, duration_ms
            FROM function_profile
            ORDER BY start_time DESC
            LIMIT ?
        ''', (limit,))
        rows = cursor.fetchall()
        conn.close()
        return rows

    def aggregate_profiles(self) -> List[AggregatedProfile]:
        """聚合原始数据，按函数名和模块名分组"""
        rows = self._fetch_raw_data()
        profile_map = {}

        for row in rows:
            key = (row['func_name'], row['module_name'])
            if key not in profile_map:
                profile_map[key] = {
                    'durations': [],
                    'total_calls': 0,
                    'total_duration_ms': 0.0
                }
            data = profile_map[key]
            data['durations'].append(row['duration_ms'])
            data['total_calls'] += 1
            data['total_duration_ms'] += row['duration_ms']

        aggregated_list = []
        for (func_name, module_name), data in profile_map.items():
            durations = data['durations']
            avg_duration = data['total_duration_ms'] / data['total_calls']
            p95 = statistics.quantiles(durations, n=20)[18] if len(durations) >= 5 else avg_duration # 计算95分位点

            profile = AggregatedProfile(
                func_name=func_name,
                module_name=module_name,
                total_calls=data['total_calls'],
                total_duration_ms=data['total_duration_ms'],
                avg_duration_ms=avg_duration,
                p95_duration_ms=p95
            )
            aggregated_list.append(profile)

        return aggregated_list

    def generate_cost_report(self, aggregated_profiles: List[AggregatedProfile]) -> List[CostReport]:
        """生成包含成本和建议的分析报告"""
        # 1. 成本映射
        profiles_with_cost = self.cost_mapper.calculate_cost(aggregated_profiles)
        # 2. 按成本排序并生成报告
        profiles_sorted = sorted(profiles_with_cost, key=lambda x: x.estimated_cost_usd, reverse=True)

        report = []
        for p in profiles_sorted:
            # 确定严重级别
            if p.avg_duration_ms >= 1000: # 使用config中的thresholds更好，此处简化
                severity = "critical"
                sugg = f"函数平均耗时{p.avg_duration_ms:.2f}ms，严重影响性能与成本。建议：检查内部循环、优化算法、引入缓存。"
            elif p.avg_duration_ms >= 500:
                severity = "warning"
                sugg = f"函数平均耗时{p.avg_duration_ms:.2f}ms，是潜在热点。建议：分析内部调用，考虑异步或批量操作。"
            else:
                severity = "info"
                sugg = f"函数性能尚可，但因调用频繁产生{p.estimated_cost_usd:.6f}美元成本。建议：监控调用量。"

            cost_report = CostReport(
                func_name=p.func_name,
                module_name=p.module_name,
                avg_duration_ms=round(p.avg_duration_ms, 2),
                total_cost_usd=round(p.estimated_cost_usd, 6),
                cost_percentage=round(p.cost_percentage, 2),
                severity=severity,
                suggestion=sugg
            )
            report.append(cost_report)

        return report

from flask import Flask, jsonify, render_template_string, request
from pathlib import Path
import json

from ..analyzer.hotspot import HotspotAnalyzer

app = Flask(__name__)
analyzer = HotspotAnalyzer()

# 简单的HTML模板用于看板
DASHBOARD_HTML = """
<!DOCTYPE html>
<html>
<head>
    <title>FinOps 性能与成本看板</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body { font-family: sans-serif; margin: 20px; }
        .chart-container { width: 80%; margin: 20px auto; }
        table { border-collapse: collapse; width: 100%; margin-top: 20px; }
        th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
        th { background-color: #f2f2f2; }
        .critical { background-color: #ffcccc; }
        .warning { background-color: #fff3cd; }
        .info { background-color: #d4edda; }
    </style>
</head>
<body>
    <h1>FinOps 性能与成本看板</h1>
    <p>基于代码剖析的性能瓶颈定位与成本优化分析</p>
    <div class="chart-container">
        <canvas id="costChart"></canvas>
    </div>
    <div id="reportTable">
        <h2>优化建议报告</h2>
        <table>
            <thead>
                <tr>
                    <th>函数名</th>
                    <th>模块</th>
                    <th>平均耗时(ms)</th>
                    <th>预估成本(USD)</th>
                    <th>成本占比(%)</th>
                    <th>严重级别</th>
                    <th>优化建议</th>
                </tr>
            </thead>
            <tbody id="reportBody">
                <!-- 数据由JavaScript动态填充 -->
            </tbody>
        </table>
    </div>

    <script>
        async function loadData() {
            const reportResp = await fetch('/api/report');
            const reportData = await reportResp.json();

            // 填充表格
            const tbody = document.getElementById('reportBody');
            tbody.innerHTML = '';
            reportData.forEach(item => {
                const row = document.createElement('tr');
                row.className = item.severity;
                row.innerHTML = `
                    <td>${item.func_name}</td>
                    <td>${item.module_name}</td>
                    <td>${item.avg_duration_ms}</td>
                    <td>${item.total_cost_usd}</td>
                    <td>${item.cost_percentage}</td>
                    <td>${item.severity}</td>
                    <td>${item.suggestion}</td>
                `;
                tbody.appendChild(row);
            });

            // 绘制成本占比饼图
            const ctx = document.getElementById('costChart').getContext('2d');
            const labels = reportData.map(d => d.func_name);
            const costData = reportData.map(d => d.total_cost_usd);
            const backgroundColors = reportData.map(d => 
                d.severity === 'critical' ? '#dc3545' :
                d.severity === 'warning' ? '#ffc107' : '#28a745'
            );

            new Chart(ctx, {
                type: 'pie',
                data: {
                    labels: labels,
                    datasets: [{
                        data: costData,
                        backgroundColor: backgroundColors,
                        borderWidth: 1
                    }]
                },
                options: {
                    responsive: true,
                    plugins: {
                        title: {
                            display: true,
                            text: '各函数预估成本分布（美元）'
                        },
                        tooltip: {
                            callbacks: {
                                label: function(context) {
                                    const label = context.label || '';
                                    const value = context.raw || 0;
                                    const percentage = context.dataset.data.length > 0 ? 
                                        ((value / context.dataset.data.reduce((a,b)=>a+b)) * 100).toFixed(2) : 0;
                                    return `${label}: $${value.toFixed(6)} (${percentage}%)`;
                                }
                            }
                        }
                    }
                }
            });
        }
        // 页面加载时获取数据
        window.onload = loadData;
        // 每30秒刷新一次
        setInterval(loadData, 30000);
    </script>
</body>
</html>
"""

@app.route('/')
def dashboard():
    """渲染主看板页面"""
    return render_template_string(DASHBOARD_HTML)

@app.route('/api/aggregated', methods=['GET'])
def get_aggregated_data():
    """API端点：获取聚合性能数据"""
    aggregated = analyzer.aggregate_profiles()
    # 转换为字典列表
    result = [item.dict() for item in aggregated]
    return jsonify(result)

@app.route('/api/report', methods=['GET'])
def get_cost_report():
    """API端点：获取成本优化报告"""
    aggregated = analyzer.aggregate_profiles()
    report = analyzer.generate_cost_report(aggregated)
    result = [item.dict() for item in report]
    return jsonify(result)

@app.route('/api/raw/count', methods=['GET'])
def get_raw_data_count():
    """API端点：获取原始数据条数，用于健康检查"""
    import sqlite3
    conn = sqlite3.connect(analyzer.db_path)
    cursor = conn.cursor()
    cursor.execute('SELECT COUNT(*) FROM function_profile')
    count = cursor.fetchone()[0]
    conn.close()
    return jsonify({"raw_data_count": count})

if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=5000)

#!/usr/bin/env python3
"""
负载模拟脚本：持续调用订单服务，生成性能数据。
"""
import sys
import time
import random
from pathlib import Path

# 添加项目根目录到路径
sys.path.insert(0, str(Path(__file__).parent.parent))

from src.service.order_service import OrderService, generate_mock_order

def simulate_continuous_load(run_seconds: int = 300, interval: float = 0.5):
    """
    模拟持续负载。
    :param run_seconds: 总运行时间（秒）
    :param interval: 每次调用之间的基础间隔（秒）
    """
    service = OrderService()
    end_time = time.time() + run_seconds
    order_count = 0

    print(f"开始模拟负载，持续{run_seconds}秒...")
    try:
        while time.time() < end_time:
            order_count += 1
            items = generate_mock_order(item_count=random.randint(1, 10))
            order_id = f"simulated_{order_count:06d}"
            try:
                result = service.process_order(order_id, items)
                print(f"[{order_count}] 处理订单 {order_id}, 结果: {result['status']}")
            except Exception as e:
                print(f"[{order_count}] 处理订单时出错: {e}")

            # 随机间隔，模拟真实流量波动
            sleep_time = interval * random.uniform(0.5, 2.0)
            time.sleep(sleep_time)
    except KeyboardInterrupt:
        print("\n模拟被用户中断。")
    finally:
        print(f"模拟结束。共处理 {order_count} 个订单。")

if __name__ == '__main__':
    # 默认运行5分钟
    simulate_continuous_load(run_seconds=300)

#!/usr/bin/env python3
"""
应用主入口。启动Flask API服务。
"""
from src.api.app import app
import webbrowser
import threading
import time

def open_browser():
    """在默认浏览器中打开看板"""
    time.sleep(1.5)  # 等待Flask启动
    webbrowser.open('http://localhost:5000')

if __name__ == '__main__':
    print("启动 FinOps 性能与成本看板服务...")
    print("看板地址: http://localhost:5000")
    print("API端点:")
    print("  - 聚合数据: GET http://localhost:5000/api/aggregated")
    print("  - 成本报告: GET http://localhost:5000/api/report")
    print("  - 数据统计: GET http://localhost:5000/api/raw/count")
    print("\n提示：请运行 `python scripts/simulate_load.py` 来生成负载和性能数据。")

    # 在后台线程中打开浏览器
    threading.Thread(target=open_browser, daemon=True).start()

    # 启动Flask应用
    app.run(debug=True, host='0.0.0.0', port=5000, use_reloader=False)

Flask==2.3.3
pydantic==2.5.0
PyYAML==6.0.1

# 进入项目目录
cd perf-cost-opt

# 创建虚拟环境（推荐）
python -m venv venv

# 激活虚拟环境
# Windows:
venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate

# 安装依赖
pip install -r requirements.txt

python run.py

python scripts/simulate_load.py

curl http://localhost:5000/api/raw/count

curl http://localhost:5000/api/report | python -m json.tool

# manual_test.py
import sys
sys.path.insert(0, '.')
from src.service.order_service import OrderService

service = OrderService()
order_items = [{"sku": "widget_a", "quantity": 2}]
result = service.process_order("manual_test_001", order_items)
print(result)