src/local_aggregation_1_1/aggregation_procedures/count_aggregation_procedure.cc

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

#include "src/local_aggregation_1_1/aggregation_procedures/count_aggregation_procedure.h"

#include <tuple>

#include "src/lib/util/datetime_util.h"
#include "src/local_aggregation_1_1/aggregation_procedures/aggregation_procedure.h"
#include "src/logger/encoder.h"
#include "src/pb/observation.pb.h"
#include "src/public/lib/statusor/status_macros.h"

namespace cobalt::local_aggregation {

bool CountAggregationProcedure::IsDaily() const {
  switch (report_type()) {
    case ReportDefinition::FLEETWIDE_OCCURRENCE_COUNTS:
    case ReportDefinition::HOURLY_VALUE_HISTOGRAMS:
    case ReportDefinition::HOURLY_VALUE_NUMERIC_STATS:
      return false;

    case ReportDefinition::UNIQUE_DEVICE_HISTOGRAMS:
    case ReportDefinition::UNIQUE_DEVICE_NUMERIC_STATS:
      return true;

    default:
      LOG(ERROR) << "Unexpected report_type for CountAggregationProcedure: " << report_type()
                 << ". Defaulting to IsDaily=true.";
      return true;
  }
}

void CountAggregationProcedure::UpdateAggregateData(const logger::EventRecord &event_record,
                                                    AggregateData *aggregate_data,
                                                    AggregationPeriodBucket * /*bucket*/) {
  uint64_t occurrence_count = event_record.event()->occurrence_event().count();
  aggregate_data->set_count(aggregate_data->count() + occurrence_count);
}

lib::statusor::StatusOr<std::unique_ptr<Observation>>
CountAggregationProcedure::GenerateSingleObservation(
    const std::vector<AggregateDataToGenerate> &buckets,
    const std::set<std::vector<uint32_t>> &event_vectors, const util::TimeInfo & /*time_info*/) {
  std::map<std::vector<uint32_t>, std::vector<const AggregateData *>> aggregates_by_event_code;
  for (const AggregateDataToGenerate &bucket : buckets) {
    for (const EventCodesAggregateData *aggregate_data : bucket.aggregate_data) {
      std::vector<uint32_t> event_vector(aggregate_data->event_codes().begin(),
                                         aggregate_data->event_codes().end());
      if (!event_vectors.count(event_vector)) {
        continue;
      }
      aggregates_by_event_code[event_vector].push_back(&aggregate_data->data());
    }
  }

  std::vector<std::tuple<std::vector<uint32_t>, int64_t>> data;
  data.reserve(aggregates_by_event_code.size());

  for (auto [event_codes, aggregates] : aggregates_by_event_code) {
    int64_t count = 0;
    for (const AggregateData *aggregate : aggregates) {
      count += aggregate->count();
    }
    data.emplace_back(std::make_tuple(event_codes, count));
  }

  if (data.empty()) {
    return {nullptr};
  }

  return logger::Encoder::EncodeIntegerObservation(data);
}

std::string CountAggregationProcedure::DebugString() const { return "COUNT"; }

}  // namespace cobalt::local_aggregation