src/trace_processor/read_trace.cc

/*
 * Copyright (C) 2019 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "perfetto/trace_processor/read_trace.h"
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
#include <utility>
#include <vector>

#include "perfetto/base/logging.h"
#include "perfetto/base/status.h"
#include "perfetto/ext/base/status_macros.h"
#include "perfetto/protozero/field.h"
#include "perfetto/protozero/proto_utils.h"
#include "perfetto/trace_processor/trace_blob.h"
#include "perfetto/trace_processor/trace_blob_view.h"
#include "perfetto/trace_processor/trace_processor.h"
#include "src/trace_processor/importers/archive/decompressing_trace_reader.h"
#include "src/trace_processor/importers/common/chunked_trace_reader.h"
#include "src/trace_processor/importers/proto/proto_trace_tokenizer.h"
#include "src/trace_processor/read_trace_internal.h"
#include "src/trace_processor/util/decompressor.h"
#include "src/trace_processor/util/trace_type.h"

#include "protos/perfetto/trace/trace.pbzero.h"
#include "protos/perfetto/trace/trace_packet.pbzero.h"

namespace perfetto::trace_processor {
namespace {

class SerializingProtoTraceReader : public ChunkedTraceReader {
 public:
  explicit SerializingProtoTraceReader(std::vector<uint8_t>* output)
      : output_(output) {}

  base::Status Parse(TraceBlobView blob) override {
    return tokenizer_.Tokenize(std::move(blob), [this](TraceBlobView packet) {
      return ParsePacket(std::move(packet));
    });
  }

  base::Status OnPushDataToSorter() override { return base::OkStatus(); }
  void OnEventsFullyExtracted() override {}

 private:
  base::Status ParsePacket(TraceBlobView packet) {
    protos::pbzero::TracePacket::Decoder decoder(packet.data(),
                                                 packet.length());
    if (decoder.has_compressed_packets() ||
        decoder.has_zstd_compressed_packets()) {
      util::CompressionType codec;
      protozero::ConstBytes field;
      if (decoder.has_compressed_packets()) {
        codec = util::CompressionType::kGzip;
        field = decoder.compressed_packets();
      } else {
        codec = util::CompressionType::kZstd;
        field = decoder.zstd_compressed_packets();
      }
      return tokenizer_.TokenizeCompressedPackets(
          codec, packet.slice(field.data, field.size),
          [this](TraceBlobView inner) {
            return ParsePacket(std::move(inner));
          });
    }

    uint8_t buffer[protozero::proto_utils::kMaxSimpleFieldEncodedSize];
    uint8_t* pos = buffer;
    pos = protozero::proto_utils::WriteVarInt(kTracePacketTag, pos);
    pos = protozero::proto_utils::WriteVarInt(packet.length(), pos);
    output_->insert(output_->end(), buffer, pos);
    output_->insert(output_->end(), packet.data(),
                    packet.data() + packet.length());
    return base::OkStatus();
  }

  static constexpr uint8_t kTracePacketTag =
      protozero::proto_utils::MakeTagLengthDelimited(
          protos::pbzero::Trace::kPacketFieldNumber);

  ProtoTraceTokenizer tokenizer_;
  std::vector<uint8_t>* output_;
};

}  // namespace

base::Status ReadTrace(
    TraceProcessor* tp,
    const char* filename,
    const std::function<void(uint64_t parsed_size)>& progress_callback,
    bool call_notify_end_of_file,
    const ReadTraceArgs& args) {
  RETURN_IF_ERROR(ReadTraceUnfinalized(tp, filename, progress_callback, args));
  if (call_notify_end_of_file) {
    return tp->NotifyEndOfFile();
  }
  return base::OkStatus();
}

base::Status DecompressTrace(const uint8_t* data,
                             size_t size,
                             std::vector<uint8_t>* output) {
  CompressedTraceType type = SniffCompressedTraceType(data, size);
  if (type == CompressedTraceType::kOther) {
    return base::ErrStatus(
        "Only GZIP, ZSTD and proto trace types are supported by "
        "DecompressTrace");
  }

  if (type == CompressedTraceType::kGzip ||
      type == CompressedTraceType::kZstd) {
    util::CompressionType codec = type == CompressedTraceType::kZstd
                                      ? util::CompressionType::kZstd
                                      : util::CompressionType::kGzip;
    std::unique_ptr<ChunkedTraceReader> reader(
        new SerializingProtoTraceReader(output));
    DecompressingTraceReader parser(std::move(reader), codec);
    RETURN_IF_ERROR(parser.ParseUnowned(data, size));
    RETURN_IF_ERROR(parser.OnPushDataToSorter());
    parser.OnEventsFullyExtracted();
    return base::OkStatus();
  }

  PERFETTO_CHECK(type == CompressedTraceType::kProto);

  // Plain proto: expand each compressed_packets / zstd_compressed_packets
  // bundle into `output`, and pass other packets through unchanged.
  protos::pbzero::Trace::Decoder decoder(data, size);
  if (size > 0 && !decoder.packet()) {
    return base::ErrStatus("Trace does not contain valid packets");
  }

  for (auto it = decoder.packet(); it; ++it) {
    protos::pbzero::TracePacket::Decoder packet(*it);

    util::CompressionType codec;
    protozero::ConstBytes bytes;
    if (packet.has_compressed_packets()) {
      codec = util::CompressionType::kGzip;
      bytes = packet.compressed_packets();
    } else if (packet.has_zstd_compressed_packets()) {
      codec = util::CompressionType::kZstd;
      bytes = packet.zstd_compressed_packets();
    } else {
      it->SerializeAndAppendTo(output);
      continue;
    }

    if (!util::IsCompressionSupported(codec)) {
      auto info = util::GetCompressionCodecInfo(codec);
      return base::ErrStatus(
          "Cannot decompress compressed_packets: %s is not enabled in this "
          "build (rebuild with %s=true)",
          info.name, info.gn_arg);
    }

    std::optional<util::DecompressedBuffer> buf =
        util::DecompressToBuffer(codec, bytes.data, bytes.size);
    if (!buf) {
      return base::ErrStatus(
          "Failed to decompress compressed_packets (ERR:tp-corrupt)");
    }
    output->insert(output->end(), buf->data.get(), buf->data.get() + buf->size);
  }
  return base::OkStatus();
}

}  // namespace perfetto::trace_processor