1 /*
2 * Copyright (C) 2018 The Android Open Source Project
3 *
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
7 *
8 * http://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16
17 #include "src/trace_processor/importers/proto/proto_trace_reader.h"
18
19 #include <string>
20
21 #include "perfetto/base/build_config.h"
22 #include "perfetto/base/logging.h"
23 #include "perfetto/ext/base/optional.h"
24 #include "perfetto/ext/base/string_view.h"
25 #include "perfetto/ext/base/utils.h"
26 #include "perfetto/protozero/proto_decoder.h"
27 #include "perfetto/protozero/proto_utils.h"
28 #include "perfetto/trace_processor/status.h"
29 #include "src/trace_processor/importers/common/clock_tracker.h"
30 #include "src/trace_processor/importers/common/event_tracker.h"
31 #include "src/trace_processor/importers/common/track_tracker.h"
32 #include "src/trace_processor/importers/ftrace/ftrace_module.h"
33 #include "src/trace_processor/importers/proto/metadata_tracker.h"
34 #include "src/trace_processor/importers/proto/packet_sequence_state.h"
35 #include "src/trace_processor/importers/proto/proto_incremental_state.h"
36 #include "src/trace_processor/storage/stats.h"
37 #include "src/trace_processor/storage/trace_storage.h"
38 #include "src/trace_processor/trace_sorter.h"
39 #include "src/trace_processor/util/descriptors.h"
40 #include "src/trace_processor/util/gzip_utils.h"
41
42 #include "protos/perfetto/common/builtin_clock.pbzero.h"
43 #include "protos/perfetto/config/trace_config.pbzero.h"
44 #include "protos/perfetto/trace/clock_snapshot.pbzero.h"
45 #include "protos/perfetto/trace/extension_descriptor.pbzero.h"
46 #include "protos/perfetto/trace/perfetto/tracing_service_event.pbzero.h"
47 #include "protos/perfetto/trace/profiling/profile_common.pbzero.h"
48 #include "protos/perfetto/trace/trace.pbzero.h"
49 #include "protos/perfetto/trace/trace_packet.pbzero.h"
50
51 namespace perfetto {
52 namespace trace_processor {
53
ProtoTraceReader(TraceProcessorContext * ctx)54 ProtoTraceReader::ProtoTraceReader(TraceProcessorContext* ctx)
55 : context_(ctx) {}
56 ProtoTraceReader::~ProtoTraceReader() = default;
57
Parse(TraceBlobView blob)58 util::Status ProtoTraceReader::Parse(TraceBlobView blob) {
59 return tokenizer_.Tokenize(std::move(blob), [this](TraceBlobView packet) {
60 return ParsePacket(std::move(packet));
61 });
62 }
63
ParseExtensionDescriptor(ConstBytes descriptor)64 util::Status ProtoTraceReader::ParseExtensionDescriptor(ConstBytes descriptor) {
65 protos::pbzero::ExtensionDescriptor::Decoder decoder(descriptor.data,
66 descriptor.size);
67
68 auto extension = decoder.extension_set();
69 return context_->descriptor_pool_->AddFromFileDescriptorSet(
70 extension.data, extension.size,
71 /*skip_prefixes*/ {},
72 /*merge_existing_messages=*/true);
73 }
74
ParsePacket(TraceBlobView packet)75 util::Status ProtoTraceReader::ParsePacket(TraceBlobView packet) {
76 protos::pbzero::TracePacket::Decoder decoder(packet.data(), packet.length());
77 if (PERFETTO_UNLIKELY(decoder.bytes_left())) {
78 return util::ErrStatus(
79 "Failed to parse proto packet fully; the trace is probably corrupt.");
80 }
81
82 // Any compressed packets should have been handled by the tokenizer.
83 PERFETTO_CHECK(!decoder.has_compressed_packets());
84
85 const uint32_t seq_id = decoder.trusted_packet_sequence_id();
86 auto* state = GetIncrementalStateForPacketSequence(seq_id);
87
88 uint32_t sequence_flags = decoder.sequence_flags();
89
90 if (decoder.incremental_state_cleared() ||
91 sequence_flags &
92 protos::pbzero::TracePacket::SEQ_INCREMENTAL_STATE_CLEARED) {
93 HandleIncrementalStateCleared(decoder);
94 } else if (decoder.previous_packet_dropped()) {
95 HandlePreviousPacketDropped(decoder);
96 }
97
98 // It is important that we parse defaults before parsing other fields such as
99 // the timestamp, since the defaults could affect them.
100 if (decoder.has_trace_packet_defaults()) {
101 auto field = decoder.trace_packet_defaults();
102 ParseTracePacketDefaults(decoder, packet.slice(field.data, field.size));
103 }
104
105 if (decoder.has_interned_data()) {
106 auto field = decoder.interned_data();
107 ParseInternedData(decoder, packet.slice(field.data, field.size));
108 }
109
110 if (decoder.has_clock_snapshot()) {
111 return ParseClockSnapshot(decoder.clock_snapshot(),
112 decoder.trusted_packet_sequence_id());
113 }
114
115 if (decoder.has_service_event()) {
116 PERFETTO_DCHECK(decoder.has_timestamp());
117 int64_t ts = static_cast<int64_t>(decoder.timestamp());
118 return ParseServiceEvent(ts, decoder.service_event());
119 }
120
121 if (decoder.has_extension_descriptor()) {
122 return ParseExtensionDescriptor(decoder.extension_descriptor());
123 }
124
125 if (decoder.sequence_flags() &
126 protos::pbzero::TracePacket::SEQ_NEEDS_INCREMENTAL_STATE) {
127 if (!seq_id) {
128 return util::ErrStatus(
129 "TracePacket specified SEQ_NEEDS_INCREMENTAL_STATE but the "
130 "TraceWriter's sequence_id is zero (the service is "
131 "probably too old)");
132 }
133
134 if (!state->IsIncrementalStateValid()) {
135 context_->storage->IncrementStats(stats::tokenizer_skipped_packets);
136 return util::OkStatus();
137 }
138 }
139
140 // Workaround a bug in the frame timeline traces which is emitting packets
141 // with zero timestamp (b/179905685).
142 // TODO(primiano): around mid-2021 there should be no traces that have this
143 // bug and we should be able to remove this workaround.
144 if (decoder.has_frame_timeline_event() && decoder.timestamp() == 0) {
145 context_->storage->IncrementStats(
146 stats::frame_timeline_event_parser_errors);
147 return util::OkStatus();
148 }
149
150 protos::pbzero::TracePacketDefaults::Decoder* defaults =
151 state->current_generation()->GetTracePacketDefaults();
152
153 int64_t timestamp;
154 if (decoder.has_timestamp()) {
155 timestamp = static_cast<int64_t>(decoder.timestamp());
156
157 uint32_t timestamp_clock_id =
158 decoder.has_timestamp_clock_id()
159 ? decoder.timestamp_clock_id()
160 : (defaults ? defaults->timestamp_clock_id() : 0);
161
162 if ((decoder.has_chrome_events() || decoder.has_chrome_metadata()) &&
163 (!timestamp_clock_id ||
164 timestamp_clock_id == protos::pbzero::BUILTIN_CLOCK_MONOTONIC)) {
165 // Chrome event timestamps are in MONOTONIC domain, but may occur in
166 // traces where (a) no clock snapshots exist or (b) no clock_id is
167 // specified for their timestamps. Adjust to trace time if we have a clock
168 // snapshot.
169 // TODO(eseckler): Set timestamp_clock_id and emit ClockSnapshots in
170 // chrome and then remove this.
171 auto trace_ts = context_->clock_tracker->ToTraceTime(
172 protos::pbzero::BUILTIN_CLOCK_MONOTONIC, timestamp);
173 if (trace_ts.has_value())
174 timestamp = trace_ts.value();
175 } else if (timestamp_clock_id) {
176 // If the TracePacket specifies a non-zero clock-id, translate the
177 // timestamp into the trace-time clock domain.
178 ClockTracker::ClockId converted_clock_id = timestamp_clock_id;
179 bool is_seq_scoped =
180 ClockTracker::IsReservedSeqScopedClockId(converted_clock_id);
181 if (is_seq_scoped) {
182 if (!seq_id) {
183 return util::ErrStatus(
184 "TracePacket specified a sequence-local clock id (%" PRIu32
185 ") but the TraceWriter's sequence_id is zero (the service is "
186 "probably too old)",
187 timestamp_clock_id);
188 }
189 converted_clock_id =
190 ClockTracker::SeqScopedClockIdToGlobal(seq_id, timestamp_clock_id);
191 }
192 auto trace_ts =
193 context_->clock_tracker->ToTraceTime(converted_clock_id, timestamp);
194 if (!trace_ts.has_value()) {
195 // ToTraceTime() will increase the |clock_sync_failure| stat on failure.
196 // We don't return an error here as it will cause the trace to stop
197 // parsing. Instead, we rely on the stat increment in ToTraceTime() to
198 // inform the user about the error.
199 return util::OkStatus();
200 }
201 timestamp = trace_ts.value();
202 }
203 } else {
204 timestamp = std::max(latest_timestamp_, context_->sorter->max_timestamp());
205 }
206 latest_timestamp_ = std::max(timestamp, latest_timestamp_);
207
208 auto& modules = context_->modules_by_field;
209 for (uint32_t field_id = 1; field_id < modules.size(); ++field_id) {
210 if (!modules[field_id].empty() && decoder.Get(field_id).valid()) {
211 for (ProtoImporterModule* module : modules[field_id]) {
212 ModuleResult res = module->TokenizePacket(decoder, &packet, timestamp,
213 state, field_id);
214 if (!res.ignored())
215 return res.ToStatus();
216 }
217 }
218 }
219
220 if (decoder.has_trace_config()) {
221 ParseTraceConfig(decoder.trace_config());
222 }
223
224 // Use parent data and length because we want to parse this again
225 // later to get the exact type of the packet.
226 context_->sorter->PushTracePacket(timestamp, state, std::move(packet));
227
228 return util::OkStatus();
229 }
230
ParseTraceConfig(protozero::ConstBytes blob)231 void ProtoTraceReader::ParseTraceConfig(protozero::ConstBytes blob) {
232 protos::pbzero::TraceConfig::Decoder trace_config(blob);
233 if (trace_config.write_into_file() && !trace_config.flush_period_ms()) {
234 PERFETTO_ELOG(
235 "It is strongly recommended to have flush_period_ms set when "
236 "write_into_file is turned on. This trace will be loaded fully "
237 "into memory before sorting which increases the likliehoold of "
238 "OOMs.");
239 }
240 }
241
HandleIncrementalStateCleared(const protos::pbzero::TracePacket::Decoder & packet_decoder)242 void ProtoTraceReader::HandleIncrementalStateCleared(
243 const protos::pbzero::TracePacket::Decoder& packet_decoder) {
244 if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
245 PERFETTO_ELOG(
246 "incremental_state_cleared without trusted_packet_sequence_id");
247 context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
248 return;
249 }
250 GetIncrementalStateForPacketSequence(
251 packet_decoder.trusted_packet_sequence_id())
252 ->OnIncrementalStateCleared();
253 for (auto& module : context_->modules) {
254 module->OnIncrementalStateCleared(
255 packet_decoder.trusted_packet_sequence_id());
256 }
257 }
258
HandlePreviousPacketDropped(const protos::pbzero::TracePacket::Decoder & packet_decoder)259 void ProtoTraceReader::HandlePreviousPacketDropped(
260 const protos::pbzero::TracePacket::Decoder& packet_decoder) {
261 if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
262 PERFETTO_ELOG("previous_packet_dropped without trusted_packet_sequence_id");
263 context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
264 return;
265 }
266 GetIncrementalStateForPacketSequence(
267 packet_decoder.trusted_packet_sequence_id())
268 ->OnPacketLoss();
269 }
270
ParseTracePacketDefaults(const protos::pbzero::TracePacket_Decoder & packet_decoder,TraceBlobView trace_packet_defaults)271 void ProtoTraceReader::ParseTracePacketDefaults(
272 const protos::pbzero::TracePacket_Decoder& packet_decoder,
273 TraceBlobView trace_packet_defaults) {
274 if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
275 PERFETTO_ELOG(
276 "TracePacketDefaults packet without trusted_packet_sequence_id");
277 context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
278 return;
279 }
280
281 auto* state = GetIncrementalStateForPacketSequence(
282 packet_decoder.trusted_packet_sequence_id());
283 state->UpdateTracePacketDefaults(std::move(trace_packet_defaults));
284 }
285
ParseInternedData(const protos::pbzero::TracePacket::Decoder & packet_decoder,TraceBlobView interned_data)286 void ProtoTraceReader::ParseInternedData(
287 const protos::pbzero::TracePacket::Decoder& packet_decoder,
288 TraceBlobView interned_data) {
289 if (PERFETTO_UNLIKELY(!packet_decoder.has_trusted_packet_sequence_id())) {
290 PERFETTO_ELOG("InternedData packet without trusted_packet_sequence_id");
291 context_->storage->IncrementStats(stats::interned_data_tokenizer_errors);
292 return;
293 }
294
295 auto* state = GetIncrementalStateForPacketSequence(
296 packet_decoder.trusted_packet_sequence_id());
297
298 // Don't parse interned data entries until incremental state is valid, because
299 // they could otherwise be associated with the wrong generation in the state.
300 if (!state->IsIncrementalStateValid()) {
301 context_->storage->IncrementStats(stats::tokenizer_skipped_packets);
302 return;
303 }
304
305 // Store references to interned data submessages into the sequence's state.
306 protozero::ProtoDecoder decoder(interned_data.data(), interned_data.length());
307 for (protozero::Field f = decoder.ReadField(); f.valid();
308 f = decoder.ReadField()) {
309 auto bytes = f.as_bytes();
310 state->InternMessage(f.id(), interned_data.slice(bytes.data, bytes.size));
311 }
312 }
313
ParseClockSnapshot(ConstBytes blob,uint32_t seq_id)314 util::Status ProtoTraceReader::ParseClockSnapshot(ConstBytes blob,
315 uint32_t seq_id) {
316 std::vector<ClockTracker::ClockValue> clocks;
317 protos::pbzero::ClockSnapshot::Decoder evt(blob.data, blob.size);
318 if (evt.primary_trace_clock()) {
319 context_->clock_tracker->SetTraceTimeClock(
320 static_cast<ClockTracker::ClockId>(evt.primary_trace_clock()));
321 }
322 for (auto it = evt.clocks(); it; ++it) {
323 protos::pbzero::ClockSnapshot::Clock::Decoder clk(*it);
324 ClockTracker::ClockId clock_id = clk.clock_id();
325 if (ClockTracker::IsReservedSeqScopedClockId(clk.clock_id())) {
326 if (!seq_id) {
327 return util::ErrStatus(
328 "ClockSnapshot packet is specifying a sequence-scoped clock id "
329 "(%" PRIu64 ") but the TracePacket sequence_id is zero",
330 clock_id);
331 }
332 clock_id = ClockTracker::SeqScopedClockIdToGlobal(seq_id, clk.clock_id());
333 }
334 int64_t unit_multiplier_ns =
335 clk.unit_multiplier_ns()
336 ? static_cast<int64_t>(clk.unit_multiplier_ns())
337 : 1;
338 clocks.emplace_back(clock_id, clk.timestamp(), unit_multiplier_ns,
339 clk.is_incremental());
340 }
341
342 uint32_t snapshot_id = context_->clock_tracker->AddSnapshot(clocks);
343
344 // Add the all the clock values to the clock snapshot table.
345 base::Optional<int64_t> trace_ts_for_check;
346 for (const auto& clock : clocks) {
347 // If the clock is incremental, we need to use 0 to map correctly to
348 // |absolute_timestamp|.
349 int64_t ts_to_convert = clock.is_incremental ? 0 : clock.absolute_timestamp;
350 base::Optional<int64_t> opt_trace_ts =
351 context_->clock_tracker->ToTraceTime(clock.clock_id, ts_to_convert);
352 if (!opt_trace_ts) {
353 // This can happen if |AddSnapshot| failed to resolve this clock. Just
354 // ignore this and move on.
355 continue;
356 }
357
358 // Double check that all the clocks in this snapshot resolve to the same
359 // trace timestamp value.
360 PERFETTO_DCHECK(!trace_ts_for_check || opt_trace_ts == trace_ts_for_check);
361 trace_ts_for_check = *opt_trace_ts;
362
363 tables::ClockSnapshotTable::Row row;
364 row.ts = *opt_trace_ts;
365 row.clock_id = static_cast<int64_t>(clock.clock_id);
366 row.clock_value = clock.absolute_timestamp;
367 row.clock_name = GetBuiltinClockNameOrNull(clock.clock_id);
368 row.snapshot_id = snapshot_id;
369
370 auto* snapshot_table = context_->storage->mutable_clock_snapshot_table();
371 snapshot_table->Insert(row);
372 }
373 return util::OkStatus();
374 }
375
GetBuiltinClockNameOrNull(uint64_t clock_id)376 base::Optional<StringId> ProtoTraceReader::GetBuiltinClockNameOrNull(
377 uint64_t clock_id) {
378 switch (clock_id) {
379 case protos::pbzero::ClockSnapshot::Clock::REALTIME:
380 return context_->storage->InternString("REALTIME");
381 case protos::pbzero::ClockSnapshot::Clock::REALTIME_COARSE:
382 return context_->storage->InternString("REALTIME_COARSE");
383 case protos::pbzero::ClockSnapshot::Clock::MONOTONIC:
384 return context_->storage->InternString("MONOTONIC");
385 case protos::pbzero::ClockSnapshot::Clock::MONOTONIC_COARSE:
386 return context_->storage->InternString("MONOTONIC_COARSE");
387 case protos::pbzero::ClockSnapshot::Clock::MONOTONIC_RAW:
388 return context_->storage->InternString("MONOTONIC_RAW");
389 case protos::pbzero::ClockSnapshot::Clock::BOOTTIME:
390 return context_->storage->InternString("BOOTTIME");
391 default:
392 return base::nullopt;
393 }
394 }
395
ParseServiceEvent(int64_t ts,ConstBytes blob)396 util::Status ProtoTraceReader::ParseServiceEvent(int64_t ts, ConstBytes blob) {
397 protos::pbzero::TracingServiceEvent::Decoder tse(blob);
398 if (tse.tracing_started()) {
399 context_->metadata_tracker->SetMetadata(metadata::tracing_started_ns,
400 Variadic::Integer(ts));
401 }
402 if (tse.tracing_disabled()) {
403 context_->metadata_tracker->SetMetadata(metadata::tracing_disabled_ns,
404 Variadic::Integer(ts));
405 }
406 if (tse.all_data_sources_started()) {
407 context_->metadata_tracker->SetMetadata(
408 metadata::all_data_source_started_ns, Variadic::Integer(ts));
409 }
410 if (tse.all_data_sources_flushed()) {
411 context_->sorter->NotifyFlushEvent();
412 }
413 if (tse.read_tracing_buffers_completed()) {
414 context_->sorter->NotifyReadBufferEvent();
415 }
416 return util::OkStatus();
417 }
418
NotifyEndOfFile()419 void ProtoTraceReader::NotifyEndOfFile() {}
420
421 } // namespace trace_processor
422 } // namespace perfetto
423