| // 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 "logger/event_aggregator.h" |
| |
| #include <algorithm> |
| #include <string> |
| #include <utility> |
| |
| #include "algorithms/rappor/rappor_config_helper.h" |
| #include "config/metric_definition.pb.h" |
| #include "logger/project_context.h" |
| #include "util/datetime_util.h" |
| #include "util/proto_util.h" |
| #include "util/status.h" |
| |
| namespace cobalt { |
| |
| using rappor::RapporConfigHelper; |
| using util::ConsistentProtoStore; |
| using util::SerializeToBase64; |
| using util::StatusCode; |
| using util::SystemClock; |
| using util::TimeToDayIndex; |
| |
| namespace logger { |
| |
| EventAggregator::EventAggregator( |
| const Encoder* encoder, const ObservationWriter* observation_writer, |
| ConsistentProtoStore* local_aggregate_proto_store, |
| ConsistentProtoStore* obs_history_proto_store, const size_t backfill_days, |
| const std::chrono::seconds aggregate_backup_interval, |
| const std::chrono::seconds generate_obs_interval, |
| const std::chrono::seconds gc_interval) |
| : encoder_(encoder), |
| observation_writer_(observation_writer), |
| local_aggregate_proto_store_(local_aggregate_proto_store), |
| obs_history_proto_store_(obs_history_proto_store) { |
| CHECK_LE(aggregate_backup_interval.count(), generate_obs_interval.count()) |
| << "aggregate_backup_interval must be less than or equal to " |
| "generate_obs_interval"; |
| CHECK_LE(aggregate_backup_interval.count(), gc_interval.count()) |
| << "aggregate_backup_interval must be less than or equal to gc_interval"; |
| CHECK_LE(backfill_days, kEventAggregatorMaxAllowedBackfillDays) |
| << "backfill_days must be less than or equal to " |
| << kEventAggregatorMaxAllowedBackfillDays; |
| aggregate_backup_interval_ = aggregate_backup_interval; |
| generate_obs_interval_ = generate_obs_interval; |
| gc_interval_ = gc_interval; |
| backfill_days_ = backfill_days; |
| auto locked = protected_aggregate_store_.lock(); |
| auto restore_aggregates_status = |
| local_aggregate_proto_store_->Read(&(locked->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."; |
| 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->local_aggregate_store = LocalAggregateStore(); |
| } |
| } |
| auto restore_history_status = obs_history_proto_store_->Read(&obs_history_); |
| switch (restore_history_status.error_code()) { |
| case StatusCode::OK: { |
| VLOG(4) << "Read AggregatedObservationHistory from disk."; |
| break; |
| } |
| case StatusCode::NOT_FOUND: { |
| VLOG(4) |
| << "No file found for obs_history_proto_store. Proceeding " |
| "with empty AggregatedObservationHistory. File will be created on " |
| "first snapshot of the AggregatedObservationHistory."; |
| 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 AggregatedObservationHistory."; |
| obs_history_ = AggregatedObservationHistory(); |
| } |
| } |
| clock_.reset(new SystemClock()); |
| } |
| |
| void EventAggregator::Start() { |
| auto locked = protected_shutdown_flag_.lock(); |
| locked->shut_down = false; |
| std::thread t([this] { this->Run(); }); |
| worker_thread_ = std::move(t); |
| } |
| |
| // TODO(pesk): Have the config parser verify that each locally |
| // aggregated report has at least one window size and that all window |
| // sizes are <= |kMaxAllowedAggregationWindowSize|. Additionally, have |
| // this method filter out any window sizes larger than |
| // |kMaxAllowedAggregationWindowSize|. |
| Status EventAggregator::UpdateAggregationConfigs( |
| const ProjectContext& project_context) { |
| auto locked = protected_aggregate_store_.lock(); |
| std::string key; |
| ReportAggregationKey key_data; |
| key_data.set_customer_id(project_context.project().customer_id()); |
| key_data.set_project_id(project_context.project().project_id()); |
| for (const auto& metric : project_context.metric_definitions()->metric()) { |
| switch (metric.metric_type()) { |
| case MetricDefinition::EVENT_OCCURRED: { |
| key_data.set_metric_id(metric.id()); |
| for (const auto& report : metric.reports()) { |
| switch (report.report_type()) { |
| case ReportDefinition::UNIQUE_N_DAY_ACTIVES: { |
| key_data.set_report_id(report.id()); |
| if (!SerializeToBase64(key_data, &key)) { |
| return kInvalidArguments; |
| } |
| // TODO(pesk): update the EventAggregator's view of a Metric |
| // or ReportDefinition when appropriate. |
| if (locked->local_aggregate_store.aggregates().count(key) == 0) { |
| AggregationConfig aggregation_config; |
| *aggregation_config.mutable_project() = |
| project_context.project(); |
| *aggregation_config.mutable_metric() = |
| *project_context.GetMetric(metric.id()); |
| *aggregation_config.mutable_report() = report; |
| ReportAggregates report_aggregates; |
| *report_aggregates.mutable_aggregation_config() = |
| aggregation_config; |
| (*locked->local_aggregate_store.mutable_aggregates())[key] = |
| report_aggregates; |
| } |
| } |
| default: |
| continue; |
| } |
| } |
| } |
| default: |
| continue; |
| } |
| } |
| return kOK; |
| } |
| |
| Status EventAggregator::LogUniqueActivesEvent(uint32_t report_id, |
| EventRecord* event_record) { |
| if (!event_record->event->has_occurrence_event()) { |
| LOG(ERROR) << "EventAggregator::LogUniqueActivesEvent can only " |
| "accept OccurrenceEvents."; |
| return kInvalidArguments; |
| } |
| ReportAggregationKey key_data; |
| key_data.set_customer_id(event_record->metric->customer_id()); |
| key_data.set_project_id(event_record->metric->project_id()); |
| key_data.set_metric_id(event_record->metric->id()); |
| key_data.set_report_id(report_id); |
| std::string key; |
| if (!SerializeToBase64(key_data, &key)) { |
| return kInvalidArguments; |
| } |
| auto locked = protected_aggregate_store_.lock(); |
| auto aggregates = |
| locked->local_aggregate_store.mutable_aggregates()->find(key); |
| if (aggregates == locked->local_aggregate_store.mutable_aggregates()->end()) { |
| LOG(ERROR) << "The Local Aggregate Store received an unexpected key."; |
| return kInvalidArguments; |
| } |
| (*(*aggregates->second.mutable_by_event_code()) |
| [event_record->event->occurrence_event().event_code()] |
| .mutable_by_day_index())[event_record->event->day_index()] |
| .mutable_activity_daily_aggregate() |
| ->set_activity_indicator(true); |
| return kOK; |
| } |
| |
| Status EventAggregator::GenerateObservationsNoWorker( |
| uint32_t final_day_index_utc, uint32_t final_day_index_local) { |
| if (worker_thread_.joinable()) { |
| LOG(ERROR) << "GenerateObservationsNoWorker() was called while " |
| "worker thread was running."; |
| return kOther; |
| } |
| return GenerateObservations(final_day_index_utc, final_day_index_local); |
| } |
| |
| Status EventAggregator::BackUpLocalAggregateStore() { |
| // Lock, copy the LocalAggregateStore, and release the lock. Write the copy |
| // to |local_aggregate_proto_store_|. |
| auto local_aggregate_store = CopyLocalAggregateStore(); |
| 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 EventAggregator::BackUpObservationHistory() { |
| auto status = obs_history_proto_store_->Write(obs_history_); |
| if (!status.ok()) { |
| LOG(ERROR) << "Failed to back up the AggregatedObservationHistory. " |
| "::cobalt::util::Status error code: " |
| << status.error_code() |
| << "\nError message: " << status.error_message() |
| << "\nError details: " << status.error_details(); |
| return kOther; |
| } |
| return kOK; |
| } |
| |
| void EventAggregator::ShutDown() { |
| if (worker_thread_.joinable()) { |
| { |
| auto locked = protected_shutdown_flag_.lock(); |
| locked->shut_down = true; |
| locked->shutdown_notifier.notify_all(); |
| } |
| worker_thread_.join(); |
| } else { |
| protected_shutdown_flag_.lock()->shut_down = true; |
| } |
| } |
| |
| void EventAggregator::Run() { |
| auto current_time = clock_->now(); |
| // Schedule Observation generation to happen in the first cycle. |
| next_generate_obs_ = current_time; |
| // Schedule garbage collection to happen |gc_interval_| seconds from now. |
| next_gc_ = current_time + gc_interval_; |
| // Acquire the mutex protecting the shutdown flag and condition variable. |
| auto locked = protected_shutdown_flag_.lock(); |
| while (true) { |
| // If shutdown has been requested, back up the LocalAggregateStore and exit. |
| if (locked->shut_down) { |
| BackUpLocalAggregateStore(); |
| return; |
| } |
| // Sleep until the next scheduled backup of the LocalAggregateStore or until |
| // notified of shutdown. Back up the LocalAggregateStore after waking. |
| auto shutdown_requested = locked.wait_for_with( |
| &(locked->shutdown_notifier), aggregate_backup_interval_, |
| [&locked]() { return locked->shut_down; }); |
| BackUpLocalAggregateStore(); |
| // If the worker thread was woken up by a shutdown request, exit. Otherwise, |
| // complete any scheduled Observation generation and garbage collection. |
| if (shutdown_requested) { |
| return; |
| } |
| // Check whether it is time to generate Observations or to garbage-collect |
| // the LocalAggregate store. If so, do that task and schedule the next |
| // occurrence. |
| DoScheduledTasks(clock_->now()); |
| } |
| } |
| |
| void EventAggregator::DoScheduledTasks( |
| std::chrono::system_clock::time_point current_time) { |
| auto current_time_t = std::chrono::system_clock::to_time_t(current_time); |
| auto current_day_index_utc = |
| TimeToDayIndex(current_time_t, MetricDefinition::UTC); |
| auto current_day_index_local = |
| TimeToDayIndex(current_time_t, MetricDefinition::LOCAL); |
| if (current_time >= next_generate_obs_) { |
| auto obs_status = GenerateObservations(current_day_index_utc - 1, |
| current_day_index_local - 1); |
| if (obs_status == kOK) { |
| BackUpObservationHistory(); |
| } else { |
| LOG(ERROR) << "GenerateObservations failed with status: " << obs_status; |
| } |
| next_generate_obs_ += generate_obs_interval_; |
| } |
| if (current_time >= next_gc_) { |
| auto gc_status = |
| GarbageCollect(current_day_index_utc - 1, current_day_index_local - 1); |
| if (gc_status == kOK) { |
| BackUpLocalAggregateStore(); |
| } else { |
| LOG(ERROR) << "GarbageCollect failed with status: " << gc_status; |
| } |
| next_gc_ += gc_interval_; |
| } |
| } |
| |
| Status EventAggregator::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; |
| } |
| // Lock, copy the LocalAggregateStore, and release the lock. Use the copy to |
| // generate observations. |
| auto local_aggregate_store = CopyLocalAggregateStore(); |
| for (auto pair : local_aggregate_store.aggregates()) { |
| const auto& config = pair.second.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(ERROR) << "The TimeZonePolicy of this MetricDefinition is invalid."; |
| return kInvalidConfig; |
| } |
| |
| const auto& report = config.report(); |
| auto max_window_size = 0u; |
| for (uint32_t window_size : report.window_size()) { |
| if (window_size > kMaxAllowedAggregationWindowSize) { |
| LOG(WARNING) << "Window size exceeding " |
| "kMaxAllowedAggregationWindowSize will be " |
| "ignored by GenerateObservations"; |
| } else if (window_size > max_window_size) { |
| max_window_size = window_size; |
| } |
| } |
| if (max_window_size == 0) { |
| LOG(ERROR) << "Each locally aggregated report must specify a positive " |
| "window size."; |
| return kInvalidConfig; |
| } |
| if (final_day_index < max_window_size) { |
| LOG(ERROR) << "final_day_index must be >= max_window_size."; |
| return kInvalidArguments; |
| } |
| |
| 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, pair.first, pair.second, num_event_codes, |
| final_day_index); |
| if (status != kOK) { |
| return status; |
| } |
| } |
| default: |
| continue; |
| } |
| } |
| default: |
| continue; |
| } |
| } |
| return kOK; |
| } |
| |
| Status EventAggregator::GarbageCollect(uint32_t day_index_utc, |
| uint32_t day_index_local) { |
| if (day_index_local == 0u) { |
| day_index_local = day_index_utc; |
| } |
| auto locked = protected_aggregate_store_.lock(); |
| for (auto pair : locked->local_aggregate_store.aggregates()) { |
| uint32_t day_index; |
| switch (pair.second.aggregation_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(ERROR) << "The TimeZonePolicy of this MetricDefinition is invalid."; |
| return kInvalidConfig; |
| } |
| // Determine the largest window size in the report associated to this |
| // key-value pair. |
| uint32_t max_window_size = 1; |
| for (uint32_t window_size : |
| pair.second.aggregation_config().report().window_size()) { |
| if (window_size > max_window_size && |
| window_size <= kMaxAllowedAggregationWindowSize) { |
| max_window_size = window_size; |
| } |
| } |
| if (max_window_size == 0) { |
| LOG(ERROR) << "Each locally aggregated report must specify a positive " |
| "window size."; |
| return kInvalidConfig; |
| } |
| if (day_index < backfill_days_ + max_window_size) { |
| LOG(ERROR) << "day_index must be >= backfill_days_ + max_window_size."; |
| return kInvalidArguments; |
| } |
| // For each event code, iterate over the sub-map of local aggregates |
| // keyed by day index. Keep buckets with day indices greater than |
| // |day_index| - |backfill_days_| - |max_window_size|, and remove |
| // all buckets with smaller day indices. |
| for (auto event_code_aggregates : pair.second.by_event_code()) { |
| for (auto day_aggregates : event_code_aggregates.second.by_day_index()) { |
| if (day_aggregates.first <= |
| day_index - backfill_days_ - max_window_size) { |
| locked->local_aggregate_store.mutable_aggregates() |
| ->at(pair.first) |
| .mutable_by_event_code() |
| ->at(event_code_aggregates.first) |
| .mutable_by_day_index() |
| ->erase(day_aggregates.first); |
| } |
| } |
| // If the day index map under this event code is empty, remove the event |
| // code from the event code map under this ReportAggregationKey. |
| if (locked->local_aggregate_store.aggregates() |
| .at(pair.first) |
| .by_event_code() |
| .at(event_code_aggregates.first) |
| .by_day_index() |
| .empty()) { |
| locked->local_aggregate_store.mutable_aggregates() |
| ->at(pair.first) |
| .mutable_by_event_code() |
| ->erase(event_code_aggregates.first); |
| } |
| } |
| } |
| return kOK; |
| } |
| |
| ////////// GenerateUniqueActivesObservations and helper methods //////////////// |
| |
| // 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 window_size) { |
| for (uint32_t day_index = obs_day_index - window_size + 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() == |
| true) { |
| 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 window_size) { |
| return (active_day_index <= obs_day_index && |
| active_day_index > obs_day_index - window_size); |
| } |
| |
| uint32_t EventAggregator::LastGeneratedDayIndex(const std::string& report_key, |
| uint32_t event_code, |
| uint32_t window_size) const { |
| 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.by_event_code().find(event_code); |
| if (event_code_history == report_history->second.by_event_code().end()) { |
| return 0u; |
| } |
| auto window_size_history = |
| event_code_history->second.by_window_size().find(window_size); |
| if (window_size_history == |
| event_code_history->second.by_window_size().end()) { |
| return 0u; |
| } |
| return window_size_history->second; |
| } |
| |
| Status EventAggregator::GenerateSingleUniqueActivesObservation( |
| const MetricRef metric_ref, const ReportDefinition* report, |
| uint32_t obs_day_index, uint32_t event_code, uint32_t window_size, |
| bool was_active) const { |
| auto encoder_result = encoder_->EncodeUniqueActivesObservation( |
| metric_ref, report, obs_day_index, event_code, was_active, window_size); |
| 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 EventAggregator::GenerateUniqueActivesObservations( |
| const MetricRef metric_ref, const std::string& report_key, |
| const ReportAggregates& report_aggregates, uint32_t num_event_codes, |
| uint32_t final_day_index) { |
| for (uint32_t event_code = 0; event_code < num_event_codes; event_code++) { |
| auto daily_aggregates = report_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.by_event_code().end()); |
| for (uint32_t window_size : |
| report_aggregates.aggregation_config().report().window_size()) { |
| // Skip any window size larger than |
| // kMaxAllowedAggregationWindowSize. |
| if (window_size > kMaxAllowedAggregationWindowSize) { |
| 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 = |
| LastGeneratedDayIndex(report_key, event_code, window_size); |
| auto first_day_index = |
| std::max(last_gen + 1, uint32_t(final_day_index - backfill_days_)); |
| // 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 of size |window_size| 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_size)) { |
| was_active = true; |
| } else { |
| active_day_index = FirstActiveDayIndexInWindow( |
| daily_aggregates->second, obs_day_index, window_size); |
| was_active = IsActivityInWindow(active_day_index, obs_day_index, |
| window_size); |
| } |
| } |
| auto status = GenerateSingleUniqueActivesObservation( |
| metric_ref, &report_aggregates.aggregation_config().report(), |
| obs_day_index, event_code, window_size, was_active); |
| if (status != kOK) { |
| return status; |
| } |
| // Update |obs_history_| with the latest date of Observation |
| // generation for this report, event code, and window size. |
| (*(*(*obs_history_.mutable_by_report_key())[report_key] |
| .mutable_by_event_code())[event_code] |
| .mutable_by_window_size())[window_size] = obs_day_index; |
| } |
| } |
| } |
| return kOK; |
| } |
| |
| } // namespace logger |
| } // namespace cobalt |
