aos/events/logging/logger_test.cc

#include "aos/events/logging/logger.h"

#include "aos/events/event_loop.h"
#include "aos/events/ping_lib.h"
#include "aos/events/pong_lib.h"
#include "aos/events/simulated_event_loop.h"
#include "glog/logging.h"
#include "gtest/gtest.h"

namespace aos {
namespace logger {
namespace testing {

namespace chrono = std::chrono;

class LoggerTest : public ::testing::Test {
 public:
  LoggerTest()
      : config_(
            aos::configuration::ReadConfig("aos/events/pingpong_config.json")),
        event_loop_factory_(&config_.message()),
        ping_event_loop_(event_loop_factory_.MakeEventLoop("ping")),
        ping_(ping_event_loop_.get()),
        pong_event_loop_(event_loop_factory_.MakeEventLoop("pong")),
        pong_(pong_event_loop_.get()) {}

  // Config and factory.
  aos::FlatbufferDetachedBuffer<aos::Configuration> config_;
  SimulatedEventLoopFactory event_loop_factory_;

  // Event loop and app for Ping
  std::unique_ptr<EventLoop> ping_event_loop_;
  Ping ping_;

  // Event loop and app for Pong
  std::unique_ptr<EventLoop> pong_event_loop_;
  Pong pong_;
};

// Tests that we can startup at all.  This confirms that the channels are all in
// the config.
TEST_F(LoggerTest, Starts) {
  const ::std::string tmpdir(getenv("TEST_TMPDIR"));
  const ::std::string logfile = tmpdir + "/logfile.bfbs";
  // Remove it.
  unlink(logfile.c_str());

  LOG(INFO) << "Logging data to " << logfile;

  {
    DetachedBufferWriter writer(logfile);
    std::unique_ptr<EventLoop> logger_event_loop =
        event_loop_factory_.MakeEventLoop("logger");

    event_loop_factory_.RunFor(chrono::milliseconds(95));

    Logger logger(&writer, logger_event_loop.get(),
                  std::chrono::milliseconds(100));
    event_loop_factory_.RunFor(chrono::milliseconds(20000));
  }

  // Even though it doesn't make any difference here, exercise the logic for
  // passing in a separate config.
  LogReader reader(logfile, &config_.message());

  // Confirm that we can remap logged channels to point to new buses.
  reader.RemapLoggedChannel<aos::examples::Ping>("/test", "/original");

  // This sends out the fetched messages and advances time to the start of the
  // log file.
  reader.Register();

  EXPECT_EQ(reader.node(), nullptr);

  std::unique_ptr<EventLoop> test_event_loop =
      reader.event_loop_factory()->MakeEventLoop("log_reader");

  int ping_count = 10;
  int pong_count = 10;

  // Confirm that the ping value matches in the remapped channel location.
  test_event_loop->MakeWatcher("/original/test",
                               [&ping_count](const examples::Ping &ping) {
                                 EXPECT_EQ(ping.value(), ping_count + 1);
                                 ++ping_count;
                               });
  // Confirm that the ping and pong counts both match, and the value also
  // matches.
  test_event_loop->MakeWatcher(
      "/test", [&pong_count, &ping_count](const examples::Pong &pong) {
        EXPECT_EQ(pong.value(), pong_count + 1);
        ++pong_count;
        EXPECT_EQ(ping_count, pong_count);
      });

  reader.event_loop_factory()->RunFor(std::chrono::seconds(100));
  EXPECT_EQ(ping_count, 2010);
}

// Tests that we can read and write rotated log files.
TEST_F(LoggerTest, RotatedLogFile) {
  const ::std::string tmpdir(getenv("TEST_TMPDIR"));
  const ::std::string logfile0 = tmpdir + "/logfile0.bfbs";
  const ::std::string logfile1 = tmpdir + "/logfile1.bfbs";
  // Remove it.
  unlink(logfile0.c_str());
  unlink(logfile1.c_str());

  LOG(INFO) << "Logging data to " << logfile0 << " and " << logfile1;

  {
    DetachedBufferWriter writer0(logfile0);
    DetachedBufferWriter writer1(logfile1);
    std::unique_ptr<EventLoop> logger_event_loop =
        event_loop_factory_.MakeEventLoop("logger");

    event_loop_factory_.RunFor(chrono::milliseconds(95));

    Logger logger(&writer0, logger_event_loop.get(),
                  std::chrono::milliseconds(100));
    event_loop_factory_.RunFor(chrono::milliseconds(10000));
    logger.Rotate(&writer1);
    event_loop_factory_.RunFor(chrono::milliseconds(10000));
  }

  // Even though it doesn't make any difference here, exercise the logic for
  // passing in a separate config.
  LogReader reader(std::vector<std::string>{logfile0, logfile1},
                   &config_.message());

  // Confirm that we can remap logged channels to point to new buses.
  reader.RemapLoggedChannel<aos::examples::Ping>("/test", "/original");

  // This sends out the fetched messages and advances time to the start of the
  // log file.
  reader.Register();

  EXPECT_EQ(reader.node(), nullptr);

  std::unique_ptr<EventLoop> test_event_loop =
      reader.event_loop_factory()->MakeEventLoop("log_reader");

  int ping_count = 10;
  int pong_count = 10;

  // Confirm that the ping value matches in the remapped channel location.
  test_event_loop->MakeWatcher("/original/test",
                               [&ping_count](const examples::Ping &ping) {
                                 EXPECT_EQ(ping.value(), ping_count + 1);
                                 ++ping_count;
                               });
  // Confirm that the ping and pong counts both match, and the value also
  // matches.
  test_event_loop->MakeWatcher(
      "/test", [&pong_count, &ping_count](const examples::Pong &pong) {
        EXPECT_EQ(pong.value(), pong_count + 1);
        ++pong_count;
        EXPECT_EQ(ping_count, pong_count);
      });

  reader.event_loop_factory()->RunFor(std::chrono::seconds(100));
  EXPECT_EQ(ping_count, 2010);
}

// Tests that a large number of messages per second doesn't overwhelm writev.
TEST_F(LoggerTest, ManyMessages) {
  const ::std::string tmpdir(getenv("TEST_TMPDIR"));
  const ::std::string logfile = tmpdir + "/logfile.bfbs";
  // Remove the log file.
  unlink(logfile.c_str());

  LOG(INFO) << "Logging data to " << logfile;

  {
    DetachedBufferWriter writer(logfile);
    std::unique_ptr<EventLoop> logger_event_loop =
        event_loop_factory_.MakeEventLoop("logger");

    std::unique_ptr<EventLoop> ping_spammer_event_loop =
        event_loop_factory_.MakeEventLoop("ping_spammer");
    aos::Sender<examples::Ping> ping_sender =
        ping_spammer_event_loop->MakeSender<examples::Ping>("/test");

    aos::TimerHandler *timer_handler =
        ping_spammer_event_loop->AddTimer([&ping_sender]() {
          aos::Sender<examples::Ping>::Builder builder =
              ping_sender.MakeBuilder();
          examples::Ping::Builder ping_builder =
              builder.MakeBuilder<examples::Ping>();
          CHECK(builder.Send(ping_builder.Finish()));
        });

    // 100 ms / 0.05 ms -> 2000 messages.  Should be enough to crash it.
    ping_spammer_event_loop->OnRun([&ping_spammer_event_loop, timer_handler]() {
      timer_handler->Setup(ping_spammer_event_loop->monotonic_now(),
                           chrono::microseconds(50));
    });

    Logger logger(&writer, logger_event_loop.get(),
                  std::chrono::milliseconds(100));

    event_loop_factory_.RunFor(chrono::milliseconds(1000));
  }
}

class MultinodeLoggerTest : public ::testing::Test {
 public:
  MultinodeLoggerTest()
      : config_(aos::configuration::ReadConfig(
            "aos/events/logging/multinode_pingpong_config.json")),
        event_loop_factory_(&config_.message(), "pi1"),
        ping_event_loop_(event_loop_factory_.MakeEventLoop("ping")),
        ping_(ping_event_loop_.get()) {}

  // Config and factory.
  aos::FlatbufferDetachedBuffer<aos::Configuration> config_;
  SimulatedEventLoopFactory event_loop_factory_;

  // Event loop and app for Ping
  std::unique_ptr<EventLoop> ping_event_loop_;
  Ping ping_;
};

// Tests that we can startup at all in a multinode configuration.
TEST_F(MultinodeLoggerTest, MultiNode) {
  constexpr chrono::seconds kTimeOffset = chrono::seconds(10000);
  constexpr uint32_t kQueueIndexOffset = 1024;
  const ::std::string tmpdir(getenv("TEST_TMPDIR"));
  const ::std::string logfile = tmpdir + "/multi_logfile.bfbs";
  // Remove it.
  unlink(logfile.c_str());

  LOG(INFO) << "Logging data to " << logfile;

  {
    std::unique_ptr<EventLoop> pong_event_loop =
        event_loop_factory_.MakeEventLoop("pong");

    std::unique_ptr<aos::RawSender> pong_sender(
        pong_event_loop->MakeRawSender(aos::configuration::GetChannel(
            pong_event_loop->configuration(), "/test", "aos.examples.Pong",
            pong_event_loop->name(), pong_event_loop->node())));

    // Ok, let's fake a remote node.  We use the fancy raw sender Send
    // method that message_gateway will use to do that.
    int pong_count = 0;
    pong_event_loop->MakeWatcher(
        "/test", [&pong_event_loop, &pong_count, &pong_sender,
                  kTimeOffset](const examples::Ping &ping) {
          flatbuffers::FlatBufferBuilder fbb;
          examples::Pong::Builder pong_builder(fbb);
          pong_builder.add_value(ping.value());
          pong_builder.add_initial_send_time(ping.send_time());
          fbb.Finish(pong_builder.Finish());

          pong_sender->Send(fbb.GetBufferPointer(), fbb.GetSize(),
                            pong_event_loop->monotonic_now() + kTimeOffset,
                            pong_event_loop->realtime_now() + kTimeOffset,
                            kQueueIndexOffset + pong_count);
          ++pong_count;
        });

    DetachedBufferWriter writer(logfile);
    std::unique_ptr<EventLoop> logger_event_loop =
        event_loop_factory_.MakeEventLoop("logger");

    event_loop_factory_.RunFor(chrono::milliseconds(95));

    Logger logger(&writer, logger_event_loop.get(),
                  std::chrono::milliseconds(100));
    event_loop_factory_.RunFor(chrono::milliseconds(20000));
  }

  LogReader reader(logfile);

  // TODO(austin): Also replay as pi2 or pi3 and make sure we see the pong
  // messages.  This won't work today yet until the log reading code gets
  // significantly better.
  SimulatedEventLoopFactory log_reader_factory(reader.logged_configuration(), reader.node());
  log_reader_factory.set_send_delay(chrono::microseconds(0));

  // This sends out the fetched messages and advances time to the start of the
  // log file.
  reader.Register(&log_reader_factory);

  ASSERT_NE(reader.node(), nullptr);
  EXPECT_EQ(reader.node()->name()->string_view(), "pi1");

  reader.event_loop_factory()->set_send_delay(chrono::microseconds(0));

  std::unique_ptr<EventLoop> test_event_loop =
      log_reader_factory.MakeEventLoop("test");

  int ping_count = 10;
  int pong_count = 10;

  // Confirm that the ping value matches.
  test_event_loop->MakeWatcher("/test",
                               [&ping_count](const examples::Ping &ping) {
                                 EXPECT_EQ(ping.value(), ping_count + 1);
                                 ++ping_count;
                               });
  // Confirm that the ping and pong counts both match, and the value also
  // matches.
  test_event_loop->MakeWatcher(
      "/test", [&test_event_loop, &ping_count, &pong_count,
                kTimeOffset](const examples::Pong &pong) {
        EXPECT_EQ(test_event_loop->context().remote_queue_index,
                  pong_count + kQueueIndexOffset);
        EXPECT_EQ(test_event_loop->context().monotonic_remote_time,
                  test_event_loop->monotonic_now() + kTimeOffset);
        EXPECT_EQ(test_event_loop->context().realtime_remote_time,
                  test_event_loop->realtime_now() + kTimeOffset);

        EXPECT_EQ(pong.value(), pong_count + 1);
        ++pong_count;
        EXPECT_EQ(ping_count, pong_count);
      });

  log_reader_factory.RunFor(std::chrono::seconds(100));
  EXPECT_EQ(ping_count, 2010);
  EXPECT_EQ(pong_count, 2010);

  reader.Deregister();
}

}  // namespace testing
}  // namespace logger
}  // namespace aos