| // 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/local_aggregation/aggregate_store.h" |
| |
| #include <algorithm> |
| #include <map> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include "src/algorithms/rappor/rappor_config_helper.h" |
| #include "src/lib/util/datetime_util.h" |
| #include "src/lib/util/proto_util.h" |
| #include "src/lib/util/status.h" |
| #include "src/local_aggregation/aggregation_utils.h" |
| #include "src/logger/internal_metrics.h" |
| #include "src/registry/packed_event_codes.h" |
| |
| namespace cobalt::local_aggregation { |
| |
| using google::protobuf::RepeatedField; |
| using logger::Encoder; |
| using logger::kInvalidArguments; |
| using logger::kOK; |
| using logger::kOther; |
| using logger::MetricRef; |
| using logger::ObservationWriter; |
| using logger::ProjectContext; |
| using logger::Status; |
| using rappor::RapporConfigHelper; |
| using util::ConsistentProtoStore; |
| using util::SerializeToBase64; |
| using util::StatusCode; |
| |
| namespace { |
| |
| ////// General helper functions. |
| |
| // Populates a ReportAggregationKey proto message and then populates a string |
| // with the base64 encoding of the serialized proto. |
| bool PopulateReportKey(uint32_t customer_id, uint32_t project_id, uint32_t metric_id, |
| uint32_t report_id, std::string* key) { |
| ReportAggregationKey key_data; |
| key_data.set_customer_id(customer_id); |
| key_data.set_project_id(project_id); |
| key_data.set_metric_id(metric_id); |
| key_data.set_report_id(report_id); |
| return SerializeToBase64(key_data, key); |
| } |
| |
| ////// Helper functions used by the constructor and UpdateAggregationConfigs(). |
| |
| // Gets and validates the window sizes and/or aggregation windows from a ReportDefinition, converts |
| // window sizes to daily aggregation windows, sorts the aggregation windows in increasing order, and |
| // adds them to an AggregationConfig. |
| // |
| // TODO(pesk): Stop looking at the window_size field of |report| once all reports have been updated |
| // to have OnDeviceAggregationWindows only. |
| bool GetSortedAggregationWindowsFromReport(const ReportDefinition& report, |
| AggregationConfig* aggregation_config) { |
| if (report.window_size_size() == 0 && report.aggregation_window_size() == 0) { |
| LOG(ERROR) << "Report must have at least one window size or aggregation window."; |
| return false; |
| } |
| std::vector<uint32_t> aggregation_days; |
| std::vector<uint32_t> aggregation_hours; |
| for (const uint32_t window_size : report.window_size()) { |
| if (window_size == 0 || window_size > kMaxAllowedAggregationDays) { |
| LOG(ERROR) << "Window size must be positive and cannot exceed " << kMaxAllowedAggregationDays; |
| return false; |
| } |
| aggregation_days.push_back(window_size); |
| } |
| for (const auto& window : report.aggregation_window()) { |
| switch (window.units_case()) { |
| case OnDeviceAggregationWindow::kDays: { |
| uint32_t num_days = window.days(); |
| if (num_days == 0 || num_days > kMaxAllowedAggregationDays) { |
| LOG(ERROR) << "Daily windows must contain at least 1 and no more than " |
| << kMaxAllowedAggregationDays << " days"; |
| return false; |
| } |
| aggregation_days.push_back(num_days); |
| break; |
| } |
| case OnDeviceAggregationWindow::kHours: { |
| uint32_t num_hours = window.hours(); |
| if (num_hours == 0 || num_hours > kMaxAllowedAggregationHours) { |
| LOG(ERROR) << "Hourly windows must contain at least 1 and no more than " |
| << kMaxAllowedAggregationHours << " hours"; |
| return false; |
| } |
| aggregation_hours.push_back(num_hours); |
| break; |
| } |
| default: |
| LOG(ERROR) << "Invalid OnDeviceAggregationWindow type " << window.units_case(); |
| } |
| } |
| std::sort(aggregation_hours.begin(), aggregation_hours.end()); |
| std::sort(aggregation_days.begin(), aggregation_days.end()); |
| for (auto num_hours : aggregation_hours) { |
| *aggregation_config->add_aggregation_window() = MakeHourWindow(num_hours); |
| } |
| for (auto num_days : aggregation_days) { |
| *aggregation_config->add_aggregation_window() = MakeDayWindow(num_days); |
| } |
| return true; |
| } |
| |
| // Creates an AggregationConfig from a ProjectContext, MetricDefinition, and |
| // ReportDefinition and populates the aggregation_config field of a specified |
| // ReportAggregates. Also sets the type of the ReportAggregates based on the |
| // ReportDefinition's type. |
| // |
| // Accepts ReportDefinitions with either at least one WindowSize, or at least one |
| // OnDeviceAggregationWindow with units in days. |
| bool PopulateReportAggregates(const ProjectContext& project_context, const MetricDefinition& metric, |
| const ReportDefinition& report, ReportAggregates* report_aggregates) { |
| if (report.window_size_size() == 0 && report.aggregation_window_size() == 0) { |
| } |
| AggregationConfig* aggregation_config = report_aggregates->mutable_aggregation_config(); |
| *aggregation_config->mutable_project() = project_context.project(); |
| *aggregation_config->mutable_metric() = *project_context.GetMetric(metric.id()); |
| *aggregation_config->mutable_report() = report; |
| if (!GetSortedAggregationWindowsFromReport(report, aggregation_config)) { |
| return false; |
| } |
| switch (report.report_type()) { |
| case ReportDefinition::UNIQUE_N_DAY_ACTIVES: { |
| report_aggregates->set_allocated_unique_actives_aggregates(new UniqueActivesReportAggregates); |
| return true; |
| } |
| case ReportDefinition::PER_DEVICE_NUMERIC_STATS: |
| case ReportDefinition::PER_DEVICE_HISTOGRAM: { |
| report_aggregates->set_allocated_numeric_aggregates(new PerDeviceNumericAggregates); |
| return true; |
| } |
| default: |
| return false; |
| } |
| } |
| |
| // Move all items from the |window_size| field to the |aggregation_window| field |
| // of each AggregationConfig, preserving the order of the items. The |aggregation_window| field |
| // should be empty if the |window_size| field is nonempty. If for some reason this is not true, log |
| // an error and discard the contents of |aggregation_window| and replace them with the migrated |
| // |window_size| values. |
| void ConvertWindowSizesToAggregationDays(LocalAggregateStore* store) { |
| for (auto [key, aggregates] : store->by_report_key()) { |
| auto config = (*store->mutable_by_report_key())[key].mutable_aggregation_config(); |
| if (config->window_size_size() > 0 && config->aggregation_window_size() > 0) { |
| LOG(ERROR) << "Config has both a window_size and an aggregation_window; discarding all " |
| "aggregation_windows"; |
| config->clear_aggregation_window(); |
| } |
| for (auto window_size : config->window_size()) { |
| *config->add_aggregation_window() = MakeDayWindow(window_size); |
| } |
| config->clear_window_size(); |
| } |
| } |
| |
| // Upgrades the LocalAggregateStore from version 0 to |kCurrentLocalAggregateStoreVersion|. |
| Status UpgradeLocalAggregateStoreFromVersion0(LocalAggregateStore* store) { |
| ConvertWindowSizesToAggregationDays(store); |
| store->set_version(kCurrentLocalAggregateStoreVersion); |
| return kOK; |
| } |
| |
| } // namespace |
| |
| AggregateStore::AggregateStore(const Encoder* encoder, const ObservationWriter* observation_writer, |
| ConsistentProtoStore* local_aggregate_proto_store, |
| ConsistentProtoStore* obs_history_proto_store, |
| const size_t backfill_days) |
| : encoder_(encoder), |
| observation_writer_(observation_writer), |
| local_aggregate_proto_store_(local_aggregate_proto_store), |
| obs_history_proto_store_(obs_history_proto_store), |
| internal_metrics_(logger::InternalMetrics::NewWithLogger(nullptr)) { |
| CHECK_LE(backfill_days, kMaxAllowedBackfillDays) |
| << "backfill_days must be less than or equal to " << kMaxAllowedBackfillDays; |
| backfill_days_ = backfill_days; |
| auto locked_store = protected_aggregate_store_.lock(); |
| locked_store->empty_local_aggregate_store = MakeNewLocalAggregateStore(); |
| auto restore_aggregates_status = |
| local_aggregate_proto_store_->Read(&(locked_store->local_aggregate_store)); |
| switch (restore_aggregates_status.error_code()) { |
| case StatusCode::OK: { |
| VLOG(4) << "Read LocalAggregateStore from disk."; |
| break; |
| } |
| case StatusCode::NOT_FOUND: { |
| VLOG(4) << "No file found for local_aggregate_proto_store. Proceeding " |
| "with empty LocalAggregateStore. File will be created on " |
| "first snapshot of the LocalAggregateStore."; |
| locked_store->local_aggregate_store = MakeNewLocalAggregateStore(); |
| break; |
| } |
| default: { |
| LOG(ERROR) << "Read to local_aggregate_proto_store failed with status code: " |
| << restore_aggregates_status.error_code() |
| << "\nError message: " << restore_aggregates_status.error_message() |
| << "\nError details: " << restore_aggregates_status.error_details() |
| << "\nProceeding with empty LocalAggregateStore."; |
| locked_store->local_aggregate_store = MakeNewLocalAggregateStore(); |
| } |
| } |
| if (auto status = MaybeUpgradeLocalAggregateStore(&(locked_store->local_aggregate_store)); |
| status != kOK) { |
| LOG(ERROR) << "Failed to upgrade LocalAggregateStore to current version with status " << status |
| << ".\nProceeding with empty " |
| "LocalAggregateStore."; |
| locked_store->local_aggregate_store = MakeNewLocalAggregateStore(); |
| } |
| |
| auto locked_obs_history = protected_obs_history_.lock(); |
| auto restore_history_status = obs_history_proto_store_->Read(&locked_obs_history->obs_history); |
| switch (restore_history_status.error_code()) { |
| case StatusCode::OK: { |
| VLOG(4) << "Read AggregatedObservationHistoryStore from disk."; |
| break; |
| } |
| case StatusCode::NOT_FOUND: { |
| VLOG(4) << "No file found for obs_history_proto_store. Proceeding " |
| "with empty AggregatedObservationHistoryStore. File will be " |
| "created on first snapshot of the AggregatedObservationHistoryStore."; |
| break; |
| } |
| default: { |
| LOG(ERROR) << "Read to obs_history_proto_store failed with status code: " |
| << restore_history_status.error_code() |
| << "\nError message: " << restore_history_status.error_message() |
| << "\nError details: " << restore_history_status.error_details() |
| << "\nProceeding with empty AggregatedObservationHistoryStore."; |
| locked_obs_history->obs_history = MakeNewObservationHistoryStore(); |
| } |
| } |
| if (auto status = MaybeUpgradeObservationHistoryStore(&locked_obs_history->obs_history); |
| status != kOK) { |
| LOG(ERROR) |
| << "Failed to upgrade AggregatedObservationHistoryStore to current version with status " |
| << status << ".\nProceeding with empty AggregatedObservationHistoryStore."; |
| locked_obs_history->obs_history = MakeNewObservationHistoryStore(); |
| } |
| } |
| |
| Status AggregateStore::MaybeInsertReportConfig(const ProjectContext& project_context, |
| const MetricDefinition& metric, |
| const ReportDefinition& report) { |
| auto locked = protected_aggregate_store_.lock(); |
| std::string key; |
| if (!PopulateReportKey(project_context.project().customer_id(), |
| project_context.project().project_id(), metric.id(), report.id(), &key)) { |
| return kInvalidArguments; |
| } |
| ReportAggregates report_aggregates; |
| if (locked->local_aggregate_store.by_report_key().count(key) == 0) { |
| if (!PopulateReportAggregates(project_context, metric, report, &report_aggregates)) { |
| return kInvalidArguments; |
| } |
| (*locked->local_aggregate_store.mutable_by_report_key())[key] = report_aggregates; |
| } |
| |
| // Make sure that the 'empty' store has the key as well. |
| ReportAggregates empty_report_aggregates; |
| if (locked->empty_local_aggregate_store.by_report_key().count(key) == 0) { |
| if (!PopulateReportAggregates(project_context, metric, report, &empty_report_aggregates)) { |
| return kInvalidArguments; |
| } |
| (*locked->empty_local_aggregate_store.mutable_by_report_key())[key] = empty_report_aggregates; |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::SetActive(uint32_t customer_id, uint32_t project_id, uint32_t metric_id, |
| uint32_t report_id, uint64_t event_code, uint32_t day_index) { |
| if (is_disabled_) { |
| return kOK; |
| } |
| std::string key; |
| if (!PopulateReportKey(customer_id, project_id, metric_id, report_id, &key)) { |
| return kInvalidArguments; |
| } |
| |
| auto locked = protected_aggregate_store_.lock(); |
| auto aggregates = locked->local_aggregate_store.mutable_by_report_key()->find(key); |
| if (aggregates == locked->local_aggregate_store.mutable_by_report_key()->end()) { |
| LOG(ERROR) << "The Local Aggregate Store received an unexpected key."; |
| return kInvalidArguments; |
| } |
| if (!aggregates->second.has_unique_actives_aggregates()) { |
| LOG(ERROR) << "The local aggregates for this report key are not of type " |
| "UniqueActivesReportAggregates."; |
| return kInvalidArguments; |
| } |
| (*(*aggregates->second.mutable_unique_actives_aggregates()->mutable_by_event_code())[event_code] |
| .mutable_by_day_index())[day_index] |
| .mutable_activity_daily_aggregate() |
| ->set_activity_indicator(true); |
| return kOK; |
| } |
| |
| Status AggregateStore::UpdateNumericAggregate(uint32_t customer_id, uint32_t project_id, |
| uint32_t metric_id, uint32_t report_id, |
| const std::string& component, uint64_t event_code, |
| uint32_t day_index, int64_t value) { |
| if (is_disabled_) { |
| return kOK; |
| } |
| std::string report_key; |
| if (!PopulateReportKey(customer_id, project_id, metric_id, report_id, &report_key)) { |
| return kInvalidArguments; |
| } |
| |
| auto locked = protected_aggregate_store_.lock(); |
| auto aggregates = locked->local_aggregate_store.mutable_by_report_key()->find(report_key); |
| if (aggregates == locked->local_aggregate_store.mutable_by_report_key()->end()) { |
| LOG(ERROR) << "The Local Aggregate Store received an unexpected key."; |
| return kInvalidArguments; |
| } |
| if (!aggregates->second.has_numeric_aggregates()) { |
| LOG(ERROR) << "The local aggregates for this report key are not of a " |
| "compatible type."; |
| return kInvalidArguments; |
| } |
| |
| auto aggregates_by_day = |
| (*(*aggregates->second.mutable_numeric_aggregates()->mutable_by_component())[component] |
| .mutable_by_event_code())[event_code] |
| .mutable_by_day_index(); |
| bool has_stored_aggregate = ((*aggregates_by_day).find(day_index) != aggregates_by_day->end()); |
| auto day_aggregate = (*aggregates_by_day)[day_index].mutable_numeric_daily_aggregate(); |
| |
| auto [status, updated_value] = GetUpdatedAggregate( |
| aggregates->second.aggregation_config().report().aggregation_type(), |
| has_stored_aggregate ? std::optional<int64_t>{day_aggregate->value()} : std::nullopt, value); |
| if (status == kOK) { |
| day_aggregate->set_value(updated_value); |
| return kOK; |
| } |
| |
| return status; |
| } |
| |
| RepeatedField<uint32_t> UnpackEventCodesProto(uint64_t packed_event_codes) { |
| RepeatedField<uint32_t> fields; |
| for (auto code : config::UnpackEventCodes(packed_event_codes)) { |
| *fields.Add() = code; |
| } |
| return fields; |
| } |
| |
| Status AggregateStore::BackUpLocalAggregateStore() { |
| // Lock, copy the LocalAggregateStore, and release the lock. Write the copy |
| // to |local_aggregate_proto_store_|. |
| auto local_aggregate_store = CopyLocalAggregateStore(); |
| size_t store_size = local_aggregate_store.ByteSizeLong(); |
| // Cannot cause a loop, since TrackDiskUsage does not use any Cobalt 1.0 metrics |
| internal_metrics_->TrackDiskUsage(logger::InternalMetrics::StorageClass::AggregateStore, |
| static_cast<int64_t>(store_size)); |
| auto status = local_aggregate_proto_store_->Write(local_aggregate_store); |
| if (!status.ok()) { |
| LOG(ERROR) << "Failed to back up the LocalAggregateStore with error code: " |
| << status.error_code() << "\nError message: " << status.error_message() |
| << "\nError details: " << status.error_details(); |
| return kOther; |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::BackUpObservationHistory() { |
| auto obs_history = protected_obs_history_.lock()->obs_history; |
| size_t history_size = obs_history.ByteSizeLong(); |
| // Cannot cause a loop, since TrackDiskUsage does not use any Cobalt 1.0 metrics |
| internal_metrics_->TrackDiskUsage(logger::InternalMetrics::StorageClass::ObservationHistory, |
| static_cast<int64_t>(history_size)); |
| auto status = obs_history_proto_store_->Write(obs_history); |
| if (!status.ok()) { |
| LOG(ERROR) << "Failed to back up the AggregatedObservationHistoryStore. " |
| "::cobalt::util::Status error code: " |
| << status.error_code() << "\nError message: " << status.error_message() |
| << "\nError details: " << status.error_details(); |
| return kOther; |
| } |
| return kOK; |
| } |
| |
| ////////////////////// GarbageCollect and helper functions ////////////////// |
| |
| namespace { |
| |
| void GarbageCollectUniqueActivesReportAggregates(uint32_t day_index, uint32_t max_aggregation_days, |
| uint32_t backfill_days, |
| UniqueActivesReportAggregates* report_aggregates) { |
| auto map_by_event_code = report_aggregates->mutable_by_event_code(); |
| for (auto event_code = map_by_event_code->begin(); event_code != map_by_event_code->end();) { |
| auto map_by_day = event_code->second.mutable_by_day_index(); |
| for (auto day = map_by_day->begin(); day != map_by_day->end();) { |
| if (day->first <= day_index - backfill_days - max_aggregation_days) { |
| day = map_by_day->erase(day); |
| } else { |
| ++day; |
| } |
| } |
| if (map_by_day->empty()) { |
| event_code = map_by_event_code->erase(event_code); |
| } else { |
| ++event_code; |
| } |
| } |
| } |
| |
| void GarbageCollectNumericReportAggregates(uint32_t day_index, uint32_t max_aggregation_days, |
| uint32_t backfill_days, |
| PerDeviceNumericAggregates* report_aggregates) { |
| auto map_by_component = report_aggregates->mutable_by_component(); |
| for (auto component = map_by_component->begin(); component != map_by_component->end();) { |
| auto map_by_event_code = component->second.mutable_by_event_code(); |
| for (auto event_code = map_by_event_code->begin(); event_code != map_by_event_code->end();) { |
| auto map_by_day = event_code->second.mutable_by_day_index(); |
| for (auto day = map_by_day->begin(); day != map_by_day->end();) { |
| if (day->first <= day_index - backfill_days - max_aggregation_days) { |
| day = map_by_day->erase(day); |
| } else { |
| ++day; |
| } |
| } |
| if (map_by_day->empty()) { |
| event_code = map_by_event_code->erase(event_code); |
| } else { |
| ++event_code; |
| } |
| } |
| if (map_by_event_code->empty()) { |
| component = map_by_component->erase(component); |
| } else { |
| ++component; |
| } |
| } |
| } |
| |
| } // namespace |
| |
| Status AggregateStore::GarbageCollect(uint32_t day_index_utc, uint32_t day_index_local) { |
| if (day_index_local == 0u) { |
| day_index_local = day_index_utc; |
| } |
| CHECK_LT(day_index_utc, UINT32_MAX); |
| CHECK_LT(day_index_local, UINT32_MAX); |
| CHECK_GE(day_index_utc, kMaxAllowedAggregationDays + backfill_days_); |
| CHECK_GE(day_index_local, kMaxAllowedAggregationDays + backfill_days_); |
| |
| auto locked = protected_aggregate_store_.lock(); |
| for (const auto& [report_key, aggregates] : locked->local_aggregate_store.by_report_key()) { |
| uint32_t day_index; |
| const auto& config = aggregates.aggregation_config(); |
| switch (config.metric().time_zone_policy()) { |
| case MetricDefinition::UTC: { |
| day_index = day_index_utc; |
| break; |
| } |
| case MetricDefinition::LOCAL: { |
| day_index = day_index_local; |
| break; |
| } |
| default: |
| LOG_FIRST_N(ERROR, 10) << "The TimeZonePolicy of this MetricDefinition is invalid."; |
| continue; |
| } |
| if (aggregates.aggregation_config().aggregation_window_size() == 0) { |
| LOG_FIRST_N(ERROR, 10) << "This ReportDefinition does not have an aggregation window."; |
| continue; |
| } |
| // PopulateReportAggregates ensured that aggregation_window has at least one element, that all |
| // aggregation windows are <= kMaxAllowedAggregationDays, and that config.aggregation_window() |
| // is sorted in increasing order. |
| uint32_t max_aggregation_days = 1u; |
| const OnDeviceAggregationWindow& largest_window = |
| config.aggregation_window(config.aggregation_window_size() - 1); |
| if (largest_window.units_case() == OnDeviceAggregationWindow::kDays) { |
| max_aggregation_days = largest_window.days(); |
| } |
| if (max_aggregation_days == 0u || max_aggregation_days > day_index) { |
| LOG_FIRST_N(ERROR, 10) << "The maximum number of aggregation days " << max_aggregation_days |
| << " of this ReportDefinition is out of range."; |
| continue; |
| } |
| // For each ReportAggregates, descend to and iterate over the sub-map of |
| // local aggregates keyed by day index. Keep buckets with day indices |
| // greater than |day_index| - |backfill_days_| - |max_aggregation_days|, and |
| // remove all buckets with smaller day indices. |
| switch (aggregates.type_case()) { |
| case ReportAggregates::kUniqueActivesAggregates: { |
| GarbageCollectUniqueActivesReportAggregates( |
| day_index, max_aggregation_days, backfill_days_, |
| locked->local_aggregate_store.mutable_by_report_key() |
| ->at(report_key) |
| .mutable_unique_actives_aggregates()); |
| break; |
| } |
| case ReportAggregates::kNumericAggregates: { |
| GarbageCollectNumericReportAggregates(day_index, max_aggregation_days, backfill_days_, |
| locked->local_aggregate_store.mutable_by_report_key() |
| ->at(report_key) |
| .mutable_numeric_aggregates()); |
| break; |
| } |
| default: |
| continue; |
| } |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::GenerateObservations(uint32_t final_day_index_utc, |
| uint32_t final_day_index_local) { |
| if (final_day_index_local == 0u) { |
| final_day_index_local = final_day_index_utc; |
| } |
| CHECK_LT(final_day_index_utc, UINT32_MAX); |
| CHECK_LT(final_day_index_local, UINT32_MAX); |
| CHECK_GE(final_day_index_utc, kMaxAllowedAggregationDays + backfill_days_); |
| CHECK_GE(final_day_index_local, kMaxAllowedAggregationDays + backfill_days_); |
| |
| // Lock, copy the LocalAggregateStore, and release the lock. Use the copy to |
| // generate observations. |
| auto local_aggregate_store = CopyLocalAggregateStore(); |
| for (const auto& [report_key, aggregates] : local_aggregate_store.by_report_key()) { |
| const auto& config = aggregates.aggregation_config(); |
| |
| const auto& metric = config.metric(); |
| auto metric_ref = MetricRef(&config.project(), &metric); |
| uint32_t final_day_index; |
| switch (metric.time_zone_policy()) { |
| case MetricDefinition::UTC: { |
| final_day_index = final_day_index_utc; |
| break; |
| } |
| case MetricDefinition::LOCAL: { |
| final_day_index = final_day_index_local; |
| break; |
| } |
| default: |
| LOG_FIRST_N(ERROR, 10) << "The TimeZonePolicy of this MetricDefinition is invalid."; |
| continue; |
| } |
| |
| const auto& report = config.report(); |
| // PopulateReportAggregates ensured that aggregation_window has at least one element, that all |
| // aggregation windows are <= kMaxAllowedAggregationDays, and that config.aggregation_window() |
| // is sorted in increasing order. |
| if (config.aggregation_window_size() == 0u) { |
| LOG_FIRST_N(ERROR, 10) << "No aggregation_window found for this report."; |
| continue; |
| } |
| uint32_t max_aggregation_days = 1u; |
| const OnDeviceAggregationWindow& largest_window = |
| config.aggregation_window(config.aggregation_window_size() - 1); |
| if (largest_window.units_case() == OnDeviceAggregationWindow::kDays) { |
| max_aggregation_days = largest_window.days(); |
| } |
| if (max_aggregation_days == 0u || max_aggregation_days > final_day_index) { |
| LOG_FIRST_N(ERROR, 10) << "The maximum number of aggregation days " << max_aggregation_days |
| << " of this ReportDefinition is out of range."; |
| continue; |
| } |
| switch (metric.metric_type()) { |
| case MetricDefinition::EVENT_OCCURRED: { |
| auto num_event_codes = RapporConfigHelper::BasicRapporNumCategories(metric); |
| |
| switch (report.report_type()) { |
| case ReportDefinition::UNIQUE_N_DAY_ACTIVES: { |
| auto status = GenerateUniqueActivesObservations(metric_ref, report_key, aggregates, |
| num_event_codes, final_day_index); |
| if (status != kOK) { |
| return status; |
| } |
| break; |
| } |
| default: |
| continue; |
| } |
| break; |
| } |
| case MetricDefinition::EVENT_COUNT: |
| case MetricDefinition::ELAPSED_TIME: |
| case MetricDefinition::FRAME_RATE: |
| case MetricDefinition::MEMORY_USAGE: { |
| switch (report.report_type()) { |
| case ReportDefinition::PER_DEVICE_NUMERIC_STATS: |
| case ReportDefinition::PER_DEVICE_HISTOGRAM: { |
| auto status = GenerateObsFromNumericAggregates(metric_ref, report_key, aggregates, |
| final_day_index); |
| if (status != kOK) { |
| return status; |
| } |
| break; |
| } |
| default: |
| continue; |
| } |
| break; |
| } |
| default: |
| continue; |
| } |
| } |
| return kOK; |
| } |
| |
| ////////// GenerateUniqueActivesObservations and helper methods //////////////// |
| |
| namespace { |
| |
| // Given the set of daily aggregates for a fixed event code, and the size and |
| // end date of an aggregation window, returns the first day index within that |
| // window on which the event code occurred. Returns 0 if the event code did |
| // not occur within the window. |
| uint32_t FirstActiveDayIndexInWindow(const DailyAggregates& daily_aggregates, |
| uint32_t obs_day_index, uint32_t aggregation_days) { |
| for (uint32_t day_index = obs_day_index - aggregation_days + 1; day_index <= obs_day_index; |
| day_index++) { |
| auto day_aggregate = daily_aggregates.by_day_index().find(day_index); |
| if (day_aggregate != daily_aggregates.by_day_index().end() && |
| day_aggregate->second.activity_daily_aggregate().activity_indicator()) { |
| return day_index; |
| } |
| } |
| return 0u; |
| } |
| |
| // Given the day index of an event occurrence and the size and end date |
| // of an aggregation window, returns true if the occurrence falls within |
| // the window and false if not. |
| bool IsActivityInWindow(uint32_t active_day_index, uint32_t obs_day_index, |
| uint32_t aggregation_days) { |
| return (active_day_index <= obs_day_index && active_day_index > obs_day_index - aggregation_days); |
| } |
| |
| } // namespace |
| |
| uint32_t AggregateStore::GetUniqueActivesLastGeneratedDayIndex(const std::string& report_key, |
| uint32_t event_code, |
| uint32_t aggregation_days) const { |
| auto obs_history = protected_obs_history_.const_lock()->obs_history; |
| auto report_history = obs_history.by_report_key().find(report_key); |
| if (report_history == obs_history.by_report_key().end()) { |
| return 0u; |
| } |
| auto event_code_history = |
| report_history->second.unique_actives_history().by_event_code().find(event_code); |
| if (event_code_history == report_history->second.unique_actives_history().by_event_code().end()) { |
| return 0u; |
| } |
| auto window_history = event_code_history->second.by_window_size().find(aggregation_days); |
| if (window_history == event_code_history->second.by_window_size().end()) { |
| return 0u; |
| } |
| return window_history->second; |
| } |
| |
| void AggregateStore::SetUniqueActivesLastGeneratedDayIndex(const std::string& report_key, |
| uint32_t event_code, |
| uint32_t aggregation_days, |
| uint32_t value) { |
| auto locked = protected_obs_history_.lock(); |
| (*(*(*locked->obs_history.mutable_by_report_key())[report_key] |
| .mutable_unique_actives_history() |
| ->mutable_by_event_code())[event_code] |
| .mutable_by_window_size())[aggregation_days] = value; |
| } |
| |
| Status AggregateStore::GenerateSingleUniqueActivesObservation( |
| const MetricRef metric_ref, const ReportDefinition* report, uint32_t obs_day_index, |
| uint32_t event_code, const OnDeviceAggregationWindow& window, bool was_active) const { |
| auto encoder_result = encoder_->EncodeUniqueActivesObservation(metric_ref, report, obs_day_index, |
| event_code, was_active, window); |
| if (encoder_result.status != kOK) { |
| return encoder_result.status; |
| } |
| if (encoder_result.observation == nullptr || encoder_result.metadata == nullptr) { |
| LOG(ERROR) << "Failed to encode UniqueActivesObservation"; |
| return kOther; |
| } |
| |
| auto writer_status = observation_writer_->WriteObservation(encoder_result.observation, |
| std::move(encoder_result.metadata)); |
| if (writer_status != kOK) { |
| return writer_status; |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::GenerateUniqueActivesObservations(const MetricRef metric_ref, |
| const std::string& report_key, |
| const ReportAggregates& report_aggregates, |
| uint32_t num_event_codes, |
| uint32_t final_day_index) { |
| CHECK_GT(final_day_index, backfill_days_); |
| // The earliest day index for which we might need to generate an |
| // Observation. |
| auto backfill_period_start = uint32_t(final_day_index - backfill_days_); |
| |
| for (uint32_t event_code = 0; event_code < num_event_codes; event_code++) { |
| auto daily_aggregates = |
| report_aggregates.unique_actives_aggregates().by_event_code().find(event_code); |
| // Have any events ever been logged for this report and event code? |
| bool found_event_code = |
| (daily_aggregates != report_aggregates.unique_actives_aggregates().by_event_code().end()); |
| for (const auto& window : report_aggregates.aggregation_config().aggregation_window()) { |
| // Skip all hourly windows, and all daily windows which are larger than |
| // kMaxAllowedAggregationDays. |
| // |
| // TODO(pesk): Generate observations for hourly windows. |
| if (window.units_case() != OnDeviceAggregationWindow::kDays) { |
| LOG(INFO) << "Skipping unsupported aggregation window."; |
| continue; |
| } |
| if (window.days() > kMaxAllowedAggregationDays) { |
| LOG(WARNING) << "GenerateUniqueActivesObservations ignoring a window " |
| "size exceeding the maximum allowed value"; |
| continue; |
| } |
| // Find the earliest day index for which an Observation has not yet |
| // been generated for this report, event code, and window size. If |
| // that day index is later than |final_day_index|, no Observation is |
| // generated on this invocation. |
| auto last_gen = GetUniqueActivesLastGeneratedDayIndex(report_key, event_code, window.days()); |
| auto first_day_index = std::max(last_gen + 1, backfill_period_start); |
| // The latest day index on which |event_type| is known to have |
| // occurred, so far. This value will be updated as we search |
| // forward from the earliest day index belonging to a window of |
| // interest. |
| uint32_t active_day_index = 0u; |
| // Iterate over the day indices |obs_day_index| for which we need |
| // to generate Observations. On each iteration, generate an |
| // Observation for the |window| ending on |obs_day_index|. |
| for (uint32_t obs_day_index = first_day_index; obs_day_index <= final_day_index; |
| obs_day_index++) { |
| bool was_active = false; |
| if (found_event_code) { |
| // If the current value of |active_day_index| falls within the |
| // window, generate an Observation of activity. If not, search |
| // forward in the window, update |active_day_index|, and generate an |
| // Observation of activity or inactivity depending on the result of |
| // the search. |
| if (IsActivityInWindow(active_day_index, obs_day_index, window.days())) { |
| was_active = true; |
| } else { |
| active_day_index = |
| FirstActiveDayIndexInWindow(daily_aggregates->second, obs_day_index, window.days()); |
| was_active = IsActivityInWindow(active_day_index, obs_day_index, window.days()); |
| } |
| } |
| auto status = GenerateSingleUniqueActivesObservation( |
| metric_ref, &report_aggregates.aggregation_config().report(), obs_day_index, event_code, |
| window, was_active); |
| if (status != kOK) { |
| return status; |
| } |
| |
| SetUniqueActivesLastGeneratedDayIndex(report_key, event_code, window.days(), obs_day_index); |
| } |
| } |
| } |
| return kOK; |
| } |
| |
| ////////// GenerateObsFromNumericAggregates and helper methods ///////////// |
| |
| uint32_t AggregateStore::GetPerDeviceNumericLastGeneratedDayIndex(const std::string& report_key, |
| const std::string& component, |
| uint32_t event_code, |
| uint32_t aggregation_days) const { |
| auto obs_history = protected_obs_history_.const_lock()->obs_history; |
| const auto& report_history = obs_history.by_report_key().find(report_key); |
| if (report_history == obs_history.by_report_key().end()) { |
| return 0u; |
| } |
| if (!report_history->second.has_per_device_numeric_history()) { |
| return 0u; |
| } |
| const auto& component_history = |
| report_history->second.per_device_numeric_history().by_component().find(component); |
| if (component_history == |
| report_history->second.per_device_numeric_history().by_component().end()) { |
| return 0u; |
| } |
| const auto& event_code_history = component_history->second.by_event_code().find(event_code); |
| if (event_code_history == component_history->second.by_event_code().end()) { |
| return 0u; |
| } |
| const auto& window_history = event_code_history->second.by_window_size().find(aggregation_days); |
| if (window_history == event_code_history->second.by_window_size().end()) { |
| return 0u; |
| } |
| return window_history->second; |
| } |
| |
| void AggregateStore::SetPerDeviceNumericLastGeneratedDayIndex(const std::string& report_key, |
| const std::string& component, |
| uint32_t event_code, |
| uint32_t aggregation_days, |
| uint32_t value) { |
| auto locked = protected_obs_history_.lock(); |
| (*(*(*(*locked->obs_history.mutable_by_report_key())[report_key] |
| .mutable_per_device_numeric_history() |
| ->mutable_by_component())[component] |
| .mutable_by_event_code())[event_code] |
| .mutable_by_window_size())[aggregation_days] = value; |
| } |
| |
| uint32_t AggregateStore::GetReportParticipationLastGeneratedDayIndex( |
| const std::string& report_key) const { |
| auto obs_history = protected_obs_history_.const_lock()->obs_history; |
| const auto& report_history = obs_history.by_report_key().find(report_key); |
| if (report_history == obs_history.by_report_key().end()) { |
| return 0u; |
| } |
| return report_history->second.report_participation_history().last_generated(); |
| } |
| |
| void AggregateStore::SetReportParticipationLastGeneratedDayIndex(const std::string& report_key, |
| uint32_t value) { |
| auto locked = protected_obs_history_.lock(); |
| (*locked->obs_history.mutable_by_report_key())[report_key] |
| .mutable_report_participation_history() |
| ->set_last_generated(value); |
| } |
| |
| void AggregateStore::DeleteData() { |
| LOG(INFO) << "AggregateStore: Deleting stored data"; |
| |
| { |
| auto locked = protected_aggregate_store_.lock(); |
| locked->local_aggregate_store = locked->empty_local_aggregate_store; |
| } |
| protected_obs_history_.lock()->obs_history = MakeNewObservationHistoryStore(); |
| } |
| |
| void AggregateStore::Disable(bool is_disabled) { |
| LOG(INFO) << "AggregateStore: " << (is_disabled ? "Disabling" : "Enabling") |
| << " event aggregate storage."; |
| is_disabled_ = is_disabled; |
| } |
| |
| Status AggregateStore::GenerateSinglePerDeviceNumericObservation( |
| const MetricRef metric_ref, const ReportDefinition* report, uint32_t obs_day_index, |
| const std::string& component, uint32_t event_code, const OnDeviceAggregationWindow& window, |
| int64_t value) const { |
| Encoder::Result encoder_result = |
| encoder_->EncodePerDeviceNumericObservation(metric_ref, report, obs_day_index, component, |
| UnpackEventCodesProto(event_code), value, window); |
| if (encoder_result.status != kOK) { |
| return encoder_result.status; |
| } |
| if (encoder_result.observation == nullptr || encoder_result.metadata == nullptr) { |
| LOG(ERROR) << "Failed to encode PerDeviceNumericObservation"; |
| return kOther; |
| } |
| |
| const auto& writer_status = observation_writer_->WriteObservation( |
| encoder_result.observation, std::move(encoder_result.metadata)); |
| if (writer_status != kOK) { |
| return writer_status; |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::GenerateSinglePerDeviceHistogramObservation( |
| const MetricRef metric_ref, const ReportDefinition* report, uint32_t obs_day_index, |
| const std::string& component, uint32_t event_code, const OnDeviceAggregationWindow& window, |
| int64_t value) const { |
| Encoder::Result encoder_result = encoder_->EncodePerDeviceHistogramObservation( |
| metric_ref, report, obs_day_index, component, UnpackEventCodesProto(event_code), value, |
| window); |
| |
| if (encoder_result.status != kOK) { |
| return encoder_result.status; |
| } |
| if (encoder_result.observation == nullptr || encoder_result.metadata == nullptr) { |
| LOG(ERROR) << "Failed to encode PerDeviceNumericObservation"; |
| return kOther; |
| } |
| |
| const auto& writer_status = observation_writer_->WriteObservation( |
| encoder_result.observation, std::move(encoder_result.metadata)); |
| if (writer_status != kOK) { |
| return writer_status; |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::GenerateSingleReportParticipationObservation(const MetricRef metric_ref, |
| const ReportDefinition* report, |
| uint32_t obs_day_index) const { |
| auto encoder_result = |
| encoder_->EncodeReportParticipationObservation(metric_ref, report, obs_day_index); |
| if (encoder_result.status != kOK) { |
| return encoder_result.status; |
| } |
| if (encoder_result.observation == nullptr || encoder_result.metadata == nullptr) { |
| LOG(ERROR) << "Failed to encode ReportParticipationObservation"; |
| return kOther; |
| } |
| |
| const auto& writer_status = observation_writer_->WriteObservation( |
| encoder_result.observation, std::move(encoder_result.metadata)); |
| if (writer_status != kOK) { |
| return writer_status; |
| } |
| return kOK; |
| } |
| |
| Status AggregateStore::GenerateObsFromNumericAggregates(const MetricRef metric_ref, |
| const std::string& report_key, |
| const ReportAggregates& report_aggregates, |
| uint32_t final_day_index) { |
| CHECK_GT(final_day_index, backfill_days_); |
| // The first day index for which we might have to generate an Observation. |
| auto backfill_period_start = uint32_t(final_day_index - backfill_days_); |
| |
| // Generate any necessary PerDeviceNumericObservations for this report. |
| for (const auto& [component, event_code_aggregates] : |
| report_aggregates.numeric_aggregates().by_component()) { |
| for (const auto& [event_code, daily_aggregates] : event_code_aggregates.by_event_code()) { |
| // Populate a helper map keyed by day indices which belong to the range |
| // [|backfill_period_start|, |final_day_index|]. The value at a day |
| // index is the list of windows, in increasing size order, for which an |
| // Observation should be generated for that day index. |
| std::map<uint32_t, std::vector<OnDeviceAggregationWindow>> windows_by_obs_day; |
| for (const auto& window : report_aggregates.aggregation_config().aggregation_window()) { |
| if (window.units_case() != OnDeviceAggregationWindow::kDays) { |
| LOG(INFO) << "Skipping unsupported aggregation window."; |
| continue; |
| } |
| auto last_gen = GetPerDeviceNumericLastGeneratedDayIndex(report_key, component, event_code, |
| window.days()); |
| auto first_day_index = std::max(last_gen + 1, backfill_period_start); |
| for (auto obs_day_index = first_day_index; obs_day_index <= final_day_index; |
| obs_day_index++) { |
| windows_by_obs_day[obs_day_index].push_back(window); |
| } |
| } |
| // Iterate over the day indices |obs_day_index| for which we might need |
| // to generate an Observation. For each day index, generate an |
| // Observation for each needed window. |
| // |
| // More precisely, for each needed window, iterate over the days within that window. If at |
| // least one numeric event was logged during the window, compute the aggregate of the numeric |
| // values over the window and generate a PerDeviceNumericObservation. Whether or not a numeric |
| // event was found, update the AggregatedObservationHistory for this report, component, event |
| // code, and window size with |obs_day_index| as the most recent date of Observation |
| // generation. This reflects the fact that all needed Observations were generated for the |
| // window ending on that date. |
| for (auto obs_day_index = backfill_period_start; obs_day_index <= final_day_index; |
| obs_day_index++) { |
| const auto& windows = windows_by_obs_day.find(obs_day_index); |
| if (windows == windows_by_obs_day.end()) { |
| continue; |
| } |
| bool found_value_for_window = false; |
| int64_t window_aggregate = 0; |
| uint32_t num_days = 0; |
| for (const auto& window : windows->second) { |
| while (num_days < window.days()) { |
| bool found_value_for_day = false; |
| const auto& day_aggregates = |
| daily_aggregates.by_day_index().find(obs_day_index - num_days); |
| if (day_aggregates != daily_aggregates.by_day_index().end()) { |
| found_value_for_day = true; |
| } |
| const auto& aggregation_type = |
| report_aggregates.aggregation_config().report().aggregation_type(); |
| switch (aggregation_type) { |
| case ReportDefinition::SUM: |
| if (found_value_for_day) { |
| window_aggregate += day_aggregates->second.numeric_daily_aggregate().value(); |
| found_value_for_window = true; |
| } |
| break; |
| case ReportDefinition::MAX: |
| if (found_value_for_day) { |
| window_aggregate = std::max( |
| window_aggregate, day_aggregates->second.numeric_daily_aggregate().value()); |
| found_value_for_window = true; |
| } |
| break; |
| case ReportDefinition::MIN: |
| if (found_value_for_day && !found_value_for_window) { |
| window_aggregate = day_aggregates->second.numeric_daily_aggregate().value(); |
| found_value_for_window = true; |
| } else if (found_value_for_day) { |
| window_aggregate = std::min( |
| window_aggregate, day_aggregates->second.numeric_daily_aggregate().value()); |
| } |
| break; |
| default: |
| LOG(ERROR) << "Unexpected aggregation type " << aggregation_type; |
| return kInvalidArguments; |
| } |
| num_days++; |
| } |
| if (found_value_for_window) { |
| Status status; |
| const ReportDefinition* report = &report_aggregates.aggregation_config().report(); |
| switch (report->report_type()) { |
| case ReportDefinition::PER_DEVICE_NUMERIC_STATS: { |
| status = GenerateSinglePerDeviceNumericObservation( |
| metric_ref, report, obs_day_index, component, event_code, window, |
| window_aggregate); |
| if (status != kOK) { |
| return status; |
| } |
| break; |
| } |
| case ReportDefinition::PER_DEVICE_HISTOGRAM: { |
| auto status = GenerateSinglePerDeviceHistogramObservation( |
| metric_ref, report, obs_day_index, component, event_code, window, |
| window_aggregate); |
| if (status != kOK) { |
| return status; |
| } |
| break; |
| } |
| default: |
| LOG(ERROR) << "Unexpected report type " << report->report_type(); |
| return kInvalidArguments; |
| } |
| } |
| |
| SetPerDeviceNumericLastGeneratedDayIndex(report_key, component, event_code, window.days(), |
| obs_day_index); |
| } |
| } |
| } |
| } |
| // Generate any necessary ReportParticipationObservations for this report. |
| auto participation_last_gen = GetReportParticipationLastGeneratedDayIndex(report_key); |
| auto participation_first_day_index = std::max(participation_last_gen + 1, backfill_period_start); |
| for (auto obs_day_index = participation_first_day_index; obs_day_index <= final_day_index; |
| obs_day_index++) { |
| GenerateSingleReportParticipationObservation( |
| metric_ref, &report_aggregates.aggregation_config().report(), obs_day_index); |
| SetReportParticipationLastGeneratedDayIndex(report_key, obs_day_index); |
| } |
| return kOK; |
| } |
| |
| LocalAggregateStore AggregateStore::MakeNewLocalAggregateStore(uint32_t version) { |
| LocalAggregateStore store; |
| store.set_version(version); |
| return store; |
| } |
| |
| AggregatedObservationHistoryStore AggregateStore::MakeNewObservationHistoryStore(uint32_t version) { |
| AggregatedObservationHistoryStore store; |
| store.set_version(version); |
| return store; |
| } |
| |
| // We can upgrade from v0, but no other versions. |
| Status AggregateStore::MaybeUpgradeLocalAggregateStore(LocalAggregateStore* store) { |
| uint32_t version = store->version(); |
| if (version == kCurrentLocalAggregateStoreVersion) { |
| return kOK; |
| } |
| VLOG(4) << "Attempting to upgrade LocalAggregateStore from version " << version; |
| switch (version) { |
| case 0u: |
| return UpgradeLocalAggregateStoreFromVersion0(store); |
| default: |
| LOG(ERROR) << "Cannot upgrade LocalAggregateStore from version " << version; |
| return kInvalidArguments; |
| } |
| } |
| |
| // The current version is the earliest version, so no other versions are accepted. |
| Status AggregateStore::MaybeUpgradeObservationHistoryStore( |
| AggregatedObservationHistoryStore* store) { |
| uint32_t version = store->version(); |
| if (version == kCurrentObservationHistoryStoreVersion) { |
| return kOK; |
| } |
| LOG(ERROR) << "Cannot upgrade AggregatedObservationHistoryStore from version " << version; |
| return kInvalidArguments; |
| } |
| |
| } // namespace cobalt::local_aggregation |
