tlbmc/metrics/software_metrics.cc

#include "tlbmc/metrics/software_metrics.h"

#include <dirent.h>
#include <sys/stat.h>

#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <filesystem>  // NOLINT: filesystem is commonly used in bmc codebase
#include <fstream>
#include <memory>
#include <sstream>
#include <string>
#include <system_error>  // NOLINT: system_error is commonly used in bmc
#include <utility>
#include <vector>

#include "google/protobuf/duration.pb.h"
#include "absl/container/flat_hash_map.h"
#include "absl/container/flat_hash_set.h"
#include "absl/functional/any_invocable.h"
#include "absl/log/log.h"
#include "absl/memory/memory.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/ascii.h"
#include "absl/strings/numbers.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/strings/strip.h"
#include "absl/synchronization/mutex.h"
#include "g3/macros.h"
#include <nlohmann/json.hpp>
#include "tlbmc/central_config/config.h"
#include "resource.pb.h"
#include "software_metrics.pb.h"
#include "tlbmc/time/time.h"
#include "tlbmc/utils/command_executor.h"
namespace milotic_tlbmc {

namespace {

// Helper to unescape systemd unit names from PSI directory paths.
// Systemd escapes special characters as \xNN.
// absl::CUnescape is unsuitable here because it is "greedy" and consumes more
// than 2 hex digits if available. For example, in a path like "\x2dconsole",
// absl::CUnescape tries to parse "2dc" as a single hex value (0x2DC > 255),
// causing an error. This function strictly parses exactly 2 hex digits after
// "\x".
std::string SystemdUnescape(absl::string_view filename) {
  std::string unescaped;
  unescaped.reserve(filename.size());
  for (size_t i = 0; i < filename.size(); ++i) {
    if (filename[i] == '\\' && i + 3 < filename.size() &&
        filename[i + 1] == 'x') {
      int32_t v = 0;
      if (absl::SimpleHexAtoi(filename.substr(i + 2, 2), &v)) {
        unescaped.push_back(static_cast<char>(v));
        i += 3;
        continue;
      }
    }
    unescaped.push_back(filename[i]);
  }
  return unescaped;
}

constexpr std::string_view kBiosKeyVersionPath =
    "run/bios_key/bios_key_version";
constexpr std::string_view kBiosKeyImageFamilyPath =
    "run/bios_key/bios_key_image_family";
constexpr std::string_view kBiosKeyValidationMethodPath =
    "run/bios_key/bios_key_validation_method";
constexpr std::string_view kBiosKeyValidationKeyDataPath =
    "run/bios_key/bios_key_validation_key_data";

}  // namespace

SoftwareMetricsAttributesStatic SoftwareMetrics::CreateStaticAttributes(
    const SoftwareMetricsConfig& software_metrics_config) {
  SoftwareMetricsAttributesStatic metric_attributes_static;
  *metric_attributes_static.mutable_software_metrics_config() =
      software_metrics_config;
  return metric_attributes_static;
}

absl::StatusOr<SocketStatStates> SoftwareMetrics::UpdateSocketStats() {
  ECCLESIA_ASSIGN_OR_RETURN(
      std::string cmd_output,
      executor_->Execute(std::string(kSocketStatCmd), false));

  SocketStatStates socket_states;
  *socket_states.mutable_timestamp() = Now();

  std::stringstream ss(cmd_output);
  std::string line;
  while (std::getline(ss, line)) {
    if (line.empty()) {
      continue;
    }

    std::stringstream line_stream(line);
    std::string state_str;
    int port_num;
    SocketStatState* state;
    if (line_stream >> state_str >> port_num) {
      if (ss_port_set_.contains(port_num)) {
        state = &(*socket_states.mutable_states())[port_num];
        if (state_str == "LISTEN") {
          state->set_listen(state->listen() + 1);
        } else if (state_str == "ESTAB") {
          state->set_established(state->established() + 1);
        } else if (state_str == "SYN-SENT") {
          state->set_syn_sent(state->syn_sent() + 1);
        } else if (state_str == "SYN-RECV") {
          state->set_syn_received(state->syn_received() + 1);
        } else if (state_str == "FIN-WAIT-1") {
          state->set_fin_wait_one(state->fin_wait_one() + 1);
        } else if (state_str == "FIN-WAIT-2") {
          state->set_fin_wait_two(state->fin_wait_two() + 1);
        } else if (state_str == "TIME-WAIT") {
          state->set_time_wait(state->time_wait() + 1);
        } else if (state_str == "CLOSE-WAIT") {
          state->set_close_wait(state->close_wait() + 1);
        } else if (state_str == "LAST-ACK") {
          state->set_last_ack(state->last_ack() + 1);
        } else if (state_str == "CLOSING") {
          state->set_closing(state->closing() + 1);
        }
      }
    }
  }
  if (socket_states.states_size() > 0) {
    return socket_states;
  }
  return absl::NotFoundError(
      "No matching NFT rules found for configured ports.");
}

absl::StatusOr<NetFilterStates> SoftwareMetrics::UpdateNetFilterStats() {
  ECCLESIA_ASSIGN_OR_RETURN(
      absl::string_view cmd_output,
      executor_->Execute(std::string(kNetFilterCmd), false));
  NetFilterStates nf_states;
  *nf_states.mutable_timestamp() = Now();
  // Parse JSON output
  nlohmann::json nft_json = nlohmann::json::parse(cmd_output, nullptr, false);
  if (nft_json.is_discarded()) {
    return absl::NotFoundError("Failed to parse JSON output from nft.");
  }
  if (!nft_json.contains("nftables") || !nft_json["nftables"].is_array()) {
    return absl::NotFoundError(
        "nftables array not found in JSON output from nft.");
  }

  // Iterate through the rules and extract data for the ports we need.
  for (const nlohmann::json& item : nft_json["nftables"]) {
    if (!item.contains("rule") || !item["rule"].is_object()) {
      continue;
    }
    const nlohmann::json& rule = item["rule"];
    if (!rule.contains("comment") || !rule["comment"].is_string()) {
      continue;
    }

    // Ensure the comment refers to one of our counter rules
    absl::string_view port_str = rule["comment"].get<absl::string_view>();
    if (!absl::ConsumePrefix(&port_str, "tcp-server-") ||
        !absl::ConsumeSuffix(&port_str, "-synack")) {
      continue;
    }

    // Make sure we have an actual port and we are configured to collect this.
    int32_t port_id;
    if (!absl::SimpleAtoi(port_str, &port_id) ||
        !nf_port_set_.contains(port_id)) {
      continue;
    }

    // This rule is for one of our target ports. Now extract the packet count.
    if (!rule.contains("expr") || !rule["expr"].is_array() ||
        rule["expr"].size() != 1) {
      LOG(WARNING) << "Rule for port " << port_id
                   << " has unexpected expr structure.";

      continue;
    }
    const nlohmann::json& counter_expr = rule["expr"][0];
    if (!counter_expr.contains("counter") ||
        !counter_expr["counter"].is_object() ||
        !counter_expr["counter"].contains("packets") ||
        !counter_expr["counter"]["packets"].is_number()) {
      LOG(WARNING) << "Rule for port " << port_id
                   << " is missing packet counter.";

      continue;
    }

    NetFilterState* state = nf_states.add_states();
    state->set_port(port_id);
    state->set_num_connections(
        counter_expr["counter"]["packets"].get<uint64_t>());
  }

  if (nf_states.states_size() > 0) {
    return nf_states;
  }

  return absl::NotFoundError(
      "No matching NFT rules found for configured ports.");
}

void SoftwareMetrics::RefreshTrueHitlessMetrics(
    absl::AnyInvocable<void()> callback) {
  absl::StatusOr<std::string> result =
      executor_->Execute(std::string(kTrueHitlessMetricsUpdateCmd), false);
  if (!result.ok()) {
    // Log every hour to avoid spamming the logs.
    LOG_EVERY_N_SEC(ERROR, 3600)
        << "Failed to start true-hitless-status-update.service: "
        << result.status();
  }

  if (callback) {
    callback();
  }
}

TrueHitlessMetrics SoftwareMetrics::GetTrueHitlessMetrics() {
  return true_hitless_metrics_parser_.ParseTrueHitlessMetricsFromJson();
}

absl::StatusOr<PressureLevels> ParsePressureOutput(absl::string_view output) {
  PressureLevels levels;
  std::stringstream ss{std::string(output)};
  std::string line;

  while (std::getline(ss, line)) {
    std::stringstream line_stream(line);
    std::string type;
    std::string avg10_str;
    std::string avg60_str;
    std::string avg300_str;
    std::string total_str;

    line_stream >> type >> avg10_str >> avg60_str >> avg300_str >> total_str;
    if (line_stream.fail()) {
      continue;
    }

    PsiPressure* pressure_data = nullptr;
    if (type == "some") {
      pressure_data = levels.mutable_some();
    } else if (type == "full") {
      pressure_data = levels.mutable_full();
    } else {
      continue;
    }

    absl::string_view avg10_sv = avg10_str;
    absl::string_view avg60_sv = avg60_str;
    absl::string_view avg300_sv = avg300_str;
    absl::string_view total_sv = total_str;
    double avg10 = 0.0;
    double avg60 = 0.0;
    double avg300 = 0.0;
    int64_t total = 0;

    if (!absl::ConsumePrefix(&avg10_sv, "avg10=") ||
        !absl::SimpleAtod(avg10_sv, &avg10)) {
      continue;
    }
    if (!absl::ConsumePrefix(&avg60_sv, "avg60=") ||
        !absl::SimpleAtod(avg60_sv, &avg60)) {
      continue;
    }
    if (!absl::ConsumePrefix(&avg300_sv, "avg300=") ||
        !absl::SimpleAtod(avg300_sv, &avg300)) {
      continue;
    }
    if (!absl::ConsumePrefix(&total_sv, "total=") ||
        !absl::SimpleAtoi(total_sv, &total)) {
      continue;
    }

    pressure_data->set_avg10(avg10);
    pressure_data->set_avg60(avg60);
    pressure_data->set_avg300(avg300);
    pressure_data->set_total(total);

    if (levels.has_some() && levels.has_full()) {
      break;
    }
  }
  if (!levels.has_some() && !levels.has_full()) {
    return absl::InvalidArgumentError("No valid pressure data found.");
  }
  return levels;
}

absl::StatusOr<std::string> ReadFileContents(const std::string& file_path) {
  std::ifstream file(file_path);
  if (!file.is_open()) {
    return absl::NotFoundError(
        absl::StrCat("Failed to open file: ", file_path));
  }

  std::stringstream buffer;
  buffer << file.rdbuf();
  return buffer.str();
}

absl::Status SoftwareMetrics::UpdatePsiData(SoftwareMetricsValue* value) {
  absl::StatusOr<PsiMetrics> psi_metrics = ReadSystemPsiMetrics();
  if (psi_metrics.ok()) {
    *value->mutable_psi_metrics() = *psi_metrics;
  }

  absl::StatusOr<std::vector<std::string>> services = ReadAvailableServices();
  if (services.ok()) {
    *value->mutable_available_services() = {services->begin(), services->end()};
    value->mutable_service_psi_metrics()->clear();
    for (const auto& service : *services) {
      absl::StatusOr<PsiMetrics> service_metrics =
          ReadPsiMetricsForService(service);
      if (service_metrics.ok()) {
        (*value->mutable_service_psi_metrics())[service] = *service_metrics;
      }
    }
  } else {
    return services.status();
  }

  return absl::OkStatus();
}

absl::StatusOr<PsiMetrics> SoftwareMetrics::ReadSystemPsiMetrics() {
  PsiMetrics psi_metrics;
  absl::StatusOr<PressureLevels> levels;

  // CPU pressure
  absl::StatusOr<std::string> cpu_file_content =
      ReadFileContents(absl::StrCat(psi_base_path_, "/cpu.pressure"));
  if (!cpu_file_content.ok()) {
    LOG(WARNING) << "Failed to read CPU pressure file: "
                 << cpu_file_content.status();
  } else {
    levels = ParsePressureOutput(*cpu_file_content);
    if (levels.ok()) {
      *psi_metrics.mutable_cpu() = *levels;
    } else {
      LOG(WARNING) << "Failed to parse CPU pressure: " << levels.status();
    }
  }

  // Memory pressure
  absl::StatusOr<std::string> memory_file_content =
      ReadFileContents(absl::StrCat(psi_base_path_, "/memory.pressure"));
  if (!memory_file_content.ok()) {
    LOG(WARNING) << "Failed to read Memory pressure file: "
                 << memory_file_content.status();
  } else {
    levels = ParsePressureOutput(*memory_file_content);
    if (levels.ok()) {
      *psi_metrics.mutable_memory() = *levels;
    } else {
      LOG(WARNING) << "Failed to parse Memory pressure: " << levels.status();
    }
  }

  // I/O pressure
  absl::StatusOr<std::string> io_file_content =
      ReadFileContents(absl::StrCat(psi_base_path_, "/io.pressure"));
  if (!io_file_content.ok()) {
    LOG(WARNING) << "Failed to read I/O pressure file: "
                 << io_file_content.status();
  } else {
    levels = ParsePressureOutput(*io_file_content);
    if (levels.ok()) {
      *psi_metrics.mutable_io() = *levels;
    } else {
      LOG(WARNING) << "Failed to parse I/O pressure: " << levels.status();
    }
  }

  return psi_metrics;
}

absl::StatusOr<std::vector<std::string>>
SoftwareMetrics::ReadAvailableServices() {
  std::string system_slice_path = absl::StrCat(psi_base_path_, "/system.slice");

  std::vector<std::string> services;
  std::error_code ec;
  if (!std::filesystem::exists(system_slice_path, ec)) {
    return absl::NotFoundError(
        absl::StrCat("System slice directory not found: ", system_slice_path));
  }

  auto dir_it = std::filesystem::directory_iterator(system_slice_path, ec);
  if (ec) {
    return absl::InternalError(
        absl::StrCat("Failed to iterate over directory: ", ec.message()));
  }

  for (auto end = std::filesystem::directory_iterator(); !ec && dir_it != end;
       dir_it.increment(ec)) {
    const auto& entry = *dir_it;
    if (entry.is_directory(ec)) {
      if (std::filesystem::exists(entry.path() / "cpu.pressure", ec)) {
        services.push_back(SystemdUnescape(entry.path().filename().string()));
      }
    }
  }
  if (ec) {
    LOG(WARNING) << "Error during directory iteration: " << ec.message();
  }

  std::sort(services.begin(), services.end());
  return services;
}

absl::StatusOr<PsiMetrics> SoftwareMetrics::ReadPsiMetricsForService(
    const std::string& service_name) {
  std::string system_slice_path = absl::StrCat(psi_base_path_, "/system.slice");

  // Build the service directory path
  std::string service_dir = absl::StrCat(system_slice_path, "/", service_name);

  std::error_code ec;
  if (!std::filesystem::exists(service_dir, ec)) {
    // Try to find the escaped name by iterating the directory
    for (const auto& entry :
         std::filesystem::directory_iterator(system_slice_path, ec)) {
      if (!entry.is_directory(ec)) {
        continue;
      }
      if (SystemdUnescape(entry.path().filename().string()) == service_name) {
        service_dir = entry.path().string();
        break;
      }
    }
  }

  // CPU pressure
  std::string cpu_path = absl::StrCat(service_dir, "/cpu.pressure");
  PsiMetrics psi_metrics;
  absl::StatusOr<PressureLevels> levels;
  absl::StatusOr<std::string> cpu_file_content = ReadFileContents(cpu_path);
  if (cpu_file_content.ok()) {
    levels = ParsePressureOutput(*cpu_file_content);
    if (levels.ok()) {
      *psi_metrics.mutable_cpu() = *levels;
    }
  }

  // Memory pressure
  std::string memory_path = absl::StrCat(service_dir, "/memory.pressure");
  absl::StatusOr<std::string> memory_file_content =
      ReadFileContents(memory_path);
  if (memory_file_content.ok()) {
    levels = ParsePressureOutput(*memory_file_content);
    if (levels.ok()) {
      *psi_metrics.mutable_memory() = *levels;
    }
  }

  // I/O pressure
  std::string io_path = absl::StrCat(service_dir, "/io.pressure");
  absl::StatusOr<std::string> io_file_content = ReadFileContents(io_path);
  if (io_file_content.ok()) {
    levels = ParsePressureOutput(*io_file_content);
    if (levels.ok()) {
      *psi_metrics.mutable_io() = *levels;
    }
  }

  return psi_metrics;
}

absl::StatusOr<PsiMetrics> SoftwareMetrics::GetPsiMetrics() const {
  if (!GetTlbmcConfig().metric_collector_module().enable_psi_metrics()) {
    return absl::NotFoundError("PSI metrics collection is disabled.");
  }
  absl::MutexLock lock(metric_data_mutex_);
  return values_.psi_metrics();
}

absl::StatusOr<std::vector<std::string>> SoftwareMetrics::GetAvailableServices()
    const {
  if (!GetTlbmcConfig().metric_collector_module().enable_psi_metrics()) {
    return absl::NotFoundError("PSI metrics collection is disabled.");
  }
  absl::MutexLock lock(metric_data_mutex_);
  return std::vector<std::string>(values_.available_services().begin(),
                                  values_.available_services().end());
}

absl::StatusOr<PsiMetrics> SoftwareMetrics::GetPsiMetricsForService(
    const std::string& service_name) const {
  if (!GetTlbmcConfig().metric_collector_module().enable_psi_metrics()) {
    return absl::NotFoundError("PSI metrics collection is disabled.");
  }
  absl::MutexLock lock(metric_data_mutex_);
  auto it = values_.service_psi_metrics().find(service_name);
  if (it == values_.service_psi_metrics().end()) {
    return absl::NotFoundError(
        absl::StrCat("Service not found: ", service_name));
  }
  return it->second;
}

absl::StatusOr<std::unique_ptr<SoftwareMetrics>>
SoftwareMetrics::CreateWithExecutorForUnitTest(
    const SoftwareMetricsConfig& config,
    absl::flat_hash_map<std::string, absl::StatusOr<std::string>>&
        command_map) {
  return CreateWithExecutorForUnitTest(config, command_map,
                                       *kDefaultPsiBasePath, "/");
}

absl::StatusOr<std::unique_ptr<SoftwareMetrics>>
SoftwareMetrics::CreateWithExecutorForUnitTest(
    const SoftwareMetricsConfig& config,
    absl::flat_hash_map<std::string, absl::StatusOr<std::string>>& command_map,
    absl::string_view psi_base_path, absl::string_view root_path) {
  SoftwareMetricsAttributesStatic software_attributes_static =
      CreateStaticAttributes(config);
  auto ptr = absl::WrapUnique(new SoftwareMetrics(software_attributes_static,
                                                  psi_base_path, root_path));

  return ptr;
}

absl::StatusOr<std::unique_ptr<SoftwareMetrics>> SoftwareMetrics::Create(
    const SoftwareMetricsConfig& config) {
  return Create(config, *kDefaultPsiBasePath, "/");
}

absl::StatusOr<std::unique_ptr<SoftwareMetrics>> SoftwareMetrics::Create(
    const SoftwareMetricsConfig& config, absl::string_view psi_base_path,
    absl::string_view root_path) {
  SoftwareMetricsAttributesStatic software_attributes_static =
      CreateStaticAttributes(config);
  return absl::WrapUnique(new SoftwareMetrics(software_attributes_static,
                                              psi_base_path, root_path));
}

void SoftwareMetrics::RefreshOnce(absl::AnyInvocable<void()> callback) {
  bool socket_stats_updated = false;
  SocketStatStates socket_stats;
  if (!ss_port_set_.empty()) {
    absl::StatusOr<SocketStatStates> res = UpdateSocketStats();
    if (res.ok()) {
      socket_stats = std::move(*res);
      socket_stats_updated = true;
    } else {
      LOG(ERROR) << "Failed to update socket stats: " << res.status();
    }
  }

  bool netfilter_stats_updated = false;
  NetFilterStates netfilter_stats;
  if (!nf_port_set_.empty()) {
    absl::StatusOr<NetFilterStates> res = UpdateNetFilterStats();
    if (res.ok()) {
      netfilter_stats = std::move(*res);
      netfilter_stats_updated = true;
    } else {
      LOG(ERROR) << "Failed to update netfilter stats: " << res.status();
    }
  }

  bool psi_updated = false;
  SoftwareMetricsValue psi_values;
  if (GetTlbmcConfig().metric_collector_module().enable_psi_metrics()) {
    absl::Status status = UpdatePsiData(&psi_values);
    if (status.ok()) {
      psi_updated = true;
    } else {
      LOG(ERROR) << "Failed to update PSI stats: " << status;
    }
  }

  if (socket_stats_updated || netfilter_stats_updated || psi_updated) {
    absl::MutexLock lock(metric_data_mutex_);
    if (socket_stats_updated) {
      values_.mutable_socket_stat_state()->Swap(&socket_stats);
    }
    if (netfilter_stats_updated) {
      values_.mutable_netfilter_state()->Swap(&netfilter_stats);
    }
    if (psi_updated) {
      values_.mutable_psi_metrics()->Swap(psi_values.mutable_psi_metrics());
      values_.mutable_available_services()->Swap(
          psi_values.mutable_available_services());
      values_.mutable_service_psi_metrics()->swap(
          *psi_values.mutable_service_psi_metrics());
    }
  }

  if (callback) {
    callback();
  }
}

absl::StatusOr<BiosKeyStatus> SoftwareMetrics::ReadBiosKeyStatus() {
  std::error_code ec;
  // Use the injected root_path_ (which is "/" in production but set to a temp
  // dir in tests) to resolve the BIOS key paths. This ensures we are sandboxed
  // correctly during integration testing without using environment variables.
  std::string root_path = root_path_;

  std::filesystem::path version_path =
      std::filesystem::path(root_path) / kBiosKeyVersionPath;
  std::filesystem::path family_path =
      std::filesystem::path(root_path) / kBiosKeyImageFamilyPath;
  std::filesystem::path method_path =
      std::filesystem::path(root_path) / kBiosKeyValidationMethodPath;
  std::filesystem::path key_data_path =
      std::filesystem::path(root_path) / kBiosKeyValidationKeyDataPath;

  if (!std::filesystem::exists(version_path, ec) ||
      !std::filesystem::exists(family_path, ec) ||
      !std::filesystem::exists(method_path, ec) ||
      !std::filesystem::exists(key_data_path, ec)) {
    return absl::NotFoundError("BIOS key status files not yet available.");
  }

  BiosKeyStatus status;

  // Read and parse Version
  ECCLESIA_ASSIGN_OR_RETURN(std::string version_str,
                            ReadFileContents(version_path.string()));
  int32_t version;
  if (!absl::SimpleAtoi(absl::StripAsciiWhitespace(version_str), &version)) {
    return absl::InternalError("Failed to parse BIOS key version.");
  }
  status.set_version(version);

  // Read and parse Image Family
  ECCLESIA_ASSIGN_OR_RETURN(std::string family_str,
                            ReadFileContents(family_path.string()));
  int32_t family;
  if (!absl::SimpleAtoi(absl::StripAsciiWhitespace(family_str), &family)) {
    return absl::InternalError("Failed to parse BIOS key image family.");
  }
  status.set_image_family(family);

  // Read Validation Method
  ECCLESIA_ASSIGN_OR_RETURN(std::string method_str,
                            ReadFileContents(method_path.string()));
  status.set_validation_method(absl::StripAsciiWhitespace(method_str));

  // Read and parse Key Data (Hex string, e.g., "0xc64d2247")
  ECCLESIA_ASSIGN_OR_RETURN(std::string key_data_str,
                            ReadFileContents(key_data_path.string()));
  std::string_view key_data_sv = absl::StripAsciiWhitespace(key_data_str);
  int64_t key_data;
  if (!absl::SimpleHexAtoi(key_data_sv, &key_data)) {
    return absl::InternalError("Failed to parse BIOS key validation key data.");
  }
  status.set_validation_key_data(key_data);

  return status;
}

void SoftwareMetrics::RefreshBiosKeyStatus() {
  if (!GetTlbmcConfig().metric_collector_module().enable_bios_key_metrics()) {
    return;
  }
  auto kr_status = ReadBiosKeyStatus();
  if (kr_status.ok()) {
    absl::MutexLock lock(metric_data_mutex_);
    *values_.mutable_bios_key_status() = *kr_status;
    LOG(INFO) << "Successfully retrieved and cached BIOS Key status.";
  } else if (!absl::IsNotFound(kr_status.status())) {
    LOG(ERROR) << "Error reading BIOS Key status: " << kr_status.status();
  }
}

}  // namespace milotic_tlbmc