为何基于psutil的Python CPU追踪器在Google Cloud Run中无法工作？

核心原因 #

1. 误用系统级CPU统计接口：代码中psutil.cpu_percent()是获取整个宿主机的CPU使用率，而非当前进程的CPU消耗。在Cloud Run的隔离容器环境中，该接口无法获取容器内进程的有效CPU数据（可能返回0或宿主机无关数值），而本地环境中系统CPU变化能间接反映进程消耗，因此本地测试看似正常。
2. CPU测量逻辑冗余且不准确：先调用psutil.cpu_percent(interval=None)重置计数器后sleep(1)，这1秒的空闲时间会拉低后续统计的CPU数值，且无法精准对应目标函数执行期间的CPU消耗。
3. Cloud Run容器的资源隔离限制：即使开启CPU始终分配，容器对系统级资源统计接口有隔离，系统级CPU统计无法映射到容器内进程的实际CPU使用情况。

解决方法 #

1. 切换为进程级CPU统计

将系统级CPU测量替换为当前进程的CPU使用率统计，使用psutil.Process(os.getpid()).cpu_percent()，修改后的代码如下：

import psutil
import os
import time

def process_memory():
    """
使用psutil库通过os.getpid()获取的进程ID访问进程的内存统计信息。
返回值对应驻留集大小(RSS)，即进程存储在RAM中的内存部分。

    Returns:
        int: 进程使用的RAM内存（字节）。
    """
    process = psutil.Process(os.getpid())
    mem_info = process.memory_info()
    return mem_info.rss

def Tracker(func):
    def wrapper(*args, **kwargs):
        # 提前获取当前进程对象，避免重复创建
        current_process = psutil.Process(os.getpid())
        
        #####################################
        ## -- 函数执行前 -- ##
        #####################################
        mem_before = process_memory()
        # 重置进程CPU计数器
        current_process.cpu_percent(interval=None)

        ##############################
        ## -- 执行目标函数 -- ##
        ##############################
        responses, final_tracker = func(*args, **kwargs)

        ###########################
        ## -- 函数执行后 -- ##
        ###########################
        mem_after = process_memory()
        consumed_memory = (mem_after - mem_before) / (1024 ** 2)  # 转换为MB
        
        # 获取目标函数执行期间的进程CPU使用率
        cpu_usage = current_process.cpu_percent(interval=None)

        final_tracker = {
            "Memory Consumed (MB)": round(consumed_memory, 2),
            "CPU Consumed (%)": round(cpu_usage, 2)
        }
 
        return responses, final_tracker
    return wrapper

2. 更精准的CPU时间统计（可选）

如果需要更高精度的CPU消耗统计，可以直接计算进程的CPU时间差与实际运行时间的比值：

def Tracker(func):
    def wrapper(*args, **kwargs):
        current_process = psutil.Process(os.getpid())
        
        mem_before = process_memory()
        # 获取执行前的进程CPU时间（用户态+内核态）
        cpu_times_before = current_process.cpu_times()
        start_real_time = time.time()

        responses, final_tracker = func(*args, **kwargs)

        mem_after = process_memory()
        cpu_times_after = current_process.cpu_times()
        elapsed_real_time = time.time() - start_real_time

        consumed_memory = (mem_after - mem_before) / (1024 ** 2)
        # 计算CPU使用率：(CPU时间差 / 实际运行时间) * 100
        cpu_time_diff = (cpu_times_after.user + cpu_times_after.system) - (cpu_times_before.user + cpu_times_before.system)
        cpu_usage = (cpu_time_diff / elapsed_real_time) * 100

        final_tracker = {
            "Memory Consumed (MB)": round(consumed_memory, 2),
            "CPU Consumed (%)": round(cpu_usage, 2)
        }
 
        return responses, final_tracker
    return wrapper

3. 确认Cloud Run配置

确保已正确配置：

• 将CPU分配模式设置为始终分配，避免函数执行时CPU被节流
• 为容器分配足够的CPU配额（建议至少1 vCPU），防止资源限制导致统计异常

内容的提问来源于stack exchange，提问作者lapiceroazul4