src/local_aggregation/event_aggregator.cc - cobalt - Git at Google

 // 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/event_aggregator.h"

 #include <algorithm>
 #include <map>
 #include <string>
 #include <utility>
 #include <vector>

 #include "src/lib/util/datetime_util.h"
 #include "src/lib/util/proto_util.h"
 #include "src/registry/packed_event_codes.h"

 namespace cobalt::local_aggregation {

 using logger::Encoder;
 using logger::EventRecord;
 using logger::kInvalidArguments;
 using logger::kOK;
 using logger::kOther;
 using logger::ObservationWriter;
 using logger::ProjectContext;
 using logger::Status;
 using util::ConsistentProtoStore;
 using util::SerializeToBase64;
 using util::SteadyClock;
 using util::TimeToDayIndex;

 namespace {

 // 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);
 }

 }  // namespace

 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) {
   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";
   aggregate_backup_interval_ = aggregate_backup_interval;
   generate_obs_interval_ = generate_obs_interval;
   gc_interval_ = gc_interval;

   aggregate_store_ =
       std::make_unique<AggregateStore>(encoder, observation_writer, local_aggregate_proto_store,
                                        obs_history_proto_store, backfill_days);

   steady_clock_ = std::make_unique<SteadyClock>();
 }

 void EventAggregator::Start(std::unique_ptr<util::SystemClockInterface> clock) {
   auto locked = protected_worker_thread_controller_.lock();
   locked->shut_down = false;
   std::thread t(std::bind(
       [this](std::unique_ptr<util::SystemClockInterface>& clock) { this->Run(std::move(clock)); },
       std::move(clock)));
   worker_thread_ = std::move(t);
 }

 // TODO(pesk): update the EventAggregator's view of a Metric
 // or ReportDefinition when appropriate.
 Status EventAggregator::UpdateAggregationConfigs(const ProjectContext& project_context) {
   auto locked = aggregate_store_->protected_aggregate_store_.lock();
   Status status;
   for (const auto& metric : project_context.metrics()) {
     switch (metric.metric_type()) {
       case MetricDefinition::EVENT_OCCURRED: {
         for (const auto& report : metric.reports()) {
           switch (report.report_type()) {
             case ReportDefinition::UNIQUE_N_DAY_ACTIVES: {
               status = aggregate_store_->MaybeInsertReportConfigLocked(
                   project_context, metric, report, &(locked->local_aggregate_store));
               if (status != kOK) {
                 return status;
               }
             }
             default:
               continue;
           }
         }
       }
       case MetricDefinition::EVENT_COUNT:
       case MetricDefinition::ELAPSED_TIME:
       case MetricDefinition::FRAME_RATE:
       case MetricDefinition::MEMORY_USAGE: {
         for (const auto& report : metric.reports()) {
           switch (report.report_type()) {
             case ReportDefinition::PER_DEVICE_NUMERIC_STATS:
             case ReportDefinition::PER_DEVICE_HISTOGRAM: {
               status = aggregate_store_->MaybeInsertReportConfigLocked(
                   project_context, metric, report, &(locked->local_aggregate_store));
               if (status != kOK) {
                 return status;
               }
             }
             default:
               continue;
           }
         }
       }
       default:
         continue;
     }
   }
   return kOK;
 }

 // Helper functions for the Log*Event() methods.
 namespace {

 // Checks that an Event has type |expected_event_type|.
 bool ValidateEventType(Event::TypeCase expected_event_type, const Event& event) {
   Event::TypeCase event_type = event.type_case();
   if (event_type != expected_event_type) {
     LOG(ERROR) << "Expected Event type is " << expected_event_type << "; found " << event_type
                << ".";
     return false;
   }
   return true;
 }

 }  // namespace

 Status EventAggregator::LogUniqueActivesEvent(uint32_t report_id, const EventRecord& event_record) {
   auto* event = event_record.event();
   if (!ValidateEventType(Event::kOccurrenceEvent, *event)) {
     return kInvalidArguments;
   }
   auto* metric = event_record.metric();
   std::string key;
   if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
                          &key)) {
     return kInvalidArguments;
   }
   auto locked = aggregate_store_->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->occurrence_event().event_code()]
         .mutable_by_day_index())[event->day_index()]
       .mutable_activity_daily_aggregate()
       ->set_activity_indicator(true);
   return kOK;
 }

 Status EventAggregator::LogCountEvent(uint32_t report_id, const EventRecord& event_record) {
   auto* event = event_record.event();
   if (!ValidateEventType(Event::kCountEvent, *event)) {
     return kInvalidArguments;
   }
   auto* metric = event_record.metric();
   std::string key;
   if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
                          &key)) {
     return kInvalidArguments;
   }
   const CountEvent& count_event = event->count_event();
   return LogNumericEvent(key, event->day_index(), count_event.component(),
                          config::PackEventCodes(count_event.event_code()), count_event.count());
 }

 Status EventAggregator::LogElapsedTimeEvent(uint32_t report_id, const EventRecord& event_record) {
   auto* event = event_record.event();
   if (!ValidateEventType(Event::kElapsedTimeEvent, *event)) {
     return kInvalidArguments;
   }
   std::string key;
   auto* metric = event_record.metric();
   if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
                          &key)) {
     return kInvalidArguments;
   }
   const ElapsedTimeEvent& elapsed_time_event = event->elapsed_time_event();
   return LogNumericEvent(key, event->day_index(), elapsed_time_event.component(),
                          config::PackEventCodes(elapsed_time_event.event_code()),
                          elapsed_time_event.elapsed_micros());
 }

 Status EventAggregator::LogFrameRateEvent(uint32_t report_id, const EventRecord& event_record) {
   auto* event = event_record.event();
   if (!ValidateEventType(Event::kFrameRateEvent, *event)) {
     return kInvalidArguments;
   }
   std::string key;
   auto* metric = event_record.metric();
   if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
                          &key)) {
     return kInvalidArguments;
   }
   const FrameRateEvent& frame_rate_event = event->frame_rate_event();
   return LogNumericEvent(key, event->day_index(), frame_rate_event.component(),
                          config::PackEventCodes(frame_rate_event.event_code()),
                          frame_rate_event.frames_per_1000_seconds());
 }

 Status EventAggregator::LogMemoryUsageEvent(uint32_t report_id, const EventRecord& event_record) {
   auto* event = event_record.event();
   if (!ValidateEventType(Event::kMemoryUsageEvent, *event)) {
     return kInvalidArguments;
   }
   std::string key;
   auto* metric = event_record.metric();
   if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
                          &key)) {
     return kInvalidArguments;
   }
   const MemoryUsageEvent& memory_usage_event = event->memory_usage_event();
   return LogNumericEvent(key, event->day_index(), memory_usage_event.component(),
                          config::PackEventCodes(memory_usage_event.event_code()),
                          memory_usage_event.bytes());
 }

 Status EventAggregator::LogNumericEvent(const std::string& report_key, uint32_t day_index,
                                         const std::string& component, uint64_t event_code,
                                         int64_t value) {
   auto locked = aggregate_store_->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 first_event_today = ((*aggregates_by_day).find(day_index) == aggregates_by_day->end());
   auto day_aggregate = (*aggregates_by_day)[day_index].mutable_numeric_daily_aggregate();
   const auto& aggregation_type =
       aggregates->second.aggregation_config().report().aggregation_type();
   switch (aggregation_type) {
     case ReportDefinition::SUM:
       day_aggregate->set_value(value + day_aggregate->value());
       return kOK;
     case ReportDefinition::MAX:
       day_aggregate->set_value(std::max(value, day_aggregate->value()));
       return kOK;
     case ReportDefinition::MIN:
       if (first_event_today) {
         day_aggregate->set_value(value);
       } else {
         day_aggregate->set_value(std::min(value, day_aggregate->value()));
       }
       return kOK;
     default:
       LOG(ERROR) << "Unexpected aggregation type " << aggregation_type;
       return kInvalidArguments;
   }
 }

 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 aggregate_store_->GenerateObservations(final_day_index_utc, final_day_index_local);
 }

 void EventAggregator::ShutDown() {
   if (worker_thread_.joinable()) {
     {
       auto locked = protected_worker_thread_controller_.lock();
       locked->shut_down = true;
       locked->shutdown_notifier.notify_all();
     }
     worker_thread_.join();
   } else {
     protected_worker_thread_controller_.lock()->shut_down = true;
   }
 }

 void EventAggregator::Run(std::unique_ptr<util::SystemClockInterface> system_clock) {
   std::chrono::steady_clock::time_point steady_time = steady_clock_->now();
   // Schedule Observation generation to happen in the first cycle.
   next_generate_obs_ = steady_time;
   // Schedule garbage collection to happen |gc_interval_| seconds from now.
   next_gc_ = steady_time + gc_interval_;
   // Acquire the mutex protecting the shutdown flag and condition variable.
   auto locked = protected_worker_thread_controller_.lock();
   while (true) {
     // If shutdown has been requested, back up the LocalAggregateStore and
     // exit.
     if (locked->shut_down) {
       aggregate_store_->BackUpLocalAggregateStore();
       return;
     }
     // Sleep until the next scheduled backup of the LocalAggregateStore or
     // until notified of shutdown. Back up the LocalAggregateStore after
     // waking.
     locked->shutdown_notifier.wait_for(locked, aggregate_backup_interval_, [&locked]() {
       if (locked->immediate_run_trigger) {
         locked->immediate_run_trigger = false;
         return true;
       }
       return locked->shut_down;
     });
     aggregate_store_->BackUpLocalAggregateStore();
     // If the worker thread was woken up by a shutdown request, exit.
     // Otherwise, complete any scheduled Observation generation and garbage
     // collection.
     if (locked->shut_down) {
       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(system_clock->now(), steady_clock_->now());
   }
 }

 void EventAggregator::DoScheduledTasks(std::chrono::system_clock::time_point system_time,
                                        std::chrono::steady_clock::time_point steady_time) {
   auto current_time_t = std::chrono::system_clock::to_time_t(system_time);
   auto yesterday_utc = TimeToDayIndex(current_time_t, MetricDefinition::UTC) - 1;
   auto yesterday_local_time = TimeToDayIndex(current_time_t, MetricDefinition::LOCAL) - 1;

   // Skip the tasks (but do schedule a retry) if either day index is too small.
   uint32_t min_allowed_day_index = kMaxAllowedAggregationDays + aggregate_store_->backfill_days_;
   bool skip_tasks =
       (yesterday_utc < min_allowed_day_index || yesterday_local_time < min_allowed_day_index);
   if (steady_time >= next_generate_obs_) {
     next_generate_obs_ += generate_obs_interval_;
     if (skip_tasks) {
       LOG_FIRST_N(ERROR, 10) << "EventAggregator is skipping Observation generation because the "
                                 "current day index is too small.";
     } else {
       auto obs_status = aggregate_store_->GenerateObservations(yesterday_utc, yesterday_local_time);
       if (obs_status == kOK) {
         aggregate_store_->BackUpObservationHistory();
       } else {
         LOG(ERROR) << "GenerateObservations failed with status: " << obs_status;
       }
     }
   }
   if (steady_time >= next_gc_) {
     next_gc_ += gc_interval_;
     if (skip_tasks) {
       LOG_FIRST_N(ERROR, 10) << "EventAggregator is skipping garbage collection because the "
                                 "current day index is too small.";
     } else {
       auto gc_status = aggregate_store_->GarbageCollect(yesterday_utc, yesterday_local_time);
       if (gc_status == kOK) {
         aggregate_store_->BackUpLocalAggregateStore();
       } else {
         LOG(ERROR) << "GarbageCollect failed with status: " << gc_status;
       }
     }
   }
 }

 }  // namespace cobalt::local_aggregation
	// 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/event_aggregator.h"

	#include <algorithm>
	#include <map>
	#include <string>
	#include <utility>
	#include <vector>

	#include "src/lib/util/datetime_util.h"
	#include "src/lib/util/proto_util.h"
	#include "src/registry/packed_event_codes.h"

	namespace cobalt::local_aggregation {

	using logger::Encoder;
	using logger::EventRecord;
	using logger::kInvalidArguments;
	using logger::kOK;
	using logger::kOther;
	using logger::ObservationWriter;
	using logger::ProjectContext;
	using logger::Status;
	using util::ConsistentProtoStore;
	using util::SerializeToBase64;
	using util::SteadyClock;
	using util::TimeToDayIndex;

	namespace {

	// 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);
	}

	} // namespace

	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) {
	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";
	aggregate_backup_interval_ = aggregate_backup_interval;
	generate_obs_interval_ = generate_obs_interval;
	gc_interval_ = gc_interval;

	aggregate_store_ =
	std::make_unique<AggregateStore>(encoder, observation_writer, local_aggregate_proto_store,
	obs_history_proto_store, backfill_days);

	steady_clock_ = std::make_unique<SteadyClock>();
	}

	void EventAggregator::Start(std::unique_ptr<util::SystemClockInterface> clock) {
	auto locked = protected_worker_thread_controller_.lock();
	locked->shut_down = false;
	std::thread t(std::bind(
	[this](std::unique_ptr<util::SystemClockInterface>& clock) { this->Run(std::move(clock)); },
	std::move(clock)));
	worker_thread_ = std::move(t);
	}

	// TODO(pesk): update the EventAggregator's view of a Metric
	// or ReportDefinition when appropriate.
	Status EventAggregator::UpdateAggregationConfigs(const ProjectContext& project_context) {
	auto locked = aggregate_store_->protected_aggregate_store_.lock();
	Status status;
	for (const auto& metric : project_context.metrics()) {
	switch (metric.metric_type()) {
	case MetricDefinition::EVENT_OCCURRED: {
	for (const auto& report : metric.reports()) {
	switch (report.report_type()) {
	case ReportDefinition::UNIQUE_N_DAY_ACTIVES: {
	status = aggregate_store_->MaybeInsertReportConfigLocked(
	project_context, metric, report, &(locked->local_aggregate_store));
	if (status != kOK) {
	return status;
	}
	}
	default:
	continue;
	}
	}
	}
	case MetricDefinition::EVENT_COUNT:
	case MetricDefinition::ELAPSED_TIME:
	case MetricDefinition::FRAME_RATE:
	case MetricDefinition::MEMORY_USAGE: {
	for (const auto& report : metric.reports()) {
	switch (report.report_type()) {
	case ReportDefinition::PER_DEVICE_NUMERIC_STATS:
	case ReportDefinition::PER_DEVICE_HISTOGRAM: {
	status = aggregate_store_->MaybeInsertReportConfigLocked(
	project_context, metric, report, &(locked->local_aggregate_store));
	if (status != kOK) {
	return status;
	}
	}
	default:
	continue;
	}
	}
	}
	default:
	continue;
	}
	}
	return kOK;
	}

	// Helper functions for the Log*Event() methods.
	namespace {

	// Checks that an Event has type \|expected_event_type\|.
	bool ValidateEventType(Event::TypeCase expected_event_type, const Event& event) {
	Event::TypeCase event_type = event.type_case();
	if (event_type != expected_event_type) {
	LOG(ERROR) << "Expected Event type is " << expected_event_type << "; found " << event_type
	<< ".";
	return false;
	}
	return true;
	}

	} // namespace

	Status EventAggregator::LogUniqueActivesEvent(uint32_t report_id, const EventRecord& event_record) {
	auto* event = event_record.event();
	if (!ValidateEventType(Event::kOccurrenceEvent, *event)) {
	return kInvalidArguments;
	}
	auto* metric = event_record.metric();
	std::string key;
	if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
	&key)) {
	return kInvalidArguments;
	}
	auto locked = aggregate_store_->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->occurrence_event().event_code()]
	.mutable_by_day_index())[event->day_index()]
	.mutable_activity_daily_aggregate()
	->set_activity_indicator(true);
	return kOK;
	}

	Status EventAggregator::LogCountEvent(uint32_t report_id, const EventRecord& event_record) {
	auto* event = event_record.event();
	if (!ValidateEventType(Event::kCountEvent, *event)) {
	return kInvalidArguments;
	}
	auto* metric = event_record.metric();
	std::string key;
	if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
	&key)) {
	return kInvalidArguments;
	}
	const CountEvent& count_event = event->count_event();
	return LogNumericEvent(key, event->day_index(), count_event.component(),
	config::PackEventCodes(count_event.event_code()), count_event.count());
	}

	Status EventAggregator::LogElapsedTimeEvent(uint32_t report_id, const EventRecord& event_record) {
	auto* event = event_record.event();
	if (!ValidateEventType(Event::kElapsedTimeEvent, *event)) {
	return kInvalidArguments;
	}
	std::string key;
	auto* metric = event_record.metric();
	if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
	&key)) {
	return kInvalidArguments;
	}
	const ElapsedTimeEvent& elapsed_time_event = event->elapsed_time_event();
	return LogNumericEvent(key, event->day_index(), elapsed_time_event.component(),
	config::PackEventCodes(elapsed_time_event.event_code()),
	elapsed_time_event.elapsed_micros());
	}

	Status EventAggregator::LogFrameRateEvent(uint32_t report_id, const EventRecord& event_record) {
	auto* event = event_record.event();
	if (!ValidateEventType(Event::kFrameRateEvent, *event)) {
	return kInvalidArguments;
	}
	std::string key;
	auto* metric = event_record.metric();
	if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
	&key)) {
	return kInvalidArguments;
	}
	const FrameRateEvent& frame_rate_event = event->frame_rate_event();
	return LogNumericEvent(key, event->day_index(), frame_rate_event.component(),
	config::PackEventCodes(frame_rate_event.event_code()),
	frame_rate_event.frames_per_1000_seconds());
	}

	Status EventAggregator::LogMemoryUsageEvent(uint32_t report_id, const EventRecord& event_record) {
	auto* event = event_record.event();
	if (!ValidateEventType(Event::kMemoryUsageEvent, *event)) {
	return kInvalidArguments;
	}
	std::string key;
	auto* metric = event_record.metric();
	if (!PopulateReportKey(metric->customer_id(), metric->project_id(), metric->id(), report_id,
	&key)) {
	return kInvalidArguments;
	}
	const MemoryUsageEvent& memory_usage_event = event->memory_usage_event();
	return LogNumericEvent(key, event->day_index(), memory_usage_event.component(),
	config::PackEventCodes(memory_usage_event.event_code()),
	memory_usage_event.bytes());
	}

	Status EventAggregator::LogNumericEvent(const std::string& report_key, uint32_t day_index,
	const std::string& component, uint64_t event_code,
	int64_t value) {
	auto locked = aggregate_store_->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 first_event_today = ((*aggregates_by_day).find(day_index) == aggregates_by_day->end());
	auto day_aggregate = (*aggregates_by_day)[day_index].mutable_numeric_daily_aggregate();
	const auto& aggregation_type =
	aggregates->second.aggregation_config().report().aggregation_type();
	switch (aggregation_type) {
	case ReportDefinition::SUM:
	day_aggregate->set_value(value + day_aggregate->value());
	return kOK;
	case ReportDefinition::MAX:
	day_aggregate->set_value(std::max(value, day_aggregate->value()));
	return kOK;
	case ReportDefinition::MIN:
	if (first_event_today) {
	day_aggregate->set_value(value);
	} else {
	day_aggregate->set_value(std::min(value, day_aggregate->value()));
	}
	return kOK;
	default:
	LOG(ERROR) << "Unexpected aggregation type " << aggregation_type;
	return kInvalidArguments;
	}
	}

	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 aggregate_store_->GenerateObservations(final_day_index_utc, final_day_index_local);
	}

	void EventAggregator::ShutDown() {
	if (worker_thread_.joinable()) {
	{
	auto locked = protected_worker_thread_controller_.lock();
	locked->shut_down = true;
	locked->shutdown_notifier.notify_all();
	}
	worker_thread_.join();
	} else {
	protected_worker_thread_controller_.lock()->shut_down = true;
	}
	}

	void EventAggregator::Run(std::unique_ptr<util::SystemClockInterface> system_clock) {
	std::chrono::steady_clock::time_point steady_time = steady_clock_->now();
	// Schedule Observation generation to happen in the first cycle.
	next_generate_obs_ = steady_time;
	// Schedule garbage collection to happen \|gc_interval_\| seconds from now.
	next_gc_ = steady_time + gc_interval_;
	// Acquire the mutex protecting the shutdown flag and condition variable.
	auto locked = protected_worker_thread_controller_.lock();
	while (true) {
	// If shutdown has been requested, back up the LocalAggregateStore and
	// exit.
	if (locked->shut_down) {
	aggregate_store_->BackUpLocalAggregateStore();
	return;
	}
	// Sleep until the next scheduled backup of the LocalAggregateStore or
	// until notified of shutdown. Back up the LocalAggregateStore after
	// waking.
	locked->shutdown_notifier.wait_for(locked, aggregate_backup_interval_, [&locked]() {
	if (locked->immediate_run_trigger) {
	locked->immediate_run_trigger = false;
	return true;
	}
	return locked->shut_down;
	});
	aggregate_store_->BackUpLocalAggregateStore();
	// If the worker thread was woken up by a shutdown request, exit.
	// Otherwise, complete any scheduled Observation generation and garbage
	// collection.
	if (locked->shut_down) {
	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(system_clock->now(), steady_clock_->now());
	}
	}

	void EventAggregator::DoScheduledTasks(std::chrono::system_clock::time_point system_time,
	std::chrono::steady_clock::time_point steady_time) {
	auto current_time_t = std::chrono::system_clock::to_time_t(system_time);
	auto yesterday_utc = TimeToDayIndex(current_time_t, MetricDefinition::UTC) - 1;
	auto yesterday_local_time = TimeToDayIndex(current_time_t, MetricDefinition::LOCAL) - 1;

	// Skip the tasks (but do schedule a retry) if either day index is too small.
	uint32_t min_allowed_day_index = kMaxAllowedAggregationDays + aggregate_store_->backfill_days_;
	bool skip_tasks =
	(yesterday_utc < min_allowed_day_index \|\| yesterday_local_time < min_allowed_day_index);
	if (steady_time >= next_generate_obs_) {
	next_generate_obs_ += generate_obs_interval_;
	if (skip_tasks) {
	LOG_FIRST_N(ERROR, 10) << "EventAggregator is skipping Observation generation because the "
	"current day index is too small.";
	} else {
	auto obs_status = aggregate_store_->GenerateObservations(yesterday_utc, yesterday_local_time);
	if (obs_status == kOK) {
	aggregate_store_->BackUpObservationHistory();
	} else {
	LOG(ERROR) << "GenerateObservations failed with status: " << obs_status;
	}
	}
	}
	if (steady_time >= next_gc_) {
	next_gc_ += gc_interval_;
	if (skip_tasks) {
	LOG_FIRST_N(ERROR, 10) << "EventAggregator is skipping garbage collection because the "
	"current day index is too small.";
	} else {
	auto gc_status = aggregate_store_->GarbageCollect(yesterday_utc, yesterday_local_time);
	if (gc_status == kOK) {
	aggregate_store_->BackUpLocalAggregateStore();
	} else {
	LOG(ERROR) << "GarbageCollect failed with status: " << gc_status;
	}
	}
	}
	}

	} // namespace cobalt::local_aggregation