tlbmc/hft/manager_debug_json.cc

#include "tlbmc/hft/manager_debug_json.h"

#include <cstdint>
#include <memory>
#include <vector>

#include "absl/strings/str_cat.h"
#include "absl/strings/str_format.h"
#include "absl/time/time.h"
#include <nlohmann/json.hpp>
#include "tlbmc/hft/core/hft_service.h"
#include "tlbmc/hft/core/manager.h"
#include "tlbmc/hft/core/manager_fake.h"
#include "tlbmc/hft/core/manager_impl.h"
#include "tlbmc/hft/task_scheduler_adapter.h"

namespace milotic_hft {

nlohmann::json SubscriptionDebugJson(
    const SubscriptionManager::Subscription& subscription) {
  const auto* active = dynamic_cast<const ActiveSubscription*>(&subscription);
  if (active == nullptr) {
    return nlohmann::json::object();
  }
  nlohmann::json json;
  json["id"] = active->GetId();
  json["params"] = absl::StrCat(active->GetParams());
  json["batches_remaining"] = active->BatchesRemaining();
  json["data_source"] = active->GetDataSourceName();
  const auto& identifiers = active->GetParams().identifiers();
  const std::vector<int64_t> last_sampled_times_ns =
      active->GetLastSampledTimesNs();
  for (int i = 0; i < identifiers.size() &&
                  i < static_cast<int>(last_sampled_times_ns.size());
       ++i) {
    nlohmann::json entry;
    entry["identifier"] = absl::StrCat(identifiers[i]);
    entry["last_sampled_time"] =
        absl::FormatTime(absl::FromUnixNanos(last_sampled_times_ns[i]));
    json["identifier_to_last_sampled_time"].push_back(entry);
  }
  return json;
}

nlohmann::json SubscriptionManagerDebugJson(
    const SubscriptionManager& manager) {
  if (const auto* impl = dynamic_cast<const SubscriptionManagerImpl*>(&manager);
      impl != nullptr) {
    SubscriptionManagerImpl::DebugSnapshot snapshot = impl->GetDebugSnapshot();
    nlohmann::json json;
    json["next_subscription_id"] = snapshot.next_subscription_id;
    json["data_source"] = snapshot.data_source_name;
    if (const auto* adapter =
            dynamic_cast<const milotic_tlbmc::HftTaskSchedulerAdapter*>(
                &impl->GetScheduler());
        adapter != nullptr) {
      json["task_scheduler"] = adapter->task_scheduler().ToJson();
    }
    json["resource_monitors"] = nlohmann::json::array();
    for (const auto& monitor : snapshot.resource_monitors) {
      nlohmann::json monitor_json;
      monitor_json["identifier"] = absl::StrCat(monitor.identifier);
      monitor_json["current_sampling_interval_ms"] =
          monitor.current_sampling_interval_ms;
      monitor_json["current_max_batch_size"] = monitor.current_max_batch_size;
      monitor_json["num_subscribers"] = monitor.subscriber_ids.size();
      monitor_json["subscribers"] = monitor.subscriber_ids;
      monitor_json["lowest_sampling_interval_subscriptions"] =
          monitor.subscribed_sampling_intervals_ms;
      monitor_json["max_batch_sizes"] = monitor.subscribed_max_batch_sizes;
      json["resource_monitors"].push_back(monitor_json);
    }
    return json;
  }
  if (dynamic_cast<const SubscriptionManagerFake*>(&manager) != nullptr) {
    return nlohmann::json{{"name", "fake_manager"}};
  }
  return nlohmann::json::object();
}

namespace {

nlohmann::json ReactorDebugJson(const internal::ServerReactorImpl& reactor) {
  nlohmann::json json;
  json["reactor"] = absl::StrFormat("%p", &reactor);
  const std::vector<std::shared_ptr<SubscriptionManager::Subscription>>
      subscriptions = reactor.GetSubscriptionsForDebug();
  json["num_subscriptions"] = subscriptions.size();
  json["subscriptions"] = nlohmann::json::array();
  for (const auto& sub : subscriptions) {
    if (sub != nullptr) {
      json["subscriptions"].push_back(SubscriptionDebugJson(*sub));
    }
  }
  return json;
}

}  // namespace

nlohmann::json HftServiceDebugJson(const HftServiceImpl& service) {
  nlohmann::json json;
  json["subscription_manager"] =
      SubscriptionManagerDebugJson(service.GetSubscriptionManager());
  json["role_to_total_sample_rate"] = nlohmann::json::array();
  for (const auto& [role, total_sample_rate] :
       service.GetRoleSampleRatesForDebug()) {
    nlohmann::json entry;
    entry["role"] = role;
    entry["total_sample_rate"] = total_sample_rate;
    json["role_to_total_sample_rate"].push_back(entry);
  }
  json["reactors"] = nlohmann::json::array();
  service.ForEachReactorForDebug(
      [&json](const internal::ServerReactorImpl& reactor) {
        json["reactors"].push_back(ReactorDebugJson(reactor));
      });
  return json;
}

}  // namespace milotic_hft