src/local_aggregation_1_1/aggregation_procedures/integer_histogram_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/integer_histogram_aggregation_procedure.h"

#include <tuple>

#include "src/lib/client/cpp/buckets_config.h"
#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/event.pb.h"
#include "src/pb/observation.pb.h"
#include "src/public/lib/statusor/status_macros.h"
#include "src/registry/metric_definition.pb.h"
#include "src/registry/report_definition.pb.h"

namespace cobalt::local_aggregation {

IntegerHistogramAggregationProcedure::IntegerHistogramAggregationProcedure(
    const MetricDefinition &metric, const ReportDefinition &report)
    : HourlyAggregationProcedure(metric, report),
      integer_buckets_(config::IntegerBucketConfig::CreateFromProto(
          (metric.has_int_buckets()) ? metric.int_buckets() : report.int_buckets())) {}

void IntegerHistogramAggregationProcedure::UpdateAggregateData(
    const logger::EventRecord &event_record, AggregateData *aggregate_data,
    AggregationPeriodBucket * /*bucket*/) {
  switch (metric_type()) {
    case MetricDefinition::INTEGER: {
      const IntegerEvent &int_event = event_record.event()->integer_event();

      ::google::protobuf::Map<uint32_t, int64_t> *histogram =
          aggregate_data->mutable_integer_histogram()->mutable_histogram();
      uint32_t index = integer_buckets_->BucketIndex(int_event.value());

      (*histogram)[index] += 1;
    } break;
    case MetricDefinition::INTEGER_HISTOGRAM: {
      const IntegerHistogramEvent &int_histogram_event =
          event_record.event()->integer_histogram_event();

      ::google::protobuf::Map<uint32_t, int64_t> *histogram =
          aggregate_data->mutable_integer_histogram()->mutable_histogram();
      for (const HistogramBucket &bucket : int_histogram_event.buckets()) {
        (*histogram)[bucket.index()] += static_cast<int64_t>(bucket.count());
      }
    } break;
    default:
      LOG(ERROR) << DebugString() << ": Saw unexpected type case "
                 << event_record.event()->type_case();
  }
}

lib::statusor::StatusOr<std::unique_ptr<Observation>>
IntegerHistogramAggregationProcedure::GenerateHourlyObservation(
    const AggregateDataToGenerate &bucket) {
  std::vector<std::tuple<std::vector<uint32_t>, std::vector<std::tuple<uint32_t, int64_t>>>> data;
  data.reserve(bucket.aggregate_data.size());

  for (const EventCodesAggregateData *aggregate_data : bucket.aggregate_data) {
    std::vector<std::tuple<uint32_t, int64_t>> histogram;
    for (const auto &[key, value] : aggregate_data->data().integer_histogram().histogram()) {
      histogram.emplace_back(std::make_tuple(key, value));
    }
    std::vector<uint32_t> event_codes(aggregate_data->event_codes().begin(),
                                      aggregate_data->event_codes().end());
    data.emplace_back(std::make_tuple(event_codes, histogram));
  }

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

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

std::string IntegerHistogramAggregationProcedure::DebugString() const {
  return "INTEGER_HISTOGRAM";
}

}  // namespace cobalt::local_aggregation