src/logger/internal_metrics.cc

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

#include <cstdint>
#include <memory>
#include <sstream>
#include <utility>

#include "src/logger/internal_metrics_config.cb.h"
#include "src/logging.h"
#include "src/public/lib/registry_identifiers.h"

namespace cobalt::logger {
namespace {

uint32_t GetProjectEventCode(const std::string& project_identifier,
                             const std::map<std::string, uint32_t>& project_map) {
  auto event_code = project_map.find(project_identifier);
  if (event_code != project_map.end()) {
    return event_code->second;
  }
  return 0;
}

uint32_t GetProjectEventCode(const std::string& project_identifier) {
  return GetProjectEventCode(project_identifier,
                             kPerProjectBytesStoredMigratedMetricDimensionProjectsNameMap);
}

}  // namespace

util::NotNullUniquePtr<InternalMetrics> InternalMetrics::NewWithLogger(
    LoggerInterface* logger, DiagnosticsInterface* diagnostics) {
  if (logger) {
    return util::MakeNotNullUniquePtr<InternalMetricsImpl>(*logger, diagnostics);
  }
  return util::MakeNotNullUniquePtr<NoOpInternalMetrics>();
}

InternalMetricsImpl::InternalMetricsImpl(LoggerInterface& logger, DiagnosticsInterface* diagnostics)
    : logger_(logger), diagnostics_(diagnostics) {}

void InternalMetricsImpl::LoggerCalled(LoggerCallsMadeMigratedMetricDimensionLoggerMethod method,
                                       const Project& project) {
  Queue([this, method, project] {
    auto status = logger_.LogOccurrence(kLoggerCallsMadeMigratedMetricId, 1, {method});

    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::LoggerCalled: LogOccurrence() returned status=" << status;
    }

    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::LoggerCalled: LogOccurrence() returned status=" << status;
    }

    if (diagnostics_ != nullptr) {
      std::ostringstream diagnostics_component;
      diagnostics_component << project.customer_name() << '/' << project.project_name();
      diagnostics_->LoggerCalled(method, diagnostics_component.str());
    }
  });
}

void InternalMetricsImpl::BytesUploaded(BytesUploadedMetricDimensionStatus upload_status,
                                        size_t byte_count) {
  Queue([this, upload_status, byte_count] {
    Status status = logger_.LogOccurrence(kBytesUploadedMetricId, static_cast<uint64_t>(byte_count),
                                          {upload_status});

    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::BytesUploaded: LogOccurrence() returned "
              << "status=" << status;
    }
  });
}

void InternalMetricsImpl::BytesUploaded(
    PerProjectBytesUploadedMigratedMetricDimensionStatus upload_status, size_t byte_count,
    const lib::ProjectIdentifier& project_identifier) {
  Queue([this, upload_status, byte_count, project_identifier] {
    std::ostringstream component;
    component << project_identifier.customer_id() << '/' << project_identifier.project_id();

    auto status = logger_.LogInteger(kPerProjectBytesUploadedMigratedMetricId,
                                     static_cast<int64_t>(byte_count),
                                     {GetProjectEventCode(component.str()), upload_status});

    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::BytesUploaded: LogInteger() returned "
              << "status=" << status;
    }
  });
}

void InternalMetricsImpl::BytesStored(
    PerProjectBytesStoredMigratedMetricDimensionStatus upload_status, size_t byte_count,
    const lib::ProjectIdentifier& project_identifier) {
  Queue([this, upload_status, byte_count, project_identifier] {
    std::ostringstream component;
    component << project_identifier.customer_id() << '/' << project_identifier.project_id();

    auto status =
        logger_.LogInteger(kPerProjectBytesStoredMigratedMetricId, static_cast<int64_t>(byte_count),
                           {GetProjectEventCode(component.str()), upload_status});

    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::BytesStored: LogInteger() returned status=" << status;
    }
  });
}

void InternalMetricsImpl::InaccurateClockEventsCached(int64_t event_count) {
  Queue([this, event_count] {
    auto status =
        logger_.LogOccurrence(kInaccurateClockEventsCachedMigratedMetricId, event_count, {});

    if (!status.ok()) {
      VLOG(1)
          << "InternalMetricsImpl::InaccurateClockEventsCached: LogOccurrence() returned status="
          << status;
    }
  });
}

void InternalMetricsImpl::InaccurateClockEventsDropped(int64_t event_count) {
  Queue([this, event_count] {
    auto status =
        logger_.LogOccurrence(kInaccurateClockEventsDroppedMigratedMetricId, event_count, {});

    if (!status.ok()) {
      VLOG(1)
          << "InternalMetricsImpl::InaccurateClockEventsDropped: LogOccurrence() returned status="
          << status;
    }
  });
}

void InternalMetricsImpl::SetSoftwareDistributionInfoCalled(
    SetSoftwareDistributionInfoCalledMigratedEventCodes event_codes) {
  Queue([this, event_codes] {
    auto status = logger_.LogOccurrence(kSetSoftwareDistributionInfoCalledMigratedMetricId, 1,
                                        event_codes.ToVector());

    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::SetSoftwareDistributionInfoCalled: LogOccurrence() returned "
                 "status="
              << status;
    }
  });
}

const float kPerMilleMultiplier = 1000.0;
void InternalMetricsImpl::TrackDiskUsage(StorageClass storage_class, size_t bytes,
                                         int64_t max_bytes) {
  Queue([this, storage_class, bytes, max_bytes] {
    // N.B. This method may only include Cobalt 1.1 metrics. Using Cobalt 1.0 metrics here have the
    // potential to cause logging loops.
    auto status =
        logger_.LogInteger(kStoreUsageMetricId, static_cast<int64_t>(bytes), {storage_class});
    if (!status.ok()) {
      VLOG(1) << "InternalMetricsImpl::TrackDiskUsage: LogInteger(disk_usage) returned status="
              << status;
    }

    if (max_bytes >= 0) {
      auto fullness_per_mille = static_cast<uint32_t>(
          (static_cast<float>(bytes) / static_cast<float>(max_bytes)) * kPerMilleMultiplier);
      status = logger_.LogInteger(kStoreFullnessMetricId, fullness_per_mille, {storage_class});
      if (!status.ok()) {
        VLOG(1) << "InternalMetricsImpl::TrackDiskUsage: LogInteger(disk_fullness) returned status="
                << status;
      }
    }

    if (diagnostics_ != nullptr) {
      diagnostics_->TrackDiskUsage(storage_class, static_cast<int64_t>(bytes), max_bytes);
    }
  });
}

void InternalMetricsImpl::LocalAggregationQuotaEvent(
    const lib::ProjectIdentifier& project_identifier, int event_type) {
  Queue([this, project_identifier, event_type] {
    if (diagnostics_ != nullptr) {
      diagnostics_->LocalAggregationQuotaEvent(project_identifier, event_type);
    }
  });
}

void InternalMetricsImpl::Queue(std::function<void()>&& lambda) {
  auto fields = fields_.lock();
  // In debug code we crash when a lambda is queued without having first created an
  // InternalMetricsFlusher. This should hopefully avoid introducing any instances of queue without
  // flush, but we don't want to crash in production in case we miss some cases.
  DCHECK(fields->flusher_count > 0)
      << "An internal metric was queued without a corresponding InternalMetricsFlusher. This can "
         "potentially cause memory leaks in production code.";
  fields->queued_lambdas.emplace_back(std::move(lambda));
}

void InternalMetricsImpl::IncrementFlushers() { fields_.lock()->flusher_count += 1; }

void InternalMetricsImpl::Flush() {
  std::vector<std::function<void()>> queued;
  {
    auto fields = fields_.lock();
    if (fields->flusher_count <= 0) {
      LOG_FIRST_N(ERROR, 10) << "Flush called without corresponding InternalMetricsFlusher. This "
                                "should not be possible.";
    }
    fields->flusher_count = std::max(fields->flusher_count - 1, 0);
    // Only perform the flush once the last flusher has been dropped.
    if (fields->flusher_count > 0) {
      return;
    }
    fields->queued_lambdas.swap(queued);
  }

  for (const auto& task : queued) {
    task();
  }
}

}  // namespace cobalt::logger