src/logger/event_loggers_test.cc

// Copyright 2018 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "src/logger/event_loggers.h"

#include <chrono>
#include <memory>
#include <string>

#include <google/protobuf/repeated_field.h>
#include <google/protobuf/text_format.h>
#include <google/protobuf/util/message_differencer.h>
#include <gtest/gtest.h>

#include "src/lib/util/clock.h"
#include "src/lib/util/datetime_util.h"
#include "src/lib/util/encrypted_message_util.h"
#include "src/lib/util/testing/test_with_files.h"
#include "src/local_aggregation/local_aggregate_storage/immediate_local_aggregate_storage.h"
#include "src/local_aggregation/local_aggregate_storage/local_aggregate_storage.h"
#include "src/local_aggregation/local_aggregation.h"
#include "src/local_aggregation/local_aggregation.pb.h"
#include "src/logger/encoder.h"
#include "src/logger/logger_test_utils.h"
#include "src/logger/project_context.h"
#include "src/logger/test_registries/cobalt1.1_test_registry.cb.h"
#include "src/pb/metadata_builder.h"
#include "src/pb/observation.pb.h"
#include "src/public/lib/status.h"
#include "src/system_data/client_secret.h"
#include "src/system_data/fake_system_data.h"

namespace cobalt::logger {

using internal::EventLogger;
using local_aggregation::LocalAggregation;
using system_data::ClientSecret;
using testing::ExpectedAggregationParams;
using testing::ExpectedPerDeviceNumericObservations;
using testing::ExpectedReportParticipationObservations;
using testing::ExpectedUniqueActivesObservations;
using testing::FakeObservationStore;
using testing::GetTestProject;
using testing::GetTestProjectContextFactory;
using testing::TestUpdateRecipient;
using util::EncryptedMessageMaker;
using util::IncrementingSystemClock;

namespace {
// Number of seconds in a day
constexpr int kDay = 60 * 60 * 24;
// Number of seconds in an ideal year
constexpr int kYear = kDay * 365;

}  // namespace

// needed for Friend class status
namespace internal {

class EventLoggersLocalAggregationTest : public util::testing::TestWithFiles {
 public:
  void SetUp() override {
    MakeTestFolder();
    observation_store_ = std::make_unique<FakeObservationStore>();
    update_recipient_ = std::make_unique<TestUpdateRecipient>();
    observation_encrypter_ = EncryptedMessageMaker::MakeUnencrypted().Unwrap();
    observation_writer_ = std::make_unique<ObservationWriter>(
        *observation_store_, update_recipient_.get(), observation_encrypter_.get());
    metadata_builder_ =
        std::make_unique<MetadataBuilder>(system_data_, system_data_cache_path(), fs());
    encoder_ = std::make_unique<Encoder>(ClientSecret::GenerateNewSecret(), *metadata_builder_);
    project_context_factory_ =
        GetTestProjectContextFactory(testing::cobalt_new::kCobaltRegistryBase64);
    project_context_ = GetTestProject(testing::cobalt_new::kCobaltRegistryBase64);
    // Create a mock clock which does not increment by default when called.
    // Set the time to 1 year after the start of Unix time so that the start
    // date of any aggregation window falls after the start of time.
    mock_clock_ = std::make_unique<IncrementingSystemClock>(std::chrono::system_clock::duration(0));
    start_time_ = std::chrono::system_clock::time_point(std::chrono::seconds(kYear));
    mock_clock_->set_time(start_time_);
    // Set the test local time converter to switch time zones 1 hour after `start_time_`.
    // This should only affect metrics with the OTHER_TIME_ZONE TimeZonePolicy.
    civil_time_converter_ = std::make_unique<util::FakeCivilTimeConverter>(
        /*start_utc_offset=*/0, /*start_isdst=*/false, /*end_utc_offset=*/1, /*end_isdst=*/true,
        /*threshold=*/start_time_ + util::kOneHour);

    CobaltConfig cfg = {.client_secret = system_data::ClientSecret::GenerateNewSecret()};

    cfg.local_aggregation_backfill_days = 0;
    cfg.local_aggregate_store_dir = local_aggregation_store_path();

    local_aggregation_ = std::make_unique<LocalAggregation>(
        cfg, *project_context_factory_, system_data_, *metadata_builder_, fs(),
        *observation_writer_, *civil_time_converter_);
  }

  local_aggregation::ImmediateLocalAggregateStorage GetLocalAggregateStorage() {
    return local_aggregation::ImmediateLocalAggregateStorage(
        local_aggregation_store_path(), fs(), *project_context_factory_, *metadata_builder_, 0);
  }

  local_aggregation::ReportAggregate GetReportAggregate(
      uint32_t metric_id, MetricDefinition::MetricType metric_type, uint32_t report_id,
      local_aggregation::LocalAggregateStorage* store) {
    std::unique_ptr<EventRecord> event_record =
        GetEventRecordWithMetricType(metric_id, metric_type);
    local_aggregation::LocalAggregateStorage::MetricAggregateRef aggregate =
        store->GetMetricAggregate(event_record->MetricIdentifier()).value();
    return aggregate.aggregate()->by_report_id().at(report_id);
  }

  std::unique_ptr<EventLogger> GetEventLoggerForMetricType(
      MetricDefinition::MetricType metric_type) {
    auto logger = EventLogger::Create(metric_type, *encoder_, *local_aggregation_,
                                      *observation_writer_, system_data_, *civil_time_converter_);
    return logger;
  }

  std::unique_ptr<EventRecord> GetEventRecordWithMetricType(
      uint32_t metric_id, MetricDefinition::MetricType metric_type) {
    auto event_record = EventRecord::MakeEventRecord(project_context_, metric_id).value();

    switch (metric_type) {
      case (MetricDefinition::OCCURRENCE): {
        OccurrenceEvent* event = event_record->event()->mutable_occurrence_event();
        event->set_count(1);
        break;
      }
      case (MetricDefinition::INTEGER): {
        IntegerEvent* event = event_record->event()->mutable_integer_event();
        event->set_value(10);
        break;
      }
      case (MetricDefinition::INTEGER_HISTOGRAM): {
        IntegerHistogramEvent* event = event_record->event()->mutable_integer_histogram_event();
        event->mutable_buckets()->Swap(
            testing::NewHistogram({0, 1, 2, 3}, {100, 101, 102, 103}).get());
        break;
      }
      case (MetricDefinition::STRING): {
        StringEvent* event = event_record->event()->mutable_string_event();
        event->set_string_value("component4");
        break;
      }
      default: {
      }
    }

    return event_record;
  }

  static ReportDefinition ReportDefinitionWithReportType(uint32_t report_id,
                                                         ReportDefinition::ReportType report_type) {
    ReportDefinition report_definition;
    report_definition.set_report_type(report_type);
    report_definition.set_id(report_id);
    return report_definition;
  }

 protected:
  std::unique_ptr<ObservationWriter> observation_writer_;
  std::unique_ptr<FakeObservationStore> observation_store_;
  std::unique_ptr<TestUpdateRecipient> update_recipient_;
  std::unique_ptr<MetadataBuilder> metadata_builder_;
  std::unique_ptr<Encoder> encoder_;
  std::unique_ptr<EncryptedMessageMaker> observation_encrypter_;
  system_data::FakeSystemData system_data_;
  std::unique_ptr<IncrementingSystemClock> mock_clock_;
  std::chrono::system_clock::time_point start_time_;
  std::unique_ptr<util::FakeCivilTimeConverter> civil_time_converter_;
  std::unique_ptr<ProjectContextFactory> project_context_factory_;
  std::shared_ptr<ProjectContext> project_context_;
  std::unique_ptr<LocalAggregation> local_aggregation_;
};

TEST_F(EventLoggersLocalAggregationTest, OccurrenceUniqueDeviceHistogramsValidation) {
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kSettingsChangedMetricId,
                                                   MetricDefinition::OCCURRENCE);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::OCCURRENCE);
  EXPECT_EQ(StatusCode::OK,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kSettingsChangedMetricId,
                                          MetricDefinition::OCCURRENCE, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, OccurrenceUniqueDeviceHistogramsValidationWrongType) {
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kSettingsChangedMetricId,
                                                   MetricDefinition::INTEGER);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::OCCURRENCE);
  EXPECT_EQ(StatusCode::INVALID_ARGUMENT,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kSettingsChangedMetricId,
                                          MetricDefinition::OCCURRENCE, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, OccurrenceUniqueDeviceHistograms) {
  ReportDefinition report = ReportDefinitionWithReportType(
      testing::cobalt_new::kSettingsChangedSettingsChangedUniqueDeviceHistogramsReportId,
      ReportDefinition::UNIQUE_DEVICE_HISTOGRAMS);
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kSettingsChangedMetricId,
                                                   MetricDefinition::OCCURRENCE);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::OCCURRENCE);
  EXPECT_EQ(
      StatusCode::OK,
      event_logger->Log(std::move(event_record), std::chrono::system_clock::now()).error_code());

  event_record = GetEventRecordWithMetricType(testing::cobalt_new::kSettingsChangedMetricId,
                                              MetricDefinition::OCCURRENCE);
  local_aggregation::ImmediateLocalAggregateStorage store = GetLocalAggregateStorage();
  local_aggregation::ReportAggregate report_aggregate = GetReportAggregate(
      testing::cobalt_new::kSettingsChangedMetricId, MetricDefinition::OCCURRENCE,
      testing::cobalt_new::kSettingsChangedSettingsChangedUniqueDeviceHistogramsReportId, &store);
  ASSERT_TRUE(report_aggregate.has_daily());
  local_aggregation::AggregationPeriodBucket daily_bucket =
      report_aggregate.daily().by_day_index().begin()->second;
  local_aggregation::AggregateData data =
      daily_bucket.system_profile_aggregates(0).by_event_code().at(0).data();
  EXPECT_EQ(data.count(), 1);
}

TEST_F(EventLoggersLocalAggregationTest, IntegerFleetwideMeansValidation) {
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kStreamingTimeMetricId,
                                                   MetricDefinition::INTEGER);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER);
  EXPECT_EQ(StatusCode::OK,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kStreamingTimeMetricId,
                                          MetricDefinition::INTEGER, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, IntegerFleetwideMeansValidationWrongType) {
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kStreamingTimeMetricId,
                                                   MetricDefinition::OCCURRENCE);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER);
  EXPECT_EQ(StatusCode::INVALID_ARGUMENT,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kStreamingTimeMetricId,
                                          MetricDefinition::INTEGER, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, IntegerFleetwideMeans) {
  ReportDefinition report = ReportDefinitionWithReportType(
      testing::cobalt_new::kStreamingTimeStreamingTimeAggregationReportId,
      ReportDefinition::FLEETWIDE_MEANS);
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kStreamingTimeMetricId,
                                                   MetricDefinition::INTEGER);

  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER);
  EXPECT_EQ(
      StatusCode::OK,
      event_logger->Log(std::move(event_record), std::chrono::system_clock::now()).error_code());

  event_record = GetEventRecordWithMetricType(testing::cobalt_new::kStreamingTimeMetricId,
                                              MetricDefinition::INTEGER);
  local_aggregation::ImmediateLocalAggregateStorage store = GetLocalAggregateStorage();
  local_aggregation::ReportAggregate report_aggregate = GetReportAggregate(
      testing::cobalt_new::kStreamingTimeMetricId, MetricDefinition::INTEGER,
      testing::cobalt_new::kStreamingTimeStreamingTimeAggregationReportId, &store);
  ASSERT_TRUE(report_aggregate.has_hourly());
  local_aggregation::AggregationPeriodBucket hourly_bucket =
      report_aggregate.hourly().by_hour_id().begin()->second;
  local_aggregation::AggregateData data =
      hourly_bucket.system_profile_aggregates(0).by_event_code().at(0).data();
  ASSERT_TRUE(data.has_sum_and_count());
  EXPECT_EQ(data.sum_and_count().sum(), 10);
}

TEST_F(EventLoggersLocalAggregationTest, IntegerHistogramFleetwideHistogramsValidation) {
  auto event_record = GetEventRecordWithMetricType(
      testing::cobalt_new::kFileSystemWriteTimesMetricId, MetricDefinition::INTEGER_HISTOGRAM);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER_HISTOGRAM);
  EXPECT_EQ(StatusCode::OK,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kFileSystemWriteTimesMetricId,
                                          MetricDefinition::INTEGER_HISTOGRAM, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, IntegerHistogramFleetwideHistogramsValidationWrongType) {
  auto event_record = GetEventRecordWithMetricType(
      testing::cobalt_new::kFileSystemWriteTimesMetricId, MetricDefinition::INTEGER);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER_HISTOGRAM);
  EXPECT_EQ(StatusCode::INVALID_ARGUMENT,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kFileSystemWriteTimesMetricId,
                                          MetricDefinition::INTEGER_HISTOGRAM, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest,
       IntegerHistogramFleetwideHistogramsValidationWrongBuckets) {
  auto event_record = GetEventRecordWithMetricType(
      testing::cobalt_new::kFileSystemWriteTimesMetricId, MetricDefinition::INTEGER_HISTOGRAM);
  // Replace with a histogram containing an invalid bucket index. The metric FileSystemWriteTimes
  // has a num_buckets = 10, with 0 being the underflow bucket and 10+1=11 being the overflow
  // bucket. So the first invalid bucket index is 12.
  event_record->event()->mutable_integer_histogram_event()->mutable_buckets()->Swap(
      testing::NewHistogram({0, 1, 2, 12}, {100, 101, 102, 103}).get());
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER_HISTOGRAM);
  EXPECT_EQ(StatusCode::INVALID_ARGUMENT,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kFileSystemWriteTimesMetricId,
                                          MetricDefinition::INTEGER_HISTOGRAM, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, IntegerHistogramFleetwideHistograms) {
  ReportDefinition report = ReportDefinitionWithReportType(
      testing::cobalt_new::kFileSystemWriteTimesFileSystemWriteTimesHistogramReportId,
      ReportDefinition::FLEETWIDE_HISTOGRAMS);
  auto event_record = GetEventRecordWithMetricType(
      testing::cobalt_new::kFileSystemWriteTimesMetricId, MetricDefinition::INTEGER_HISTOGRAM);

  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::INTEGER_HISTOGRAM);
  EXPECT_EQ(
      StatusCode::OK,
      event_logger->Log(std::move(event_record), std::chrono::system_clock::now()).error_code());

  event_record = GetEventRecordWithMetricType(testing::cobalt_new::kFileSystemWriteTimesMetricId,
                                              MetricDefinition::INTEGER_HISTOGRAM);
  local_aggregation::ImmediateLocalAggregateStorage store = GetLocalAggregateStorage();
  local_aggregation::ReportAggregate report_aggregate = GetReportAggregate(
      testing::cobalt_new::kFileSystemWriteTimesMetricId, MetricDefinition::INTEGER_HISTOGRAM,
      testing::cobalt_new::kFileSystemWriteTimesFileSystemWriteTimesHistogramReportId, &store);
  ASSERT_TRUE(report_aggregate.has_hourly());
  local_aggregation::AggregationPeriodBucket hourly_bucket =
      report_aggregate.hourly().by_hour_id().begin()->second;
  local_aggregation::AggregateData data =
      hourly_bucket.system_profile_aggregates(0).by_event_code().at(0).data();
  ASSERT_TRUE(data.has_integer_histogram());
  const auto& histogram = data.integer_histogram();
  EXPECT_EQ(histogram.histogram().at(0), 100);
  EXPECT_EQ(histogram.histogram().at(1), 101);
  EXPECT_EQ(histogram.histogram().at(2), 102);
  EXPECT_EQ(histogram.histogram().at(3), 103);
}

TEST_F(EventLoggersLocalAggregationTest, StringStringHistogramsValidation) {
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kComponentMetricId,
                                                   MetricDefinition::STRING);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::STRING);
  EXPECT_EQ(StatusCode::OK,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kComponentMetricId,
                                          MetricDefinition::STRING, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, StringStringHistogramsValidationWrongType) {
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kComponentMetricId,
                                                   MetricDefinition::INTEGER);
  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::STRING);
  EXPECT_EQ(StatusCode::INVALID_ARGUMENT,
            event_logger
                ->PrepareAndValidateEvent(testing::cobalt_new::kComponentMetricId,
                                          MetricDefinition::STRING, event_record.get())
                .error_code());
}

TEST_F(EventLoggersLocalAggregationTest, StringStringHistograms) {
  ReportDefinition report = ReportDefinitionWithReportType(
      testing::cobalt_new::kComponentComponentHistogramReportId, ReportDefinition::STRING_COUNTS);
  auto event_record = GetEventRecordWithMetricType(testing::cobalt_new::kComponentMetricId,
                                                   MetricDefinition::STRING);

  auto event_logger = GetEventLoggerForMetricType(MetricDefinition::STRING);
  EXPECT_EQ(
      StatusCode::OK,
      event_logger->Log(std::move(event_record), std::chrono::system_clock::now()).error_code());

  event_record = GetEventRecordWithMetricType(testing::cobalt_new::kComponentMetricId,
                                              MetricDefinition::STRING);
  local_aggregation::ImmediateLocalAggregateStorage store = GetLocalAggregateStorage();
  local_aggregation::ReportAggregate report_aggregate =
      GetReportAggregate(testing::cobalt_new::kComponentMetricId, MetricDefinition::STRING,
                         testing::cobalt_new::kComponentComponentHistogramReportId, &store);
  ASSERT_TRUE(report_aggregate.has_hourly());
  local_aggregation::AggregationPeriodBucket hourly_bucket =
      report_aggregate.hourly().by_hour_id().begin()->second;
  local_aggregation::AggregateData data =
      hourly_bucket.system_profile_aggregates(0).by_event_code().at(0).data();
  ASSERT_TRUE(data.has_string_histogram());
  const auto& histogram = data.string_histogram();
  EXPECT_EQ(histogram.histogram().at(0), 1);
}

TEST_F(EventLoggersLocalAggregationTest, OccurrenceOtherTimeZone) {
  ReportDefinition report = ReportDefinitionWithReportType(
      testing::cobalt_new::kOccurrenceOtherTzFleetwideOccurrenceCountsReportId,
      ReportDefinition::FLEETWIDE_OCCURRENCE_COUNTS);
  std::unique_ptr<EventRecord> event_record = GetEventRecordWithMetricType(
      testing::cobalt_new::kOccurrenceOtherTzMetricId, MetricDefinition::OCCURRENCE);
  std::unique_ptr<EventLogger> event_logger =
      GetEventLoggerForMetricType(MetricDefinition::OCCURRENCE);

  // Log a count of 1 for start_time_, which has hour ID 17521 in the initial time zone of
  // `civil_time_converter_`.
  EXPECT_EQ(StatusCode::OK, event_logger->Log(std::move(event_record), start_time_).error_code());

  event_record = GetEventRecordWithMetricType(testing::cobalt_new::kOccurrenceOtherTzMetricId,
                                              MetricDefinition::OCCURRENCE);
  event_record->event()->mutable_occurrence_event()->set_count(2);

  // Log a count of 2 for `start_time_ + 1` hour, which has hour ID 17524 in the final time zone of
  // `civil_time_converter_`.
  EXPECT_EQ(StatusCode::OK,
            event_logger->Log(std::move(event_record), start_time_ + util::kOneHour).error_code());

  local_aggregation::ImmediateLocalAggregateStorage store = GetLocalAggregateStorage();
  local_aggregation::ReportAggregate report_aggregate = GetReportAggregate(
      testing::cobalt_new::kOccurrenceOtherTzMetricId, MetricDefinition::OCCURRENCE,
      testing::cobalt_new::kOccurrenceOtherTzFleetwideOccurrenceCountsReportId, &store);

  // Check that the EventLogger recorded the correct hour IDs and event counts.
  ASSERT_TRUE(report_aggregate.has_hourly());
  ASSERT_EQ(report_aggregate.hourly().by_hour_id_size(), 2);

  ASSERT_EQ(report_aggregate.hourly().by_hour_id().count(17521), 1);
  EXPECT_EQ(report_aggregate.hourly()
                .by_hour_id()
                .at(17521)
                .system_profile_aggregates(0)
                .by_event_code()
                .at(0)
                .data()
                .count(),
            1);

  ASSERT_EQ(report_aggregate.hourly().by_hour_id().count(17524), 1);
  EXPECT_EQ(report_aggregate.hourly()
                .by_hour_id()
                .at(17524)
                .system_profile_aggregates(0)
                .by_event_code()
                .at(0)
                .data()
                .count(),
            2);
}

}  // namespace internal
}  // namespace cobalt::logger