/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include <grpc/impl/codegen/port_platform.h>

#include <stdint.h>
#include <stdio.h>
#include <chrono>
#include <ctime>
#include <iterator>

#include "src/cpp/server/load_reporter/constants.h"
#include "src/cpp/server/load_reporter/get_cpu_stats.h"
#include "src/cpp/server/load_reporter/load_reporter.h"

#include "opencensus/stats/internal/set_aggregation_window.h"

namespace grpc {
namespace load_reporter {

CpuStatsProvider::CpuStatsSample CpuStatsProviderDefaultImpl::GetCpuStats() {
  return GetCpuStatsImpl();
}

CensusViewProvider::CensusViewProvider()
    : tag_key_token_(::opencensus::stats::TagKey::Register(kTagKeyToken)),
      tag_key_host_(::opencensus::stats::TagKey::Register(kTagKeyHost)),
      tag_key_user_id_(::opencensus::stats::TagKey::Register(kTagKeyUserId)),
      tag_key_status_(::opencensus::stats::TagKey::Register(kTagKeyStatus)),
      tag_key_metric_name_(
          ::opencensus::stats::TagKey::Register(kTagKeyMetricName)) {
  // One view related to starting a call.
  auto vd_start_count =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewStartCount)
          .set_measure(kMeasureStartCount)
          .set_aggregation(::opencensus::stats::Aggregation::Sum())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .set_description(
              "Delta count of calls started broken down by <token, host, "
              "user_id>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_start_count);
  view_descriptor_map_.emplace(kViewStartCount, vd_start_count);
  // Four views related to ending a call.
  // If this view is set as Count of kMeasureEndBytesSent (in hope of saving one
  // measure), it's infeasible to prepare fake data for testing. That's because
  // the OpenCensus API to make up view data will add the input data as separate
  // measurements instead of setting the data values directly.
  auto vd_end_count =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewEndCount)
          .set_measure(kMeasureEndCount)
          .set_aggregation(::opencensus::stats::Aggregation::Sum())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .add_column(tag_key_status_)
          .set_description(
              "Delta count of calls ended broken down by <token, host, "
              "user_id, status>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_end_count);
  view_descriptor_map_.emplace(kViewEndCount, vd_end_count);
  auto vd_end_bytes_sent =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewEndBytesSent)
          .set_measure(kMeasureEndBytesSent)
          .set_aggregation(::opencensus::stats::Aggregation::Sum())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .add_column(tag_key_status_)
          .set_description(
              "Delta sum of bytes sent broken down by <token, host, user_id, "
              "status>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_end_bytes_sent);
  view_descriptor_map_.emplace(kViewEndBytesSent, vd_end_bytes_sent);
  auto vd_end_bytes_received =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewEndBytesReceived)
          .set_measure(kMeasureEndBytesReceived)
          .set_aggregation(::opencensus::stats::Aggregation::Sum())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .add_column(tag_key_status_)
          .set_description(
              "Delta sum of bytes received broken down by <token, host, "
              "user_id, status>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_end_bytes_received);
  view_descriptor_map_.emplace(kViewEndBytesReceived, vd_end_bytes_received);
  auto vd_end_latency_ms =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewEndLatencyMs)
          .set_measure(kMeasureEndLatencyMs)
          .set_aggregation(::opencensus::stats::Aggregation::Sum())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .add_column(tag_key_status_)
          .set_description(
              "Delta sum of latency in ms broken down by <token, host, "
              "user_id, status>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_end_latency_ms);
  view_descriptor_map_.emplace(kViewEndLatencyMs, vd_end_latency_ms);
  // Two views related to other call metrics.
  auto vd_metric_call_count =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewOtherCallMetricCount)
          .set_measure(kMeasureOtherCallMetric)
          .set_aggregation(::opencensus::stats::Aggregation::Count())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .add_column(tag_key_metric_name_)
          .set_description(
              "Delta count of calls broken down by <token, host, user_id, "
              "metric_name>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_metric_call_count);
  view_descriptor_map_.emplace(kViewOtherCallMetricCount, vd_metric_call_count);
  auto vd_metric_value =
      ::opencensus::stats::ViewDescriptor()
          .set_name(kViewOtherCallMetricValue)
          .set_measure(kMeasureOtherCallMetric)
          .set_aggregation(::opencensus::stats::Aggregation::Sum())
          .add_column(tag_key_token_)
          .add_column(tag_key_host_)
          .add_column(tag_key_user_id_)
          .add_column(tag_key_metric_name_)
          .set_description(
              "Delta sum of call metric value broken down "
              "by <token, host, user_id, metric_name>.");
  ::opencensus::stats::SetAggregationWindow(
      ::opencensus::stats::AggregationWindow::Delta(), &vd_metric_value);
  view_descriptor_map_.emplace(kViewOtherCallMetricValue, vd_metric_value);
}

double CensusViewProvider::GetRelatedViewDataRowDouble(
    const ViewDataMap& view_data_map, const char* view_name,
    size_t view_name_len, const std::vector<grpc::string>& tag_values) {
  auto it_vd = view_data_map.find(grpc::string(view_name, view_name_len));
  GPR_ASSERT(it_vd != view_data_map.end());
  GPR_ASSERT(it_vd->second.type() ==
             ::opencensus::stats::ViewData::Type::kDouble);
  auto it_row = it_vd->second.double_data().find(tag_values);
  GPR_ASSERT(it_row != it_vd->second.double_data().end());
  return it_row->second;
}

uint64_t CensusViewProvider::GetRelatedViewDataRowInt(
    const ViewDataMap& view_data_map, const char* view_name,
    size_t view_name_len, const std::vector<grpc::string>& tag_values) {
  auto it_vd = view_data_map.find(grpc::string(view_name, view_name_len));
  GPR_ASSERT(it_vd != view_data_map.end());
  GPR_ASSERT(it_vd->second.type() ==
             ::opencensus::stats::ViewData::Type::kInt64);
  auto it_row = it_vd->second.int_data().find(tag_values);
  GPR_ASSERT(it_row != it_vd->second.int_data().end());
  GPR_ASSERT(it_row->second >= 0);
  return it_row->second;
}

CensusViewProviderDefaultImpl::CensusViewProviderDefaultImpl() {
  for (const auto& p : view_descriptor_map()) {
    const grpc::string& view_name = p.first;
    const ::opencensus::stats::ViewDescriptor& vd = p.second;
    // We need to use pair's piecewise ctor here, otherwise the deleted copy
    // ctor of View will be called.
    view_map_.emplace(std::piecewise_construct,
                      std::forward_as_tuple(view_name),
                      std::forward_as_tuple(vd));
  }
}

CensusViewProvider::ViewDataMap CensusViewProviderDefaultImpl::FetchViewData() {
  gpr_log(GPR_DEBUG, "[CVP %p] Starts fetching Census view data.", this);
  ViewDataMap view_data_map;
  for (auto& p : view_map_) {
    const grpc::string& view_name = p.first;
    ::opencensus::stats::View& view = p.second;
    if (view.IsValid()) {
      view_data_map.emplace(view_name, view.GetData());
      gpr_log(GPR_DEBUG, "[CVP %p] Fetched view data (view: %s).", this,
              view_name.c_str());
    } else {
      gpr_log(
          GPR_DEBUG,
          "[CVP %p] Can't fetch view data because view is invalid (view: %s).",
          this, view_name.c_str());
    }
  }
  return view_data_map;
}

grpc::string LoadReporter::GenerateLbId() {
  while (true) {
    if (next_lb_id_ > UINT32_MAX) {
      gpr_log(GPR_ERROR, "[LR %p] The LB ID exceeds the max valid value!",
              this);
      return "";
    }
    int64_t lb_id = next_lb_id_++;
    // Overflow should never happen.
    GPR_ASSERT(lb_id >= 0);
    // Convert to padded hex string for a 32-bit LB ID. E.g, "0000ca5b".
    char buf[kLbIdLength + 1];
    snprintf(buf, sizeof(buf), "%08lx", lb_id);
    grpc::string lb_id_str(buf, kLbIdLength);
    // The client may send requests with LB ID that has never been allocated
    // by this load reporter. Those IDs are tracked and will be skipped when
    // we generate a new ID.
    if (!load_data_store_.IsTrackedUnknownBalancerId(lb_id_str)) {
      return lb_id_str;
    }
  }
}

::grpc::lb::v1::LoadBalancingFeedback
LoadReporter::GenerateLoadBalancingFeedback() {
  std::unique_lock<std::mutex> lock(feedback_mu_);
  auto now = std::chrono::system_clock::now();
  // Discard records outside the window until there is only one record
  // outside the window, which is used as the base for difference.
  while (feedback_records_.size() > 1 &&
         !IsRecordInWindow(feedback_records_[1], now)) {
    feedback_records_.pop_front();
  }
  if (feedback_records_.size() < 2) {
    return ::grpc::lb::v1::LoadBalancingFeedback::default_instance();
  }
  // Find the longest range with valid ends.
  auto oldest = feedback_records_.begin();
  auto newest = feedback_records_.end() - 1;
  while (std::distance(oldest, newest) > 0 &&
         (newest->cpu_limit == 0 || oldest->cpu_limit == 0)) {
    // A zero limit means that the system info reading was failed, so these
    // records can't be used to calculate CPU utilization.
    if (newest->cpu_limit == 0) --newest;
    if (oldest->cpu_limit == 0) ++oldest;
  }
  if (std::distance(oldest, newest) < 1 ||
      oldest->end_time == newest->end_time ||
      newest->cpu_limit == oldest->cpu_limit) {
    return ::grpc::lb::v1::LoadBalancingFeedback::default_instance();
  }
  uint64_t rpcs = 0;
  uint64_t errors = 0;
  for (auto p = newest; p != oldest; --p) {
    // Because these two numbers are counters, the oldest record shouldn't be
    // included.
    rpcs += p->rpcs;
    errors += p->errors;
  }
  double cpu_usage = newest->cpu_usage - oldest->cpu_usage;
  double cpu_limit = newest->cpu_limit - oldest->cpu_limit;
  std::chrono::duration<double> duration_seconds =
      newest->end_time - oldest->end_time;
  lock.unlock();
  ::grpc::lb::v1::LoadBalancingFeedback feedback;
  feedback.set_server_utilization(static_cast<float>(cpu_usage / cpu_limit));
  feedback.set_calls_per_second(
      static_cast<float>(rpcs / duration_seconds.count()));
  feedback.set_errors_per_second(
      static_cast<float>(errors / duration_seconds.count()));
  return feedback;
}

::google::protobuf::RepeatedPtrField<::grpc::lb::v1::Load>
LoadReporter::GenerateLoads(const grpc::string& hostname,
                            const grpc::string& lb_id) {
  std::lock_guard<std::mutex> lock(store_mu_);
  auto assigned_stores = load_data_store_.GetAssignedStores(hostname, lb_id);
  GPR_ASSERT(assigned_stores != nullptr);
  GPR_ASSERT(!assigned_stores->empty());
  ::google::protobuf::RepeatedPtrField<::grpc::lb::v1::Load> loads;
  for (PerBalancerStore* per_balancer_store : *assigned_stores) {
    GPR_ASSERT(!per_balancer_store->IsSuspended());
    if (!per_balancer_store->load_record_map().empty()) {
      for (const auto& p : per_balancer_store->load_record_map()) {
        const auto& key = p.first;
        const auto& value = p.second;
        auto load = loads.Add();
        load->set_load_balance_tag(key.lb_tag());
        load->set_user_id(key.user_id());
        load->set_client_ip_address(key.GetClientIpBytes());
        load->set_num_calls_started(static_cast<int64_t>(value.start_count()));
        load->set_num_calls_finished_without_error(
            static_cast<int64_t>(value.ok_count()));
        load->set_num_calls_finished_with_error(
            static_cast<int64_t>(value.error_count()));
        load->set_total_bytes_sent(static_cast<int64_t>(value.bytes_sent()));
        load->set_total_bytes_received(
            static_cast<int64_t>(value.bytes_recv()));
        load->mutable_total_latency()->set_seconds(
            static_cast<int64_t>(value.latency_ms() / 1000));
        load->mutable_total_latency()->set_nanos(
            (static_cast<int32_t>(value.latency_ms()) % 1000) * 1000000);
        for (const auto& p : value.call_metrics()) {
          const grpc::string& metric_name = p.first;
          const CallMetricValue& metric_value = p.second;
          auto call_metric_data = load->add_metric_data();
          call_metric_data->set_metric_name(metric_name);
          call_metric_data->set_num_calls_finished_with_metric(
              metric_value.num_calls());
          call_metric_data->set_total_metric_value(
              metric_value.total_metric_value());
        }
        if (per_balancer_store->lb_id() != lb_id) {
          // This per-balancer store is an orphan assigned to this receiving
          // balancer.
          AttachOrphanLoadId(load, *per_balancer_store);
        }
      }
      per_balancer_store->ClearLoadRecordMap();
    }
    if (per_balancer_store->IsNumCallsInProgressChangedSinceLastReport()) {
      auto load = loads.Add();
      load->set_num_calls_in_progress(
          per_balancer_store->GetNumCallsInProgressForReport());
      if (per_balancer_store->lb_id() != lb_id) {
        // This per-balancer store is an orphan assigned to this receiving
        // balancer.
        AttachOrphanLoadId(load, *per_balancer_store);
      }
    }
  }
  return loads;
}

void LoadReporter::AttachOrphanLoadId(
    ::grpc::lb::v1::Load* load, const PerBalancerStore& per_balancer_store) {
  if (per_balancer_store.lb_id() == kInvalidLbId) {
    load->set_load_key_unknown(true);
  } else {
    // We shouldn't set load_key_unknown to any value in this case because
    // load_key_unknown and orphaned_load_identifier are under an oneof struct.
    load->mutable_orphaned_load_identifier()->set_load_key(
        per_balancer_store.load_key());
    load->mutable_orphaned_load_identifier()->set_load_balancer_id(
        per_balancer_store.lb_id());
  }
}

void LoadReporter::AppendNewFeedbackRecord(uint64_t rpcs, uint64_t errors) {
  CpuStatsProvider::CpuStatsSample cpu_stats;
  if (cpu_stats_provider_ != nullptr) {
    cpu_stats = cpu_stats_provider_->GetCpuStats();
  } else {
    // This will make the load balancing feedback generation a no-op.
    cpu_stats = {0, 0};
  }
  std::unique_lock<std::mutex> lock(feedback_mu_);
  feedback_records_.emplace_back(std::chrono::system_clock::now(), rpcs, errors,
                                 cpu_stats.first, cpu_stats.second);
}

void LoadReporter::ReportStreamCreated(const grpc::string& hostname,
                                       const grpc::string& lb_id,
                                       const grpc::string& load_key) {
  std::lock_guard<std::mutex> lock(store_mu_);
  load_data_store_.ReportStreamCreated(hostname, lb_id, load_key);
  gpr_log(GPR_INFO,
          "[LR %p] Report stream created (host: %s, LB ID: %s, load key: %s).",
          this, hostname.c_str(), lb_id.c_str(), load_key.c_str());
}

void LoadReporter::ReportStreamClosed(const grpc::string& hostname,
                                      const grpc::string& lb_id) {
  std::lock_guard<std::mutex> lock(store_mu_);
  load_data_store_.ReportStreamClosed(hostname, lb_id);
  gpr_log(GPR_INFO, "[LR %p] Report stream closed (host: %s, LB ID: %s).", this,
          hostname.c_str(), lb_id.c_str());
}

void LoadReporter::ProcessViewDataCallStart(
    const CensusViewProvider::ViewDataMap& view_data_map) {
  auto it = view_data_map.find(kViewStartCount);
  if (it != view_data_map.end()) {
    for (const auto& p : it->second.int_data()) {
      const std::vector<grpc::string>& tag_values = p.first;
      const uint64_t start_count = static_cast<uint64_t>(p.second);
      const grpc::string& client_ip_and_token = tag_values[0];
      const grpc::string& host = tag_values[1];
      const grpc::string& user_id = tag_values[2];
      LoadRecordKey key(client_ip_and_token, user_id);
      LoadRecordValue value = LoadRecordValue(start_count);
      {
        std::unique_lock<std::mutex> lock(store_mu_);
        load_data_store_.MergeRow(host, key, value);
      }
    }
  }
}

void LoadReporter::ProcessViewDataCallEnd(
    const CensusViewProvider::ViewDataMap& view_data_map) {
  uint64_t total_end_count = 0;
  uint64_t total_error_count = 0;
  auto it = view_data_map.find(kViewEndCount);
  if (it != view_data_map.end()) {
    for (const auto& p : it->second.int_data()) {
      const std::vector<grpc::string>& tag_values = p.first;
      const uint64_t end_count = static_cast<uint64_t>(p.second);
      const grpc::string& client_ip_and_token = tag_values[0];
      const grpc::string& host = tag_values[1];
      const grpc::string& user_id = tag_values[2];
      const grpc::string& status = tag_values[3];
      // This is due to a bug reported internally of Java server load reporting
      // implementation.
      // TODO(juanlishen): Check whether this situation happens in OSS C++.
      if (client_ip_and_token.size() == 0) {
        gpr_log(GPR_DEBUG,
                "Skipping processing Opencensus record with empty "
                "client_ip_and_token tag.");
        continue;
      }
      LoadRecordKey key(client_ip_and_token, user_id);
      const uint64_t bytes_sent = CensusViewProvider::GetRelatedViewDataRowInt(
          view_data_map, kViewEndBytesSent, sizeof(kViewEndBytesSent) - 1,
          tag_values);
      const uint64_t bytes_received =
          CensusViewProvider::GetRelatedViewDataRowInt(
              view_data_map, kViewEndBytesReceived,
              sizeof(kViewEndBytesReceived) - 1, tag_values);
      const uint64_t latency_ms = CensusViewProvider::GetRelatedViewDataRowInt(
          view_data_map, kViewEndLatencyMs, sizeof(kViewEndLatencyMs) - 1,
          tag_values);
      uint64_t ok_count = 0;
      uint64_t error_count = 0;
      total_end_count += end_count;
      if (std::strcmp(status.c_str(), kCallStatusOk) == 0) {
        ok_count = end_count;
      } else {
        error_count = end_count;
        total_error_count += end_count;
      }
      LoadRecordValue value = LoadRecordValue(
          0, ok_count, error_count, bytes_sent, bytes_received, latency_ms);
      {
        std::unique_lock<std::mutex> lock(store_mu_);
        load_data_store_.MergeRow(host, key, value);
      }
    }
  }
  AppendNewFeedbackRecord(total_end_count, total_error_count);
}

void LoadReporter::ProcessViewDataOtherCallMetrics(
    const CensusViewProvider::ViewDataMap& view_data_map) {
  auto it = view_data_map.find(kViewOtherCallMetricCount);
  if (it != view_data_map.end()) {
    for (const auto& p : it->second.int_data()) {
      const std::vector<grpc::string>& tag_values = p.first;
      const int64_t num_calls = p.second;
      const grpc::string& client_ip_and_token = tag_values[0];
      const grpc::string& host = tag_values[1];
      const grpc::string& user_id = tag_values[2];
      const grpc::string& metric_name = tag_values[3];
      LoadRecordKey key(client_ip_and_token, user_id);
      const double total_metric_value =
          CensusViewProvider::GetRelatedViewDataRowDouble(
              view_data_map, kViewOtherCallMetricValue,
              sizeof(kViewOtherCallMetricValue) - 1, tag_values);
      LoadRecordValue value = LoadRecordValue(
          metric_name, static_cast<uint64_t>(num_calls), total_metric_value);
      {
        std::unique_lock<std::mutex> lock(store_mu_);
        load_data_store_.MergeRow(host, key, value);
      }
    }
  }
}

void LoadReporter::FetchAndSample() {
  gpr_log(GPR_DEBUG,
          "[LR %p] Starts fetching Census view data and sampling LB feedback "
          "record.",
          this);
  CensusViewProvider::ViewDataMap view_data_map =
      census_view_provider_->FetchViewData();
  ProcessViewDataCallStart(view_data_map);
  ProcessViewDataCallEnd(view_data_map);
  ProcessViewDataOtherCallMetrics(view_data_map);
}

}  // namespace load_reporter
}  // namespace grpc