Metrics API

性能指标收集模块,用于监控和分析任务执行情况。

MetricsCollector

MetricsCollector(total_executions=0, successful_executions=0, failed_executions=0, total_duration=0.0, task_stats=dict(), start_time=datetime.now()) dataclass

指标收集器,用于收集调度器性能指标。

示例

collector = MetricsCollector()

# 记录任务执行
collector.record_task_execution("my_task", success=True, duration=1.5)

# 获取统计信息
stats = collector.get_stats()
print(f"总执行次数: {stats['total_executions']}")

record_task_execution(task_name, success, duration)

记录任务执行。

参数

task_name: 任务名称 success: 是否成功 duration: 执行时长(秒)

get_stats()

获取统计信息。

返回值

包含所有统计信息的字典

export_prometheus()

导出 Prometheus 格式的指标。

返回值

Prometheus 格式的指标文本

示例

metrics_text = collector.export_prometheus()
# 可以通过 HTTP 端点暴露给 Prometheus

format_table(*, show_tasks=True, max_tasks=20)

格式化指标为美观的表格展示。

参数

show_tasks: 是否显示任务级别统计 max_tasks: 最多显示的任务数量

返回值

格式化的表格字符串

示例

print(collector.format_table())

format_task_detail(task_name)

格式化单个任务的详细统计。

参数

task_name: 任务名称

返回值

格式化的详细统计字符串

抛出

KeyError: 任务不存在

reset()

重置所有统计信息。