将任意时间戳数据转换为 Perfetto

在本指南中,你将学习如何:

如果你有来自自己系统的现有日志或时间戳数据,你不需要错过 Perfetto 强大的可视化和 profile 功能。通过将你的数据转换为 Perfetto 的原生基于 protobuf 的 trace 格式,你可以创建可以在 Perfetto UI 中打开并使用 Trace Processor 查询的合成 traces。

本页面提供了如何以编程方式生成这些合成 traces 的指南。

基础:Perfetto 的 Trace 格式

Perfetto trace 文件(.pftrace 或 .perfetto-trace)是一系列 TracePacket 消息,包装在根 Trace 消息中。每个 TracePacket 可以包含各种类型的数据。

对于从自定义数据生成 traces,在 TracePacket 中最常用和最灵活的负载是 TrackEvent。TrackEvent 允许你定义:

以编程方式生成 Traces

本指南中的示例使用 Python 和 perfetto Python 库中的帮助器类来演示如何构造这些 protobuf 消息。但是,底层原理和 protobuf 定义与语言无关。你可以在任何具有 Protocol Buffer 支持的编程语言中生成 Perfetto traces。

无论使用什么语言,核心任务都是构造 TracePacket 消息(根据 Perfetto 的 protobuf schemas)并将它们序列化为二进制文件。

Python 脚本模板

对于以下部分中的 Python 示例,我们将使用脚本模板。此脚本处理创建 trace 文件和序列化 TracePacket 消息的基础知识。你将使用要创建的 trace 数据类型的特定逻辑填充 populate_packets 函数。

首先,确保你已安装 perfetto 库,它提供了必要的 protobuf 类,可能还有一个构建器实用程序(如你设计的 TraceProtoBuilder 类或库中的等效项)。

pip install perfetto

这是 Python 脚本模板。将此保存为 trace_converter_template.py 或类似名称。每个后续示例将向你展示在 populate_packets 函数中放置什么代码。

点击展开/折叠 Python 代码
#!/usr/bin/env python3 import uuid from perfetto.trace_builder.proto_builder import TraceProtoBuilder from perfetto.protos.perfetto.trace.perfetto_trace_pb2 import TrackEvent, TrackDescriptor, ProcessDescriptor, ThreadDescriptor # 为此数据包序列定义唯一 ID(每个 trace 生产者生成一次) TRUSTED_PACKET_SEQUENCE_ID = 1001 # 选择任意唯一整数 def get_uuid() -> int: return uuid.uuid4().int & ((1 << 63) - 1) def populate_packets(builder: TraceProtoBuilder): """ This function is where you will define and add your TracePackets to the trace. The examples in the following sections will provide the specific code to insert here. Args: builder: An instance of TraceProtoBuilder to add packets to. """ # ======== 在此处开始你的数据包创建代码 ======== # 示例(稍后将由特定示例替换): # # packet = builder.add_packet() # packet.timestamp = 1000 # packet.track_event.type = TrackEvent.TYPE_SLICE_BEGIN # packet.track_event.name = "My Example Event" # packet.track_event.track_uuid = 12345 # # packet2 = builder.add_packet() # packet2.timestamp = 2000 # packet2.track_event.type = TrackEvent.TYPE_SLICE_END # packet2.track_event.track_uuid = 12345 # # ======== 在此处结束你的数据包创建代码 ======== # 添加代码时删除此 'pass' pass def main(): """ Initializes the TraceProtoBuilder, calls populate_packets to fill it, and then writes the resulting trace to a file. """ builder = TraceProtoBuilder() populate_packets(builder) output_filename = "my_custom_trace.pftrace" with open(output_filename, 'wb') as f: f.write(builder.serialize()) print(f"Trace written to {output_filename}") print("Open with [https://ui.perfetto.dev](https://ui.perfetto.dev).") if __name__ == "__main__": main()

使用此模板:

  1. 将上面的代码保存为 Python 文件(例如 trace_converter_template.py)。
  2. 对于以下部分中的每个示例(例如,"Thread-scoped slices"、"Counters"),复制该部分提供的 Python 代码并将其粘贴到 trace_converter_template.py 文件中的 populate_packets 函数中,替换示例占位符内容。
  3. 运行脚本:python trace_converter_template.py。这将生成 my_custom_trace.pftrace。

TraceProtoBuilder 类(从 perfetto pip 包导入)帮助管理构成 Trace 的 TracePacket 消息列表。populate_packets 函数是你根据特定数据定义这些 packets 内容的地方。

创建基本 Timeline Slices

在 Perfetto 中表示活动的最基本方式是"slice"。slice 只是一个具有开始时间和持续时间的命名事件。Slices 存在于"tracks"上,tracks 是 Perfetto UI 中的可视 Timeline。本质上,slices 用于任何你想说"一个命名活动发生在特定时间间隔内"的情况。

Slices 可以表示的常见示例包括:

要从自定义数据创建 slices,你通常需要:

  1. 定义一个 track,你的 slices 将出现在其中。这是使用 TrackDescriptor 数据包完成的。对于基本自定义数据,你可以创建一个不绑定到特定进程或线程的通用 track。
  2. 对于数据中的每个事件,发出 TrackEvent 数据包以标记 slice 的开始和结束。

Python 示例:基本 Slices

假设你有表示任务的数据,具有名称、开始时间和结束时间。以下是如何将它们转换为自定义 track 上的 Perfetto slices。此第一个示例将显示不同的、非嵌套的 slices 和单个 instant 事件。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# 为你的自定义 track 定义唯一的 UUID(生成一个 64 位随机数) CUSTOM_TRACK_UUID = 12345678 # 示例 UUID # 1. 定义自定义 Track # 此数据包描述将显示事件的 track。 # 在 trace 开始时发出一次。 packet = builder.add_packet() packet.track_descriptor.uuid = CUSTOM_TRACK_UUID packet.track_descriptor.name = "My Custom Data Timeline" # 2. 为此自定义 track 发出事件 # 示例事件 1:"Task A" packet = builder.add_packet() packet.timestamp = 1000 # 开始时间(纳秒) packet.track_event.type = TrackEvent.TYPE_SLICE_BEGIN packet.track_event.track_uuid = CUSTOM_TRACK_UUID # 与 track 关联 packet.track_event.name = "Task A" packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID packet = builder.add_packet() packet.timestamp = 1500 # 结束时间(纳秒) packet.track_event.type = TrackEvent.TYPE_SLICE_END packet.track_event.track_uuid = CUSTOM_TRACK_UUID packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 示例事件 2:"Task B" - 稍后发生的单独、非嵌套任务 packet = builder.add_packet() packet.timestamp = 1600 # 开始时间(纳秒) packet.track_event.type = TrackEvent.TYPE_SLICE_BEGIN packet.track_event.track_uuid = CUSTOM_TRACK_UUID packet.track_event.name = "Task B" packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID packet = builder.add_packet() packet.timestamp = 1800 # 结束时间(纳秒) packet.track_event.type = TrackEvent.TYPE_SLICE_END packet.track_event.track_uuid = CUSTOM_TRACK_UUID packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 示例事件 3:瞬时事件 packet = builder.add_packet() packet.timestamp = 1900 # 时间戳(纳秒) packet.track_event.type = TrackEvent.TYPE_INSTANT packet.track_event.track_uuid = CUSTOM_TRACK_UUID packet.track_event.name = "Milestone Y" packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Basic Timeline Slices

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询这些 slices:

SELECT ts, dur, name FROM slice JOIN track ON slice.track_id = track.id WHERE track.name = 'My Custom Data Timeline';

嵌套 Slices(分层活动)

通常,一个活动或操作由几个子活动组成,这些子活动必须在主活动完成之前完成。嵌套 slices 非常适合表示这些分层关系。关键规则是子 slices 必须在父 slice 开始之后开始,并在父 slice 结束之前结束。

这在以下情况非常常见:

Perfetto UI 将直观地嵌套这些 slices,使层次结构清晰。

Python 示例:嵌套 Slices

此示例演示在自定义 track 上创建多个嵌套 slices 堆栈。packets 按时间戳顺序发出以正确表示嵌套。我们将在 populate_packets 中定义一个小的帮助函数 add_event 以减少样板代码。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# 为此示例的自定义 track 定义唯一的 UUID NESTED_SLICE_TRACK_UUID = 987654321 # 示例 UUID # 1. 定义用于嵌套 Slices 的自定义 Track # 开始时发出一次。 packet = builder.add_packet() packet.track_descriptor.uuid = NESTED_SLICE_TRACK_UUID packet.track_descriptor.name = "My Nested Operations Timeline" # 添加 TrackEvent 数据包的帮助函数 def add_event(ts, event_type, name=None): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = NESTED_SLICE_TRACK_UUID if name: packet.track_event.name = name packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # --- 堆栈 1:Operation Alpha --- add_event(ts=2000, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Operation Alpha") add_event(ts=2050, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Alpha.LoadConfig") add_event(ts=2150, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Alpha.LoadConfig add_event(ts=2200, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Alpha.Execute") add_event(ts=2250, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Alpha.Execute.SubX") add_event(ts=2350, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Alpha.Execute.SubX add_event(ts=2400, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Alpha.Execute.SubY") add_event(ts=2500, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Alpha.Execute.SubY add_event(ts=2800, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Alpha.Execute add_event(ts=3000, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Operation Alpha # --- 堆栈 2:Operation Beta(在同一 track 上) --- add_event(ts=3200, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Operation Beta") add_event(ts=3250, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Beta.Initialize") add_event(ts=3350, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Beta.Initialize add_event(ts=3400, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Beta.Process") add_event(ts=3700, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Beta.Process add_event(ts=3800, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Operation Beta # --- 所有堆栈后的独立 slice --- add_event(ts=4000, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Cleanup") add_event(ts=4100, event_type=TrackEvent.TYPE_SLICE_END) # 关闭 Cleanup

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Nested Slices

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询这些嵌套 slices 并查看它们的层次结构:

SELECT ts, dur, name, depth FROM slice JOIN track ON slice.track_id = track.id WHERE track.name = 'My Nested Operations Timeline' ORDER BY ts;

异步 Slices 和重叠事件

许多系统处理异步操作,其中多个活动可以同时进行,并且它们的生命周期可以重叠而无需严格嵌套。示例包括:

在这些场景中,如果你使用的是开始/结束 slice 语义,则无法将所有这些重叠事件表示在单个 track 上,因为 TYPE_SLICE_END 总是关闭该特定 track 上最近打开的 slice。

Perfetto 对此进行建模的方式是将每个并发的、可能重叠的操作分配到其自己的唯一 track,具有唯一的 UUID。为了在 Perfetto UI 中实现这些相关异步操作的视觉分组,你可以给这些单独的操作 tracks 的每个 TrackDescriptor 指定相同的 name(例如,"Network Connections"或"File I/O")。slices 本身在这些 tracks 上可以有不同的名称(例如,"GET :/api/data"、"Read /config.txt")。

Perfetto UI 将组合或视觉上合并具有相同名称的 tracks。这是约定,可以由用户控制。有关更多详细信息,请参阅有关控制合并的部分: synthetic track event reference docs。

Python 示例:异步 Slices

假设我们正在跟踪活动网络连接。每个连接都是一个独立的异步事件。我们将给所有连接 tracks 相同的名称以鼓励 UI 对它们进行分组。我们将使用帮助函数来定义 tracks 和添加事件。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数:

点击展开/折叠 Python 代码
# 用于 UI 分组的所有单个连接 tracks 的通用名称 ASYNC_TRACK_GROUP_NAME = "HTTP Connections" # 使用唯一 UUID 定义新 track 的帮助函数 def define_track(group_name): track_uuid = get_uuid() packet = builder.add_packet() packet.track_descriptor.uuid = track_uuid packet.track_descriptor.name = group_name return track_uuid # 向特定 track 添加开始或结束 slice 事件的帮助函数 def add_slice_event(ts, event_type, event_track_uuid, name=None): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = event_track_uuid if name: packet.track_event.name = name packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # --- 网络连接 1 --- conn1_track_uuid = define_track(ASYNC_TRACK_GROUP_NAME) add_slice_event(ts=1000, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=conn1_track_uuid, name="GET /data/config") add_slice_event(ts=1500, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=conn1_track_uuid) # --- 网络连接 2(与连接 1 重叠) --- conn2_track_uuid = define_track(ASYNC_TRACK_GROUP_NAME) add_slice_event(ts=1100, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=conn2_track_uuid, name="POST /submit/form") add_slice_event(ts=2000, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=conn2_track_uuid) # --- 网络连接 3(在 1 结束后开始,与 2 重叠) --- conn3_track_uuid = define_track(ASYNC_TRACK_GROUP_NAME) add_slice_event(ts=1600, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=conn3_track_uuid, name="GET /status/check") add_slice_event(ts=2200, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=conn3_track_uuid)

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Asynchronous Slices

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询跨所有 HTTP 连接 tracks 的这些重叠 slices:

SELECT ts, dur, name FROM slice JOIN track ON slice.track_id = track.id WHERE track.name = 'HTTP Connections' ORDER BY ts;

Counters(随时间变化的值)

Counters 用于表示随时间变化的数值。它们非常适合跟踪非基于事件的 metrics 或状态,而是反映连续或采样数量的 metrics 或状态。

Counters 可以表示的常见示例包括:

要创建 counter track,你需要:

  1. 为你的 counter 定义一个 TrackDescriptor。此 track 需要一个 uuid、一个 name,重要的是,它的 counter 字段应该被填充。这告诉 Perfetto 将此 track 视为 counter。
  2. 发出带有 type: TYPE_COUNTER 的 TrackEvent packets。每个这样的 packet 应该有一个 timestamp 和一个 counter_value(整数)或一个 double_counter_value(双精度浮点数)。

Python 示例:Counters

假设我们想要跟踪随时间变化的未完成网络请求的数量。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# Counter track 的 UUID OUTSTANDING_REQUESTS_TRACK_UUID = get_uuid() # 1. 定义 Counter Track packet = builder.add_packet() track_desc = packet.track_descriptor track_desc.uuid = OUTSTANDING_REQUESTS_TRACK_UUID track_desc.name = "Outstanding Network Requests" # 要将此标记为 counter track,请设置 'counter' 字段为存在。 track_desc.counter.SetInParent() # 添加 counter 事件的帮助函数 def add_counter_event(ts, value): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = TrackEvent.TYPE_COUNTER packet.track_event.track_uuid = OUTSTANDING_REQUESTS_TRACK_UUID packet.track_event.counter_value = value packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 2. 随时间发出 counter 值 add_counter_event(ts=1000, value=0) add_counter_event(ts=1100, value=1) # 一个请求开始 add_counter_event(ts=1200, value=2) # 第二个请求开始 add_counter_event(ts=1300, value=3) # 第三个请求开始 add_counter_event(ts=1400, value=2) # 第一个请求完成 add_counter_event(ts=1500, value=2) # 无变化 add_counter_event(ts=1600, value=1) # 第二个请求完成 add_counter_event(ts=1700, value=0) # 第三个请求完成 add_counter_event(ts=1800, value=1) # 新请求开始 add_counter_event(ts=1900, value=0) # 最后一个请求完成

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Counters

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询 counter 值:

SELECT ts, value FROM counter JOIN track ON counter.track_id = track.id WHERE track.name = 'Outstanding Network Requests';

Flows(连接因果关系事件)

Flows 用于在视觉上连接具有显式因果或依赖关系的 slices,特别是当这些 slices 发生在不同的 tracks 上时(如不同的线程甚至不同的进程)。它们对于理解系统一个部分中的操作如何触发或启用另一个部分中的操作至关重要。

将 flows 视为从"原因"或"调度"事件到"结果"或"处理"事件绘制箭头。常见场景包括:

在 Perfetto 的 TrackEvent 模型中,你通过以下方式建立 flow:

  1. 将一个或多个唯一的 64 位 flow_id 分配给属于 flow 的 TrackEvent。此 ID 充当链接。
  2. 通常,flow_id 被添加到 TYPE_SLICE_BEGIN 或 TYPE_SLICE_END 事件,以标记从/到该 slice 的因果链接的起点或终点。
  3. 然后将相同的 flow_id 添加到另一个 TrackEvent(通常是不同 track 上的 TYPE_SLICE_BEGIN),以显示该因果链接操作的继续或处理。

Perfetto UI 将绘制箭头连接共享共同 flow_id 的 slices,使依赖链显式化。

替代方案:Correlation IDs 对于属于相同逻辑操作但不是因果连接的事件,请考虑使用 correlation IDs 代替或除了 flows。Correlation IDs 将相关事件视觉上分组(例如,使用一致的颜色)而不暗示因果关系。有关详细信息,请参阅高级指南中的 Linking Related Events with Correlation IDs 部分。

Python 示例:Flows

让我们建模一个简单的系统,其中"Request Handler" track 将工作调度到"Data Processor" track。我们将使用 flows 将请求调度链接到其处理,然后将处理完成链接回处理程序确认完成。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# --- 定义自定义 Tracks --- REQUEST_HANDLER_TRACK_UUID = get_uuid() DATA_PROCESSOR_TRACK_UUID = get_uuid() # Request Handler Track packet = builder.add_packet() packet.track_descriptor.uuid = REQUEST_HANDLER_TRACK_UUID packet.track_descriptor.name = "Request Handler" # Data Processor Track packet = builder.add_packet() packet.track_descriptor.uuid = DATA_PROCESSOR_TRACK_UUID packet.track_descriptor.name = "Data Processor" # 添加 slice 事件(BEGIN 或 END)的帮助函数 def add_slice_event(ts, event_type, event_track_uuid, name=None, flow_ids=None): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = event_track_uuid if name: packet.track_event.name = name if flow_ids: for flow_id in flow_ids: packet.track_event.flow_ids.append(flow_id) packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # --- 为因果链接定义唯一的 flow IDs --- DISPATCH_TO_PROCESS_FLOW_ID = get_uuid() PROCESS_COMPLETION_FLOW_ID = get_uuid() # 1. Request Handler:调度数据处理(第一个 flow 的起点) add_slice_event(ts=1000, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=REQUEST_HANDLER_TRACK_UUID, name="DispatchProcessing", flow_ids=[DISPATCH_TO_PROCESS_FLOW_ID]) add_slice_event(ts=1050, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=REQUEST_HANDLER_TRACK_UUID) # 2. Data Processor:处理数据(来自处理程序的调度的 flow) # 此 slice 的 BEGIN 事件包含 DISPATCH_TO_PROCESS_FLOW_ID,链接它。 # 它也从其 BEGIN 事件开始 PROCESS_COMPLETION_FLOW_ID。 add_slice_event(ts=1100, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=DATA_PROCESSOR_TRACK_UUID, name="ProcessDataItem", flow_ids=[DISPATCH_TO_PROCESS_FLOW_ID, PROCESS_COMPLETION_FLOW_ID]) add_slice_event(ts=1300, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=DATA_PROCESSOR_TRACK_UUID) # 3. Request Handler:确认完成(PROCESS_COMPLETION_FLOW_ID 在此结束) add_slice_event(ts=1350, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=REQUEST_HANDLER_TRACK_UUID, name="AcknowledgeCompletion", flow_ids=[PROCESS_COMPLETION_FLOW_ID]) add_slice_event(ts=1400, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=REQUEST_HANDLER_TRACK_UUID)

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Flows

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询 slices 之间的 flow 连接:

SELECT slice_out.name AS source_slice, slice_in.name AS dest_slice FROM flow JOIN slice AS slice_out ON flow.slice_out = slice_out.id JOIN slice AS slice_in ON flow.slice_in = slice_in.id;

使用层次结构对 Tracks 进行分组

随着 traces 变得更加复杂,你可能希望将相关的 tracks 组合在一起,以创建更有组织且易于理解的可视化。Perfetto 允许你使用 TrackDescriptor 中的 parent_uuid 字段定义 tracks 之间的父子关系。

这在以下情况下很有用:

父 track 可以服务于两个主要目的:

Perfetto UI 通常会将这些渲染为可展开的树。

Python 示例:Track Hierarchies

让我们创建一个层次结构:

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# --- 定义 Track UUIDs --- main_system_track_uuid = get_uuid() subsystem_a_track_uuid = get_uuid() subsystem_b_track_uuid = get_uuid() detail_a1_track_uuid = get_uuid() # 定义 TrackDescriptor 的帮助函数 def define_custom_track(track_uuid, name, parent_track_uuid=None): packet = builder.add_packet() desc = packet.track_descriptor desc.uuid = track_uuid desc.name = name if parent_track_uuid: desc.parent_uuid = parent_track_uuid # 添加 slice 事件的帮助函数 def add_slice_event(ts, event_type, event_track_uuid, name=None): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = event_track_uuid if name: packet.track_event.name = name packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 1. 定义 Track 层次结构 define_custom_track(main_system_track_uuid, "Main System") define_custom_track(subsystem_a_track_uuid, "Subsystem A", parent_track_uuid=main_system_track_uuid) define_custom_track(subsystem_b_track_uuid, "Subsystem B", parent_track_uuid=main_system_track_uuid) define_custom_track(detail_a1_track_uuid, "Detail A.1", parent_track_uuid=subsystem_a_track_uuid) # 2. 在层次结构中的各种 tracks 上发出 slices # 在父"Main System" track 上的 slice(摘要/总体活动) add_slice_event(ts=4800, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=main_system_track_uuid, name="System Initialization Phase") add_slice_event(ts=7000, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=main_system_track_uuid) # 在"Detail A.1"(子"Subsystem A")上的 slice add_slice_event(ts=5000, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=detail_a1_track_uuid, name="Activity in A.1") add_slice_event(ts=5500, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=detail_a1_track_uuid) # 在"Subsystem B"上的 slice add_slice_event(ts=6000, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=subsystem_b_track_uuid, name="Work in Subsystem B") add_slice_event(ts=6200, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=subsystem_b_track_uuid) # 在"Detail A.1"上的另一个 slice add_slice_event(ts=5600, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=detail_a1_track_uuid, name="Further Activity in A.1") add_slice_event(ts=5800, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=detail_a1_track_uuid)

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Grouping Tracks with Hierarchies

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询跨 track 层次结构的 slices:

SELECT slice.ts, slice.dur, slice.name, track.name AS track_name FROM slice JOIN track ON slice.track_id = track.id WHERE track.name IN ('Main System', 'Subsystem A', 'Subsystem B', 'Detail A.1') ORDER BY slice.ts;

用于瀑布图 / Trace 视图的 Track 层次结构

Track 层次结构的另一个强大用途是可视化复杂操作或请求的分解,类似于分布式 tracing 系统中显示"trace 视图"或"span 视图"的方式。这在操作涉及跨不同逻辑组件的顺序或并行步骤时非常有用,并且你希望以瀑布图或甘特图样式查看这些步骤的时间和关系。

在此模型中:

Python 示例:服务请求分解

假设前端服务发出一个请求,涉及调用两个后端服务:Authentication Service 和 Data Service。只有在 Authentication Service 调用完成后才能进行 Data Service 调用。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# --- 定义 Track UUIDs --- root_request_track_uuid = get_uuid() auth_service_call_track_uuid = get_uuid() data_service_call_track_uuid = get_uuid() # data_service_call 中的内部步骤的 UUID data_service_internal_step_track_uuid = get_uuid() # 定义 TrackDescriptor 的帮助函数 def define_custom_track(track_uuid, name, parent_track_uuid=None): packet = builder.add_packet() desc = packet.track_descriptor desc.uuid = track_uuid desc.name = name if parent_track_uuid: desc.parent_uuid = parent_track_uuid # 添加 slice 事件的帮助函数 def add_slice_event(ts, event_type, event_track_uuid, name=None): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = event_track_uuid if name: packet.track_event.name = name packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 1. 定义整体请求的根 Track define_custom_track(root_request_track_uuid, "Frontend Request: /api/user/profile") # 在其自己的 track 上为前端请求的总持续时间添加 slice add_slice_event(ts=10000, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=root_request_track_uuid, name="Total Request Duration") # 2. 将每个服务调用(span)的子 tracks 定义为根请求的子项 define_custom_track(auth_service_call_track_uuid, "Call: AuthService.AuthenticateUser", parent_track_uuid=root_request_track_uuid) define_custom_track(data_service_call_track_uuid, "Call: DataService.GetUserData", parent_track_uuid=root_request_track_uuid) # 3. 在这些服务调用 tracks 上发出 slices # Auth Service Call add_slice_event(ts=10100, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=auth_service_call_track_uuid, name="AuthService.AuthenticateUser") add_slice_event(ts=10300, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=auth_service_call_track_uuid) # Data Service Call(在 Auth 完成后开始) add_slice_event(ts=10350, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=data_service_call_track_uuid, name="DataService.GetUserData") # 模拟 DataService.GetUserData 内的内部步骤,显示在其自己的子 track 上 # 此 track 将是"Call: DataService.GetUserData" track 的子项。 define_custom_track(data_service_internal_step_track_uuid, "Internal: QueryDatabase", parent_track_uuid=data_service_call_track_uuid) add_slice_event(ts=10400, event_type=TrackEvent.TYPE_SLICE_BEGIN, event_track_uuid=data_service_internal_step_track_uuid, name="QueryDatabase") add_slice_event(ts=10550, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=data_service_internal_step_track_uuid) add_slice_event(ts=10600, event_type=TrackEvent.TYPE_SLICE_END, # DataService.GetUserData 的结束 event_track_uuid=data_service_call_track_uuid) # 前端请求总计的结束 add_slice_event(ts=10700, event_type=TrackEvent.TYPE_SLICE_END, event_track_uuid=root_request_track_uuid)

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Track Hierarchies for Waterfall / Trace Views

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询此请求分解以分析时间:

SELECT slice.ts, slice.dur, slice.name, track.name AS service FROM slice JOIN track ON slice.track_id = track.id WHERE track.name LIKE '%Request%' OR track.name LIKE '%Service%' ORDER BY slice.ts;

向事件添加调试注解

调试注解允许你将任意键值数据附加到任何 TrackEvent。它们在你检查 Perfetto UI 中的各个事件时出现,对于提供关于特定 slices 或 instants 期间发生情况的额外上下文非常有用。

调试注解可用于:

调试注解支持各种数据类型,包括基本值(字符串、整数、布尔值、双精度浮点数)、嵌套字典和数组。它们使用 DebugAnnotation protobuf 消息,可以表示复杂的嵌套结构。

Python 示例:基本调试注解

此示例演示如何向 track 事件添加简单的键值调试注解。这对于附加附加信息(如对象 IDs、状态值或其他上下文数据)很有用。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# 为你的自定义 track 定义唯一的 UUID DEBUG_TRACK_UUID = 87654321 # 1. 定义自定义 Track packet = builder.add_packet() packet.track_descriptor.uuid = DEBUG_TRACK_UUID packet.track_descriptor.name = "Debug Annotations Example" # 添加带有调试注解的 slice 事件的帮助函数 def add_slice_with_debug_annotations(ts, event_type, name=None, debug_annotations=None): packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = DEBUG_TRACK_UUID if name: packet.track_event.name = name # 添加调试注解 if debug_annotations: for key, value in debug_annotations.items(): annotation = packet.track_event.debug_annotations.add() annotation.name = key # 根据类型设置适当的值字段 if isinstance(value, bool): annotation.bool_value = value elif isinstance(value, int): annotation.int_value = value elif isinstance(value, float): annotation.double_value = value elif isinstance(value, str): annotation.string_value = value packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 2. 创建带有各种调试注解的 slices add_slice_with_debug_annotations( ts=1000, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="Database Query", debug_annotations={ "query_id": 12345, "table_name": "users", "is_cached": False, "timeout_ms": 5000.0 } ) add_slice_with_debug_annotations( ts=1200, event_type=TrackEvent.TYPE_SLICE_END ) # 另一个带有不同类型注解的示例 add_slice_with_debug_annotations( ts=1500, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="HTTP Request", debug_annotations={ "method": "POST", "url": "/api/users/create", "content_length": 2048, "keep_alive": True } ) add_slice_with_debug_annotations( ts=1800, event_type=TrackEvent.TYPE_SLICE_END )

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Adding Debug Annotations

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询调试注解:

SELECT slice.name, EXTRACT_ARG(slice.arg_set_id, 'debug.query_id') AS query_id FROM slice JOIN track ON slice.track_id = track.id WHERE track.name = 'Debug Annotations Example';

Python 示例:嵌套调试注解

调试注解可以表示复杂的嵌套数据结构,包括字典和数组。这在需要附加结构化信息(如配置对象、值数组或层次数据)时非常有用。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# 为你的自定义 track 定义唯一的 UUID NESTED_DEBUG_TRACK_UUID = 87654322 # 1. 定义自定义 Track packet = builder.add_packet() packet.track_descriptor.uuid = NESTED_DEBUG_TRACK_UUID packet.track_descriptor.name = "Nested Debug Annotations" # 2. 创建带有嵌套调试注解的 slice packet = builder.add_packet() packet.timestamp = 2000 packet.track_event.type = TrackEvent.TYPE_SLICE_BEGIN packet.track_event.track_uuid = NESTED_DEBUG_TRACK_UUID packet.track_event.name = "Complex Operation" # 添加带有嵌套结构的字典注解 config_annotation = packet.track_event.debug_annotations.add() config_annotation.name = "config" # 添加字典条目 db_entry = config_annotation.dict_entries.add() db_entry.name = "database" db_entry.string_value = "postgres://localhost:5432/mydb" timeout_entry = config_annotation.dict_entries.add() timeout_entry.name = "timeout_ms" timeout_entry.int_value = 30000 retry_entry = config_annotation.dict_entries.add() retry_entry.name = "retry_enabled" retry_entry.bool_value = True # 添加数组注解 servers_annotation = packet.track_event.debug_annotations.add() servers_annotation.name = "server_list" # 添加数组值 server1 = servers_annotation.array_values.add() server1.string_value = "server-1.example.com" server2 = servers_annotation.array_values.add() server2.string_value = "server-2.example.com" server3 = servers_annotation.array_values.add() server3.string_value = "server-3.example.com" packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 结束 slice packet = builder.add_packet() packet.timestamp = 2500 packet.track_event.type = TrackEvent.TYPE_SLICE_END packet.track_event.track_uuid = NESTED_DEBUG_TRACK_UUID packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Nested Debug Annotations

你可以在 Perfetto UI 的 Query 标签页中使用 SQL 或使用 Trace Processor 查询嵌套调试注解:

SELECT slice.name, EXTRACT_ARG(slice.arg_set_id, 'debug.config.database') AS database, EXTRACT_ARG(slice.arg_set_id, 'debug.server_list[0]') AS first_server FROM slice JOIN track ON slice.track_id = track.id WHERE track.name = 'Nested Debug Annotations';

将调用栈附加到事件

调用栈(也称为堆栈跟踪或回溯)显示导致特定事件的函数调用序列。将调用栈添加到你的 trace 事件对于理解触发特定操作的代码路径非常宝贵。

有两种不同的方式将调用栈与事件关联:

  1. 内联调用栈:将堆栈帧直接嵌入每个事件中,包含函数名称和可选的源位置。这很简单,不需要设置,当 trace 大小不是问题或调用栈是唯一的时非常理想。
  2. Interned 调用栈:定义一次调用栈结构并从多个事件中通过 ID 引用它。当调用栈频繁重复或需要二进制/映射信息以进行符号化时,这要高效得多。

本指南涵盖内联调用栈,非常适合入门。对于重复的调用栈或需要二进制映射信息时,请改用 interned callstacks。

Python 示例:Inline Callstacks

每个帧包括函数名称,以及可选的源文件和行号。

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数。

点击展开/折叠 Python 代码
# 为你的自定义 track 定义唯一的 UUID CALLSTACK_TRACK_UUID = 98765432 def emit_track_event( ts, event_type, name=None, frames=None, ): """Helper to write a TrackEvent with an optional inline callstack.""" packet = builder.add_packet() packet.timestamp = ts packet.track_event.type = event_type packet.track_event.track_uuid = CALLSTACK_TRACK_UUID if name is not None: packet.track_event.name = name if frames: for function, source, line in frames: frame = packet.track_event.callstack.frames.add() frame.function_name = function if source: frame.source_file = source if line is not None: frame.line_number = line packet.trusted_packet_sequence_id = TRUSTED_PACKET_SEQUENCE_ID # 1. 定义自定义 Track packet = builder.add_packet() packet.track_descriptor.uuid = CALLSTACK_TRACK_UUID packet.track_descriptor.name = "Operations with Callstacks" # 2. 创建带有内联调用栈的 slice emit_track_event( ts=3000, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="ProcessRequest", frames=[ ("main", "/src/app.cc", 42), ("HandleIncomingRequests", "/src/server.cc", 128), ("ProcessRequest", "/src/request_handler.cc", 256), ], ) # 在 slice 完成时结束带有调用栈捕获的 slice emit_track_event( ts=3500, event_type=TrackEvent.TYPE_SLICE_END, frames=[ ("main", None, None), ("HandleIncomingRequests", None, None), ("FinalizeRequest", "/src/request_handler.cc", 512), ], ) # 3. 另一个带有最小调用栈的 slice(仅函数名称) emit_track_event( ts=4000, event_type=TrackEvent.TYPE_SLICE_BEGIN, name="AllocateMemory", frames=[ ("main", None, None), ("HandleIncomingRequests", None, None), ("AllocateMemory", None, None), ], ) # 结束 slice emit_track_event( ts=4200, event_type=TrackEvent.TYPE_SLICE_END, )

NOTE: 帧从最外层(堆栈底部,例如 main())到最内层(堆栈顶部,事件发生的地方)排序。

当你在 slice 结束事件上提供调用栈时,Trace Processor 会将其与开始调用栈分开存储(存储为 end_callsite_id 而非 callsite_id)。这对于快速比较进入/退出堆栈非常方便。

运行脚本后,打开生成的 my_custom_trace.pftrace 文件(在 Perfetto UI 中)将显示以下输出:

Inline Callstacks

请注意,你还可以进行"区域选择"(又称框选)以获取调用栈的火焰图:

Inline Callstacks Area Select

默认情况下,每个调用栈在火焰图中只计数一次。如果要为每次出现赋予一个值(例如分配的字节数),请参见 Weighted Callstacks and Custom Measures。

添加 Trace 级别元数据

有时你想附加的数据不属于任何特定的事件、slice 或 track,而是描述整个 trace,比如 build ID、生成 trace 的工具版本或采集它的机器主机名。Perfetto 的 TraceAttributes 消息允许你将此类 trace 级别的元数据直接附加到 TracePacket,无需 track 或时间戳。

Trace 级别元数据适用于:

要附加 trace 级别元数据:

  1. 对于每条元数据,向 TracePacket 的 trace_attributes.attribute 字段添加一个 Attribute,设置其 key 以及 string_value 或 long_value。
  2. 使用反向域名风格的前缀为你的 key 设置命名空间(例如 com.example.myapp.build_id),以免与其他生产者使用的 key 冲突。
  3. 在 trace 的任何位置发出这些数据包。不需要 timestamp、track_uuid 或 trusted_packet_sequence_id。

NOTE: 仅支持字符串和 64 位整数值。如果相同的 key 被多次设置,无论是在单个数据包内还是跨多个数据包,最后写入的值生效。

除了写入数据包外,还可以在录制 trace 时通过 perfetto --add-attribute key=value 设置属性;两者最终都会出现在相同的 trace_attribute.* 行中。Trace archive 还可以通过 attributes 部分进行注释(见 trace manifest),该部分使用自己的 manifest_attribute.* 命名空间。

Python 示例:Trace Metadata

将以下 Python 代码复制到 trace_converter_template.py 脚本中的 populate_packets(builder) 函数中。

点击展开/折叠 Python 代码
def add_attribute(key, value): packet = builder.add_packet() attr = packet.trace_attributes.attribute.add() attr.key = key if isinstance(value, str): attr.string_value = value else: attr.long_value = value add_attribute("com.example.pipeline.version", "2.1.0") add_attribute("com.example.pipeline.num_workers", 8) add_attribute("com.example.hostname", "build-server-01")

与本页中的其他示例不同,Timeline 视图中没有 track 或 slice 可看。相反,运行脚本并打开 my_custom_trace.pftrace(在 Perfetto UI 中)后,这些属性会显示在 "Overview" 页面上(可从左侧侧边栏进入),位于 "Metadata" 标签页的 "Trace Attributes" 部分下,与 trace 的其他元数据一起。

Overview 页面中的 Trace 属性

在 Trace Processor 中,每个属性都成为 metadata 表中的一行,其 key 带有 trace_attribute. 前缀,以将自定义属性与内置元数据分开。你可以在 Perfetto UI 的 Query 标签页中使用 SQL 查询 trace 级别元数据,或使用 Trace Processor:

SELECT name, ifnull(str_value, cast(int_value as text)) AS value FROM metadata WHERE name GLOB 'trace_attribute.*';

将数据导出

一旦你的 trace 加载完成,trace processor 可以使用 export 子命令将解析后的表写回出来:

trace_processor export perfetto -o archive.tar my_custom_trace.pftrace trace_processor export arrow_tar -o tables.tar my_custom_trace.pftrace trace_processor export sqlite -o trace.db my_custom_trace.pftrace

perfetto 可以由同一版本的 trace processor 实例重新加载,arrow_tar 面向 pandas、Polars 或 pyarrow 等外部工具,而 sqlite 会生成一个任何 SQLite 工具都能打开的数据库。关于如何在各种格式之间选择以及每种格式的具体内容,请参阅导出 trace 数据和 trace_processor CLI 参考。

下一步

你现在了解了如何使用 Python 和 TrackEvent 将自定义时间戳数据转换为 Perfetto traces。通过这些技术,你可以表示 slices、counters、flows、track 层次结构、调试注解、调用栈和 trace 级别元数据。

一旦你将自定义数据转换为 Perfetto trace 格式(.pftrace 文件),你可以: