src/pb/metadata_builder.cc

// Copyright 2020 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/pb/metadata_builder.h"

#include <chrono>

#include "google/protobuf/repeated_field.h"
#include "src/lib/util/clock.h"
#include "src/lib/util/consistent_proto_store.h"
#include "src/lib/util/datetime_util.h"
#include "src/logging.h"
#include "src/pb/common.pb.h"
#include "src/pb/system_data_history.pb.h"
#include "src/registry/metric_definition.pb.h"
#include "src/registry/report_definition.pb.h"
#include "src/registry/window_size.pb.h"
#include "src/system_data/system_data.h"

namespace cobalt {

MetadataBuilder::MetadataBuilder(system_data::SystemDataInterface* system_data,
                                 util::ValidatedClockInterface* validated_clock,
                                 std::string system_profile_cache_dir, util::FileSystem* fs)
    : system_data_(system_data), validated_clock_(validated_clock) {
  if (system_data_) {
    system_data_->OnChange([this]() { SnapshotSystemData(); });
  }
  if (!validated_clock_) {
    owned_clock_ =
        std::make_unique<util::AlwaysAccurateClock>(std::make_unique<util::SystemClock>());
    validated_clock_ = owned_clock_.get();
  }
  if (!system_profile_cache_dir.empty()) {
    history_store_ =
        std::make_unique<util::ConsistentProtoStore>(std::move(system_profile_cache_dir), fs);
    history_store_->Read(&*system_profile_history_.lock());
  } else {
    LOG(WARNING) << "No system_profile_cache_dir provided to MetadataBuilder. The system profile "
                    "cache will not be persisted to disk";
  }
}

namespace {

template <class InputIterator>
std::optional<const SystemProfile*> FindProfile(
    InputIterator first, InputIterator last,
    const std::function<bool(const SystemProfileSnapshot&)>& comparator) {
  while (first != last) {
    if (comparator(*first)) {
      return &first->system_profile();
    }

    ++first;
  }

  return std::nullopt;
}

std::function<bool(const SystemProfileSnapshot&)> TZCompare(
    ReportDefinition::ReportType report_type, MetricDefinition::TimeZonePolicy time_zone,
    const std::function<bool(uint32_t)>& day_cmp, const std::function<bool(uint32_t)>& hour_cmp) {
  return [report_type, time_zone, hour_cmp, day_cmp](const SystemProfileSnapshot& snapshot) {
    switch (report_type) {
      case ReportDefinition::FLEETWIDE_OCCURRENCE_COUNTS:
      case ReportDefinition::HOURLY_VALUE_NUMERIC_STATS:
      case ReportDefinition::HOURLY_VALUE_HISTOGRAMS:
      case ReportDefinition::STRING_COUNTS:
        switch (time_zone) {
          case MetricDefinition::LOCAL:
            return hour_cmp(snapshot.hour_id_local());
          case MetricDefinition::UTC:
          default:
            return hour_cmp(snapshot.hour_id_utc());
        }
      default:
        switch (time_zone) {
          case MetricDefinition::LOCAL:
            return day_cmp(snapshot.day_index_local());
          case MetricDefinition::UTC:
          default:
            return day_cmp(snapshot.day_index_utc());
        }
    }
  };
}

std::optional<const SystemProfile*> FindProfileForSelectFirst(
    const SystemProfileHistory& system_profile_history, ReportDefinition::ReportType report_type,
    MetricDefinition::TimeZonePolicy time_zone, uint32_t day_index, uint32_t hour_id) {
  // Find the first system profile whose hour_id or day_index equals the expected one.
  std::optional<const SystemProfile*> profile = FindProfile(
      system_profile_history.snapshots().begin(), system_profile_history.snapshots().end(),
      TZCompare(
          report_type, time_zone,
          [day_index](uint32_t other_day_index) { return other_day_index == day_index; },
          [hour_id](uint32_t other_hour_id) { return other_hour_id == hour_id; }));

  if (!profile) {
    // Otherwise, find the *last* system profile whose hour_id or day_index is less than the
    // expected one.
    profile = FindProfile(
        system_profile_history.snapshots().rbegin(), system_profile_history.snapshots().rend(),
        TZCompare(
            report_type, time_zone,
            [day_index](uint32_t other_day_index) { return other_day_index < day_index; },
            [hour_id](uint32_t other_hour_id) { return other_hour_id < hour_id; }));
  }

  return profile;
}

std::optional<const SystemProfile*> FindProfileForSelectLast(
    const SystemProfileHistory& system_profile_history, ReportDefinition::ReportType report_type,
    MetricDefinition::TimeZonePolicy time_zone, uint32_t day_index, uint32_t hour_id) {
  // Find the last system profile whose hour_id or day_index is less than or equal to the
  // expected value.
  return FindProfile(
      system_profile_history.snapshots().rbegin(), system_profile_history.snapshots().rend(),
      TZCompare(
          report_type, time_zone,
          [day_index](uint32_t other_day_index) { return other_day_index <= day_index; },
          [hour_id](uint32_t other_hour_id) { return other_hour_id <= hour_id; }));
}

}  // namespace

std::unique_ptr<ObservationMetadata> MetadataBuilder::Build(const logger::MetricRef& metric,
                                                            const ReportDefinition& report,
                                                            uint32_t day_index,
                                                            uint32_t hour_id) const {
  auto metadata = std::make_unique<ObservationMetadata>();

  metadata->set_customer_id(metric.project().customer_id());
  metadata->set_project_id(metric.project().project_id());
  metadata->set_metric_id(metric.metric_id());
  metadata->set_report_id(report.id());
  metadata->set_day_index(day_index);

  std::optional<const SystemProfile*> profile = std::nullopt;

  switch (report.system_profile_selection()) {
    case SystemProfileSelectionPolicy::SELECT_FIRST: {
      auto selected_day_index = day_index;
      switch (report.local_aggregation_period()) {
        case WindowSize::UNSET:
          break;
        default:
          selected_day_index -= report.local_aggregation_period() - 1;
          break;
      }
      profile = FindProfileForSelectFirst(
          *system_profile_history_.const_lock(), report.report_type(),
          metric.metric_definition()->time_zone_policy(), selected_day_index, hour_id);

    } break;
    case SystemProfileSelectionPolicy::SELECT_LAST:
    case SystemProfileSelectionPolicy::SELECT_DEFAULT:
    default:
      profile = FindProfileForSelectLast(
          *system_profile_history_.const_lock(), report.report_type(),
          metric.metric_definition()->time_zone_policy(), day_index, hour_id);
      break;
  }

  if (profile) {
    FilteredSystemProfile(report, *profile.value(), metadata->mutable_system_profile());
  } else {
    // No snapshots made (problem?)
    FilteredSystemProfile(report, metadata->mutable_system_profile());
  }

  return metadata;
}  // namespace cobalt

void MetadataBuilder::FilteredSystemProfile(const ReportDefinition& report,
                                            SystemProfile* profile_out) const {
  if (system_data_) {
    FilteredSystemProfile(report, system_data_->system_profile(), profile_out);
  }
}

void MetadataBuilder::FilteredSystemProfile(const ReportDefinition& report,
                                            const SystemProfile& profile_in,
                                            SystemProfile* profile_out) {
  for (const auto& field : report.system_profile_field()) {
    switch (field) {
      case SystemProfileField::OS:
        profile_out->set_os(profile_in.os());
        break;
      case SystemProfileField::ARCH:
        profile_out->set_arch(profile_in.arch());
        break;
      case SystemProfileField::BOARD_NAME:
        profile_out->set_board_name(profile_in.board_name());
        break;
      case SystemProfileField::PRODUCT_NAME:
        profile_out->set_product_name(profile_in.product_name());
        break;
      case SystemProfileField::SYSTEM_VERSION:
        profile_out->set_system_version(profile_in.system_version());
        break;
      case SystemProfileField::CHANNEL:
        profile_out->set_channel(profile_in.channel());
        break;
      case SystemProfileField::REALM:
        profile_out->set_realm(profile_in.realm());
        break;
      case SystemProfileField::BUILD_TYPE:
        profile_out->set_build_type(profile_in.build_type());
        break;
    }
  }
}

void MetadataBuilder::SnapshotSystemData() {
  if (!system_data_) {
    // If there is no valid system_data_, we can do nothing.
    return;
  }

  std::optional<std::chrono::system_clock::time_point> now = validated_clock_->now();
  if (!now) {
    // The system clock is not accurate, we will wait for this method to be called again by
    // `CobaltService::SystemClockIsAccurate`
    return;
  }

  auto history = system_profile_history_.lock();
  if (history->snapshots_size() > 0 && history->snapshots().rbegin()->SerializeAsString() ==
                                           system_data_->system_profile().SerializeAsString()) {
    // The most recent snapshot is the same as the current system_profile. No need to snapshot
    // again.
    return;
  }

  SystemProfileSnapshot* snapshot = history->add_snapshots();
  snapshot->mutable_system_profile()->CopyFrom(system_data_->system_profile());

  util::TimeInfo utc = util::TimeInfo::FromTimePoint(*now, MetricDefinition::UTC);
  util::TimeInfo local = util::TimeInfo::FromTimePoint(*now, MetricDefinition::LOCAL);

  snapshot->set_day_index_utc(utc.day_index);
  snapshot->set_day_index_local(local.day_index);

  snapshot->set_hour_id_utc(utc.hour_id);
  snapshot->set_hour_id_local(local.hour_id);

  snapshot->set_timestamp(util::ToUnixSeconds(*now));

  WriteHistory(std::move(history));
}

void MetadataBuilder::WriteHistory(
    util::ProtectedFields<SystemProfileHistory>::LockedFieldsPtr history) {
  if (history_store_) {
    history_store_->Write(*history);
  }
}

void MetadataBuilder::CleanupBefore(std::chrono::hours hours_ago) {
  std::optional<std::chrono::system_clock::time_point> now = validated_clock_->now();
  if (!now) {
    // Without an accurate clock, cleanup is impossible.
    return;
  }

  std::chrono::system_clock::time_point delete_before = *now - hours_ago;
  uint32_t delete_before_timestamp = util::ToUnixSeconds(delete_before);

  google::protobuf::RepeatedPtrField<SystemProfileSnapshot> kept_snapshots;
  auto history = system_profile_history_.lock();
  for (auto& snapshot : *history->mutable_snapshots()) {
    if (snapshot.timestamp() >= delete_before_timestamp) {
      kept_snapshots.Add()->Swap(&snapshot);
    }
  }

  history->mutable_snapshots()->Swap(&kept_snapshots);

  WriteHistory(std::move(history));
}

}  // namespace cobalt