aos/events/logging/log_cat.cc

#include <algorithm>
#include <iostream>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <vector>

#include "absl/strings/escaping.h"
#include "aos/configuration.h"
#include "aos/events/logging/log_reader.h"
#include "aos/events/simulated_event_loop.h"
#include "aos/init.h"
#include "aos/json_to_flatbuffer.h"
#include "gflags/gflags.h"

DEFINE_string(
    name, "",
    "Name to match for printing out channels. Empty means no name filter.");
DEFINE_string(type, "",
              "Channel type to match for printing out channels. Empty means no "
              "type filter.");
DEFINE_bool(fetch, false,
            "If true, also print out the messages from before the start of the "
            "log file");
DEFINE_bool(raw, false,
            "If true, just print the data out unsorted and unparsed");
DEFINE_string(raw_header, "",
              "If set, the file to read the header from in raw mode");
DEFINE_bool(format_raw, true,
            "If true and --raw is specified, print out raw data, but use the "
            "schema to format the data.");
DEFINE_int32(max_vector_size, 100,
             "If positive, vectors longer than this will not be printed");
DEFINE_bool(pretty, false,
            "If true, pretty print the messages on multiple lines");
DEFINE_bool(print, true,
            "If true, actually print the messages.  If false, discard them, "
            "confirming they can be parsed.");
DEFINE_bool(print_parts_only, false,
            "If true, only print out the results of logfile sorting.");

// Print the flatbuffer out to stdout, both to remove the unnecessary cruft from
// glog and to allow the user to readily redirect just the logged output
// independent of any debugging information on stderr.
void PrintMessage(const std::string_view node_name, const aos::Channel *channel,
                  const aos::Context &context,
                  aos::FastStringBuilder *builder) {
  builder->Reset();
  CHECK(flatbuffers::Verify(*channel->schema(),
                            *channel->schema()->root_table(),
                            static_cast<const uint8_t *>(context.data),
                            static_cast<size_t>(context.size)))
      << ": Corrupted flatbuffer on " << channel->name()->c_str() << " "
      << channel->type()->c_str();

  aos::FlatbufferToJson(
      builder, channel->schema(), static_cast<const uint8_t *>(context.data),
      {FLAGS_pretty, static_cast<size_t>(FLAGS_max_vector_size)});

  if (context.monotonic_remote_time != context.monotonic_event_time) {
    std::cout << node_name << context.realtime_event_time << " ("
              << context.monotonic_event_time << ") sent "
              << context.realtime_remote_time << " ("
              << context.monotonic_remote_time << ") "
              << channel->name()->c_str() << ' ' << channel->type()->c_str()
              << ": " << *builder << std::endl;
  } else {
    std::cout << node_name << context.realtime_event_time << " ("
              << context.monotonic_event_time << ") "
              << channel->name()->c_str() << ' ' << channel->type()->c_str()
              << ": " << *builder << std::endl;
  }
}

int main(int argc, char **argv) {
  gflags::SetUsageMessage(
      "Usage:\n"
      "  log_cat [args] logfile1 logfile2 ...\n"
      "\n"
      "This program provides a basic interface to dump data from a logfile to "
      "stdout. Given a logfile, channel name filter, and type filter, it will "
      "print all the messages in the logfile matching the filters. The message "
      "filters work by taking the values of --name and --type and printing any "
      "channel whose name contains --name as a substr and whose type contains "
      "--type as a substr. Not specifying --name or --type leaves them free. "
      "Calling this program without --name or --type specified prints out all "
      "the logged data.");
  aos::InitGoogle(&argc, &argv);

  if (FLAGS_raw) {
    if (argc != 2) {
      LOG(FATAL) << "Expected 1 logfile as an argument.";
    }
    aos::logger::SpanReader reader(argv[1]);
    absl::Span<const uint8_t> raw_log_file_header_span = reader.ReadMessage();

    if (raw_log_file_header_span == absl::Span<const uint8_t>()) {
      LOG(WARNING) << "Empty log file on " << reader.filename();
      return 0;
    }

    // Now, reproduce the log file header deduplication logic inline so we can
    // print out all the headers we find.
    aos::SizePrefixedFlatbufferVector<aos::logger::LogFileHeader>
        log_file_header(raw_log_file_header_span);
    if (!log_file_header.Verify()) {
      LOG(ERROR) << "Header corrupted on " << reader.filename();
      return 1;
    }
    while (true) {
      absl::Span<const uint8_t> maybe_header_data = reader.PeekMessage();
      if (maybe_header_data == absl::Span<const uint8_t>()) {
        break;
      }

      aos::SizePrefixedFlatbufferSpan<aos::logger::LogFileHeader> maybe_header(
          maybe_header_data);
      if (maybe_header.Verify()) {
        std::cout << aos::FlatbufferToJson(
                         log_file_header,
                         {.multi_line = FLAGS_pretty,
                          .max_vector_size =
                              static_cast<size_t>(FLAGS_max_vector_size)})
                  << std::endl;
        LOG(WARNING) << "Found duplicate LogFileHeader in "
                     << reader.filename();
        log_file_header =
            aos::SizePrefixedFlatbufferVector<aos::logger::LogFileHeader>(
                maybe_header_data);

        reader.ConsumeMessage();
      } else {
        break;
      }
    }

    // And now use the final sha256 to match the raw_header.
    std::optional<aos::logger::MessageReader> raw_header_reader;
    const aos::logger::LogFileHeader *full_header = &log_file_header.message();
    if (!FLAGS_raw_header.empty()) {
      raw_header_reader.emplace(FLAGS_raw_header);
      std::cout << aos::FlatbufferToJson(
                       full_header, {.multi_line = FLAGS_pretty,
                                     .max_vector_size = static_cast<size_t>(
                                         FLAGS_max_vector_size)})
                << std::endl;
      CHECK_EQ(
          full_header->configuration_sha256()->string_view(),
          aos::logger::Sha256(raw_header_reader->raw_log_file_header().span()));
      full_header = raw_header_reader->log_file_header();
    }
    std::cout << aos::FlatbufferToJson(full_header,
                                       {.multi_line = FLAGS_pretty,
                                        .max_vector_size = static_cast<size_t>(
                                            FLAGS_max_vector_size)})
              << std::endl;

    while (true) {
      const aos::SizePrefixedFlatbufferSpan<aos::logger::MessageHeader> message(
          reader.ReadMessage());
      if (message.span() == absl::Span<const uint8_t>()) {
        break;
      }
      const auto *const channels = full_header->configuration()->channels();
      const size_t channel_index = message.message().channel_index();
      CHECK_LT(channel_index, channels->size());
      const aos::Channel *const channel = channels->Get(channel_index);

      CHECK(message.Verify()) << absl::BytesToHexString(std::string_view(
          reinterpret_cast<const char *>(message.span().data()),
          message.span().size()));

      if (message.message().data() != nullptr) {
        CHECK(channel->has_schema());

        CHECK(flatbuffers::Verify(
            *channel->schema(), *channel->schema()->root_table(),
            message.message().data()->data(), message.message().data()->size()))
            << ": Corrupted flatbuffer on " << channel->name()->c_str() << " "
            << channel->type()->c_str();
      }

      if (FLAGS_format_raw && message.message().data() != nullptr) {
        std::cout << aos::configuration::StrippedChannelToString(channel) << " "
                  << aos::FlatbufferToJson(message, {.multi_line = FLAGS_pretty,
                                                     .max_vector_size = 4})
                  << ": "
                  << aos::FlatbufferToJson(
                         channel->schema(), message.message().data()->data(),
                         {FLAGS_pretty,
                          static_cast<size_t>(FLAGS_max_vector_size)})
                  << std::endl;
      } else {
        std::cout << aos::configuration::StrippedChannelToString(channel) << " "
                  << aos::FlatbufferToJson(
                         message, {FLAGS_pretty,
                                   static_cast<size_t>(FLAGS_max_vector_size)})
                  << std::endl;
      }
    }
    return 0;
  }

  if (argc < 2) {
    LOG(FATAL) << "Expected at least 1 logfile as an argument.";
  }

  const std::vector<std::string> unsorted_logfiles =
      aos::logger::FindLogs(argc, argv);

  const std::vector<aos::logger::LogFile> logfiles =
      aos::logger::SortParts(unsorted_logfiles);

  for (auto &it : logfiles) {
    VLOG(1) << it;
    if (FLAGS_print_parts_only) {
      std::cout << it << std::endl;
    }
  }
  if (FLAGS_print_parts_only) {
    return 0;
  }

  aos::logger::LogReader reader(logfiles);

  aos::FastStringBuilder builder;

  aos::SimulatedEventLoopFactory event_loop_factory(reader.configuration());
  reader.Register(&event_loop_factory);

  std::vector<std::unique_ptr<aos::EventLoop>> printer_event_loops;

  bool found_channel = false;

  for (const aos::Node *node :
       aos::configuration::GetNodes(event_loop_factory.configuration())) {
    std::unique_ptr<aos::EventLoop> printer_event_loop =
        event_loop_factory.MakeEventLoop("printer", node);
    printer_event_loop->SkipTimingReport();
    printer_event_loop->SkipAosLog();

    struct MessageInfo {
      std::string node_name;
      std::unique_ptr<aos::RawFetcher> fetcher;
    };
    std::vector<MessageInfo> messages_before_start;

    const flatbuffers::Vector<flatbuffers::Offset<aos::Channel>> *channels =
        printer_event_loop->configuration()->channels();

    for (flatbuffers::uoffset_t i = 0; i < channels->size(); i++) {
      const aos::Channel *channel = channels->Get(i);
      const flatbuffers::string_view name = channel->name()->string_view();
      const flatbuffers::string_view type = channel->type()->string_view();
      if (name.find(FLAGS_name) != std::string::npos &&
          type.find(FLAGS_type) != std::string::npos) {
        if (!aos::configuration::ChannelIsReadableOnNode(
                channel, printer_event_loop->node())) {
          continue;
        }
        VLOG(1) << "Listening on " << name << " " << type;

        std::string node_name =
            node == nullptr ? ""
                            : std::string(node->name()->string_view()) + " ";

        CHECK_NOTNULL(channel->schema());

        // Fetch the last message on this channel from before the log start
        // time.
        if (FLAGS_fetch) {
          std::unique_ptr<aos::RawFetcher> fetcher =
              printer_event_loop->MakeRawFetcher(channel);
          if (fetcher->Fetch()) {
            MessageInfo message{.node_name = node_name,
                                .fetcher = std::move(fetcher)};
            // Insert it sorted into the vector so we can print in time order
            // instead of channel order at the start.
            auto it = std::lower_bound(
                messages_before_start.begin(), messages_before_start.end(),
                message, [](const MessageInfo &lhs, const MessageInfo &rhs) {
                  if (lhs.fetcher->context().monotonic_event_time <
                      rhs.fetcher->context().monotonic_event_time) {
                    return true;
                  }
                  if (lhs.fetcher->context().monotonic_event_time >
                      rhs.fetcher->context().monotonic_event_time) {
                    return false;
                  }
                  return lhs.fetcher->channel() < rhs.fetcher->channel();
                });
            messages_before_start.insert(it, std::move(message));
          }
        }

        printer_event_loop->MakeRawWatcher(
            channel, [channel, node_name, &builder](const aos::Context &context,
                                                    const void * /*message*/) {
              if (FLAGS_print) {
                PrintMessage(node_name, channel, context, &builder);
              }
            });
        found_channel = true;
      }
    }

    // Print the messages from before the log start time.
    // TODO(austin): Sort between nodes too when it becomes annoying enough.
    for (const MessageInfo &message : messages_before_start) {
      if (FLAGS_print) {
        PrintMessage(message.node_name, message.fetcher->channel(),
                     message.fetcher->context(), &builder);
      }
    }
    printer_event_loops.emplace_back(std::move(printer_event_loop));

    std::cout << std::endl;
    std::cout << (node != nullptr ? (node->name()->str() + " ") : "")
              << "Log starting at " << reader.realtime_start_time(node) << " ("
              << reader.monotonic_start_time(node) << ")";
    std::cout << std::endl << std::endl;
  }

  if (!found_channel) {
    LOG(FATAL) << "Could not find any channels";
  }

  if (FLAGS_fetch) {
    // New line to separate fetched messages from non-fetched messages.
    std::cout << std::endl;
  }

  event_loop_factory.Run();

  reader.Deregister();

  return 0;
}