src/local_aggregation_1_1/local_aggregate_storage/immediate_local_aggregate_storage.h

// 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.

#ifndef COBALT_SRC_LOCAL_AGGREGATION_1_1_LOCAL_AGGREGATE_STORAGE_IMMEDIATE_LOCAL_AGGREGATE_STORAGE_H_
#define COBALT_SRC_LOCAL_AGGREGATION_1_1_LOCAL_AGGREGATE_STORAGE_IMMEDIATE_LOCAL_AGGREGATE_STORAGE_H_

#include <cstdint>
#include <string>

#include "src/lib/util/consistent_proto_store.h"
#include "src/lib/util/file_system.h"
#include "src/local_aggregation_1_1/local_aggregate_storage/local_aggregate_storage.h"
#include "src/local_aggregation_1_1/local_aggregation.pb.h"
#include "src/logger/project_context_factory.h"
#include "src/registry/cobalt_registry.pb.h"

namespace cobalt::local_aggregation {

// ImmediateLocalAggregateStorage implements LocalAggregateStorage with immediate write to disk.
//
// This implementation uses a directory structure as follows:
//
// <customer_id>/
//   <project_id>/
//     <metric_id>
//     <metric_id>
//   <project_id>/
//     <metric_id>
// <customer_id>/
//   ...
//
// Each call to SaveMetricAggregate uses ConsistentProtoStore to overwrite the appropriate
// <metric_id> file.
//
// Note: This implementation fares poorly on systems with slow file system writes, and especially on
// those that suffer from write amplification.
class ImmediateLocalAggregateStorage : public LocalAggregateStorage {
 public:
  // Constructor for a ImmediateLocalAggregateStorage object
  //
  // |base_directory|: The absolute path to the directory where the local aggregation files are
  //                   stored. This directory doesn't need to exist yet, but its parent
  //                   directory must already exist.
  // |fs|: An instance of the FileSystem interface. Used for reading/writing files.
  // |global_project_context_factory|: The current global registry.
  ImmediateLocalAggregateStorage(
      std::string base_directory, util::FileSystem *fs,
      const logger::ProjectContextFactory *global_project_context_factory);

  ~ImmediateLocalAggregateStorage() override = default;
  lib::statusor::StatusOr<MetricAggregateRef> GetMetricAggregate(uint32_t customer_id,
                                                                 uint32_t project_id,
                                                                 uint32_t metric_id) override;

  void DeleteData() override;

 protected:
  util::Status SaveMetricAggregate(uint32_t customer_id, uint32_t project_id,
                                   uint32_t metric_id) override;

  void ResetInternalMetrics(logger::LoggerInterface *internal_logger) override {}

 private:
  // DeleteOutdatedMetrics walks the filesystem from the |base_directory_| down and deletes
  // MetricAggregate files, and project directories that do not exist in the CobaltRegistry.
  //
  // TODO(fxbug.dev/51390): Customers that are not present in the registry should be deleted
  // too.
  void DeleteOutdatedMetrics();

  // InitializePersistentStore iterates through the registry and creates the customer/project
  // directories for all of the metrics in the registry (if they don't already exist). Additionally,
  // it adds empty MetricAggregate objects to the |aggregates_| object. It does not create metric
  // files, since at this point they are guaranteed to be empty.
  void InitializePersistentStore();

  const std::string base_directory_;
  util::FileSystem *fs_;
  util::ConsistentProtoStore proto_store_;
  const logger::ProjectContextFactory *global_project_context_factory_;
  std::map<std::tuple<uint32_t, uint32_t, uint32_t>, MetricAggregate> aggregates_;
  std::mutex mutex_;
};

}  // namespace cobalt::local_aggregation

#endif  // COBALT_SRC_LOCAL_AGGREGATION_1_1_LOCAL_AGGREGATE_STORAGE_IMMEDIATE_LOCAL_AGGREGATE_STORAGE_H_