Python Flask内存泄漏分析定位

通过蓝图注册内存跟踪分析接口

示例代码如下，开启内存分析跟踪需要在Flask服务启动前注入环境变量
export PYTHONTRACEMALLOC=1

from typing import Literal
from flask import Blueprint, jsonify, request
import tracemalloc
import os
import linecache

snapshot = None
run_path = os.path.dirname(__file__) # 根据实际需要修改为项目目录，目的是方便后续只追踪本项目的内存泄漏


app_bus_blueprint = Blueprint('memory', __name__)

def filter_traces(snapshot: tracemalloc.Snapshot, left_trace:Literal['only my code', 'all', 'beside my code']='only my code'):
    filter_list = (
        # tracemalloc.Filter(False, "<frozen importlib._bootstrap>"),
        # tracemalloc.Filter(False, "<unknown>"),
        # tracemalloc.Filter(False, "<frozen importlib._bootstrap_external>"),
        tracemalloc.Filter(False, tracemalloc.__file__),
        tracemalloc.Filter(False, linecache.__file__),
        tracemalloc.Filter(False, f"*.vscode-server*"),
        # tracemalloc.Filter(True, jober.__file__),
    )
    trace_store = {
        'only my code': (
            tracemalloc.Filter(True, f"{run_path}*"),
        ),
        'all': (),
        'beside my code':(
            tracemalloc.Filter(False, f"{run_path}*"),
        )
    }
    filter_list = filter_list + trace_store[left_trace]
    snapshot = snapshot.filter_traces(filter_list)
    return snapshot

def display_top(snapshot: tracemalloc.Snapshot, key_type='lineno', limit=30):
    lines = []
    # snapshot = filter_traces(snapshot)
    top_stats = snapshot.statistics(key_type)
    lines.append("Top %s lines" % limit)
    for index, stat in enumerate(top_stats[:limit], 1):
        frame = stat.traceback[0]
        lines.append("#%s: %s:%s: %.1f KiB"
              % (index, frame.filename, frame.lineno, stat.size / 1024))
        # lines.extend(stat.traceback.format())
        line = linecache.getline(frame.filename, frame.lineno).strip()
        if line:
            lines.append('    %s' % line)
        lines.append('---------------------------------')
    other = top_stats[limit:]
    if other:
        size = sum(stat.size for stat in other)
        lines.append("%s other: %.1f KiB" % (len(other), size / 1024))
    total = sum(stat.size for stat in top_stats)
    lines.append("Total allocated size: %.1f KiB" % (total / 1024))
    return lines

if str(os.getenv('PYTHONTRACEMALLOC', 0)) == '1':
    tracemalloc.start(25)
    @app_bus_blueprint.route('/memory_snapshot', methods=['POST', 'GET'], strict_slashes=False)
    def memory_snapshot():
        is_compare = isinstance(request.args.get('compare', False), str)
        global snapshot
        if not snapshot:
            snapshot = tracemalloc.take_snapshot()
            snapshot = filter_traces(snapshot)
            return "Taken snapshot."
        else:
            lines = []
            snapshot_current = tracemalloc.take_snapshot()
            snapshot_current = filter_traces(snapshot_current)
            if is_compare:
                top_stats = snapshot_current.compare_to(snapshot, 'lineno')
                # 过滤出只有增长的内存分配
                increased_stats = [stat for stat in top_stats if stat.size_diff > 0]
                # 取出增长最多的前10条数据
                top_increased_stats = sorted(increased_stats, key=lambda stat: stat.size_diff, reverse=True)
                for stat in top_increased_stats[:20]:
                    lines.append("%s memory blocks: %.1f KiB" % (stat.count, stat.size / 1024))
                    lines.extend(stat.traceback.format())
                    # lines.append(str(stat))
                    lines.append('-----------------------------------')
                total_increased = sum(stat.size_diff for stat in top_increased_stats)
                total_decreased = sum(stat.size_diff for stat in top_stats if stat.size_diff < 0)
                totol_allocated = sum(stat.size for stat in top_stats)
                lines.append(f"Total increased size: {total_increased / 1024:.1f} KiB")
                lines.append(f"Total decreased size: {total_decreased / 1024:.1f} KiB")
                lines.append(f"Absolute change size: {(total_increased + total_decreased) / 1024:.1f} KiB")
                lines.append(f"Total allocated size: {totol_allocated / 1024:.1f} KiB")
            else:
                lines = display_top(snapshot_current, key_type='traceback')
            snapshot = snapshot_current
            return jsonify(lines)

通过trace filter，可以选择’only my code’, ‘all’, 'beside my code’三种trace筛选策略，意思为：只跟踪我的工作区代码，所有，非我的代码/第三方包。

参考文章：

1. 获取一个内存块的溯源；
2. 定位python内存泄漏问题