tlbmc/sensors/redfish_aggregated_batch.cc

#include "tlbmc/sensors/redfish_aggregated_batch.h"

#include <chrono>  // NOLINT: chrono is commonly used in BMC
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include "absl/cleanup/cleanup.h"
#include "absl/functional/any_invocable.h"
#include "absl/log/log.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include <nlohmann/json.hpp>
#include "reading_range_config.pb.h"
#include "redfish_aggregated_batch_config.pb.h"
#include "tlbmc/http/http_client_interface.h"
#include "resource.pb.h"
#include "tlbmc/sensors/redfish_aggregated_sensor.h"

namespace milotic_tlbmc {

constexpr absl::string_view kPort = "80";

absl::StatusOr<std::shared_ptr<RedfishAggregatedBatch>>
RedfishAggregatedBatch::Create(
    const RedfishAggregatedBatchConfig& config,
    const std::shared_ptr<boost::asio::io_context>& io_context,
    std::unique_ptr<HttpClientInterface> http_client,
    std::vector<std::shared_ptr<RedfishAggregatedSensor>>&& sensors) {
  return std::make_shared<RedfishAggregatedBatch>(
      config, io_context, std::move(http_client), std::move(sensors));
}

RedfishAggregatedBatch::RedfishAggregatedBatch(
    const RedfishAggregatedBatchConfig& config,
    const std::shared_ptr<boost::asio::io_context>& io_context,
    std::unique_ptr<HttpClientInterface> http_client,
    std::vector<std::shared_ptr<RedfishAggregatedSensor>>&& sensors)
    : io_context_(io_context),
      config_(config),
      http_client_(std::move(http_client)),
      redfish_location_(config.redfish_location()),
      managed_sensors_(std::move(sensors)) {
  for (auto& sensor : managed_sensors_) {
    sensor->SetIsManaged(true);
  }
}

void RedfishAggregatedBatch::RefreshOnceAsync(
    absl::AnyInvocable<void()> callback) {
  std::weak_ptr<RedfishAggregatedBatch> self = shared_from_this();
  boost::asio::post(
      *io_context_,
      [self(std::move(self)), callback = std::move(callback)]() mutable {
        std::shared_ptr<RedfishAggregatedBatch> sensor = self.lock();
        if (!sensor) {
          LOG(WARNING) << "Sensor is destroyed; cancel the refresh.";
          return;
        }

        absl::Cleanup callback_runner = [callback =
                                             std::move(callback)]() mutable {
          if (callback) {
            callback();
          }
        };

        std::chrono::steady_clock::time_point refresh_start_time =
            std::chrono::steady_clock::now();
        for (const auto& redfish_sensor : sensor->managed_sensors_) {
          redfish_sensor->SetLastRefreshStartTimeFromBatch(refresh_start_time);
        }

        std::chrono::steady_clock::time_point aggregated_polling_start_time =
            std::chrono::steady_clock::now();
        for (const auto& redfish_sensor : sensor->managed_sensors_) {
          redfish_sensor->SetAggregatedPollingStartTimeFromBatch(
              aggregated_polling_start_time);
        }

        std::weak_ptr<RedfishAggregatedBatch> weak_sensor = sensor;

        // Send an async request to the HMC.
        sensor->http_client_->SendRequest(
            sensor->config_.source(), kPort, sensor->redfish_location_,
            /*use_tls=*/false,
            [weak_sensor, cb = std::move(callback_runner)](
                boost::beast::error_code ec,
                boost::beast::http::response<boost::beast::http::string_body>
                    res) mutable {
              std::shared_ptr<RedfishAggregatedBatch> locked_weak_sensor =
                  weak_sensor.lock();
              if (!locked_weak_sensor) {
                LOG(WARNING)
                    << "Sensor destroyed before HandleBatchResponse could run.";
                return;
              }

              locked_weak_sensor->HandleBatchResponse(
                  ec, res,
                  [cb = std::move(cb)]() mutable { std::move(cb).Invoke(); });
            });
      });
}

absl::Status RedfishAggregatedBatch::ProcessResponseStatus(
    boost::beast::error_code ec,
    boost::beast::http::response<boost::beast::http::string_body> res) {
  // For aggregated sensors, we expect to get a valid response with an empty
  // error code,a 2xx status code and a non-empty body.
  if (ec) {
    // Check for error code.
    return absl::InternalError(
        absl::StrCat("Failed to read sensor value with error: ", ec.message()));
  }
  if (res.result_int() < 200 || res.result_int() >= 300) {
    // Check for response code.
    return absl::InternalError(
        absl::StrCat("Failed to read sensor value with error: Bad response "
                     "code ",
                     res.result_int()));
  }
  if (res.body().empty()) {
    // Check for empty body.
    return absl::InternalError(
        "Failed to read sensor value with error: Empty body");
  }
  return absl::OkStatus();
}

void RedfishAggregatedBatch::HandleBatchResponse(
    boost::beast::error_code ec,
    boost::beast::http::response<boost::beast::http::string_body> res,
    absl::AnyInvocable<void()> callback) {
  // This method checks for the validity of the response and updates the state
  // and timestamp for the managed sensors. It also updates the aggregated
  // polling metrics for the managed sensors.
  absl::Cleanup callback_runner = [callback = std::move(callback)]() mutable {
    if (callback) {
      callback();
    }
  };

  std::chrono::steady_clock::time_point aggregated_polling_end_time =
      std::chrono::steady_clock::now();
  for (const auto& redfish_sensor : managed_sensors_) {
    redfish_sensor->UpdateAggregatedPollingMetricsFromBatch(
        aggregated_polling_end_time);
  }

  absl::Status response_status = ProcessResponseStatus(ec, res);

  // Update the state and timestamp if the response is not valid.
  if (!response_status.ok()) {
    LOG(WARNING) << "RedfishAggregatedBatch::HandleBatchResponse failed "
                    "with error: "
                 << response_status.message();
    for (const auto& redfish_sensor : managed_sensors_) {
      redfish_sensor->UpdateStateAndTimestamp(STATUS_STALE,
                                              response_status.message());
    }
    return;
  }

  nlohmann::json root_json = nlohmann::json::parse(res.body(), nullptr, false);
  if (root_json.is_discarded()) {
    for (const auto& redfish_sensor : managed_sensors_) {
      redfish_sensor->UpdateStateAndTimestamp(STATUS_STALE,
                                              "Invalid JSON format");
    }
    return;
  }

  // Update all managed sensors with the batch response.
  for (const auto& redfish_sensor : managed_sensors_) {
    redfish_sensor->UpdateReadingFromBatch(root_json);
  }
}

}  // namespace milotic_tlbmc