tlbmc/store/store_hft_adapter.cc

#include "tlbmc/store/store_hft_adapter.h"

#include <memory>
#include <string>
#include <vector>

#include "absl/log/log.h"
#include "absl/memory/memory.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/match.h"
#include "absl/strings/str_join.h"
#include "absl/strings/str_split.h"
#include "absl/strings/substitute.h"
#include "absl/time/time.h"
#include "g3/macros.h"
#include "identifier.pb.h"
#include "payload.pb.h"
#include "sensor_identifier.pb.h"
#include "sensor_payload.pb.h"
#include "subscription_params.pb.h"
#include "tlbmc/collector/collector.h"
#include "topology_config.pb.h"
#include "sensor.pb.h"
#include "tlbmc/sensors/sensor.h"
#include "tlbmc/store/store.h"
#include "tlbmc/time/time.h"

namespace milotic_tlbmc {

namespace {

std::vector<milotic_hft::Identifier> GetSensorIdentifiers(const Store* store) {
  std::vector<milotic_hft::Identifier> identifiers;
  for (const std::shared_ptr<const Sensor>& sensor : store->GetAllSensors()) {
    milotic_hft::Identifier identifier;
    identifier.mutable_sensor_identifier()->set_name(sensor->GetKey());
    identifiers.push_back(identifier);
  }
  return identifiers;
}

absl::Status ConfigureSensors(Store* store, int sampling_interval_ms,
                              const milotic_hft::Identifier& identifier) {
  return store->ConfigureCollection(
      {.sampling_interval_ms = sampling_interval_ms,
       .key = identifier.sensor_identifier().name()},
      Collector::Type::kSensor);
}

absl::StatusOr<milotic_hft::Payload> CollectSensorData(
    const milotic_hft::SensorIdentifier& sensor_identifier, Store* store,
    absl::Time start_time) {
  milotic_hft::Payload payload;
  milotic_hft::HighFrequencySensorsReadingsBatch* batch =
      payload.mutable_high_frequency_sensors_readings_batch();
  std::shared_ptr<const Sensor> sensor =
      store->GetSensorBySensorKey(sensor_identifier.name());
  if (sensor == nullptr) {
    return absl::NotFoundError(
        absl::Substitute("Sensor $0 not found", sensor_identifier.name()));
  }
  milotic_hft::HighFrequencySensorsReadings* sensors =
      batch->add_high_frequency_sensors();
  milotic_hft::SensorIdentifier* sensor_identifier_from_readings =
      sensors->mutable_sensor_identifier();
  const SensorAttributesStatic& static_attributes =
      sensor->GetSensorAttributesStatic();

  // Get devpath for the sensor.
  // 1. Get config name from static attributes.
  // 2. Get topology node from config name.
  // 3. Get devpath from topology node.
  ECCLESIA_ASSIGN_OR_RETURN(
      const TopologyConfigNode* topology_node,
      store->GetFruTopology(
          static_attributes.entity_common_config().board_config_name()));

  // Replace the local root identifier with /phys when exporting the devpath.
  const std::string& devpath = topology_node->location_context().devpath();
  if (!absl::StartsWith(devpath, "/phys")) {
    std::vector<std::string> devpath_parts = absl::StrSplit(devpath, '/');
    if (devpath_parts.size() < 2) {
      return absl::InternalError(
          absl::Substitute("Invalid devpath: $0", devpath));
    }
    devpath_parts[1] = "phys";
    sensor_identifier_from_readings->set_devpath(
        absl::StrJoin(devpath_parts, "/"));
  } else {
    sensor_identifier_from_readings->set_devpath(devpath);
  }
  sensor_identifier_from_readings->set_name(sensor->GetKey());
  sensor_identifier_from_readings->set_source(milotic_hft::SENSOR_SOURCE_BMC);
  sensor_identifier_from_readings->set_units(
      ConvertToHftUnit(static_attributes.unit()));
  for (const auto& sensor_data :
       sensor->GetSensorDataHistorySince(start_time)) {
    milotic_hft::HighFrequencySensorReading* reading =
        sensors->add_timestamped_readings();
    reading->set_float_reading(static_cast<float>(sensor_data->reading()));
    reading->set_timestamp_ns(
        absl::ToUnixNanos(DecodeGoogleApiProto(sensor_data->timestamp())));
  }
  return payload;
}

}  // namespace

milotic_hft::SensorUnits ConvertToHftUnit(SensorUnit unit) {
  switch (unit) {
    case UNIT_DEGREE_CELSIUS:
      return milotic_hft::SENSOR_UNIT_DEGREES;
    case UNIT_WATT:
      return milotic_hft::SENSOR_UNIT_WATTS;
    case UNIT_AMPERE:
      return milotic_hft::SENSOR_UNIT_AMPS;
    case UNIT_VOLT:
      return milotic_hft::SENSOR_UNIT_VOLTS;
    case UNIT_REVOLUTION_PER_MINUTE:
      return milotic_hft::SENSOR_UNIT_RPM;
    case UNIT_PERCENT:
      return milotic_hft::SENSOR_UNIT_PERCENT;
    default:
      return milotic_hft::SENSOR_UNIT_UNSPECIFIED;
  }
}

absl::StatusOr<std::unique_ptr<StoreHftAdapter>> StoreHftAdapter::Create(
    Store* store) {
  return absl::WrapUnique(new StoreHftAdapter(store));
}

absl::Status StoreHftAdapter::ConfigureSamplingInterval(
    const milotic_hft::Identifier& identifier, int sampling_interval_ms) {
  LOG(WARNING) << "Configure called for identifier: " << identifier;
  switch (identifier.identifier_case()) {
    case milotic_hft::Identifier::kSensorIdentifier:
      ECCLESIA_RETURN_IF_ERROR(
          ConfigureSensors(store_, sampling_interval_ms, identifier));
      break;
    // Add support for other telemetry types.
    default:
      return absl::InvalidArgumentError(absl::Substitute(
          "Identifier $0 is not supported", identifier.identifier_case()));
  }
  return absl::OkStatus();
}

absl::Status StoreHftAdapter::ResetSamplingIntervalToDefault(
    const milotic_hft::Identifier& identifier) {
  switch (identifier.identifier_case()) {
    case milotic_hft::Identifier::kSensorIdentifier:
      return store_->ConfigureCollection(
          {.sampling_interval_ms = 0,  // Setting 0 resets to default value.
           .key = identifier.sensor_identifier().name()},
          Collector::Type::kSensor);
    // Add support for other telemetry types.
    default:
      return absl::InvalidArgumentError(absl::Substitute(
          "Identifier $0 is not supported", identifier.identifier_case()));
  }
}

absl::StatusOr<milotic_hft::Payload> StoreHftAdapter::Collect(
    const milotic_hft::Identifier& identifier, absl::Time start_time) const {
  switch (identifier.identifier_case()) {
    case milotic_hft::Identifier::kSensorIdentifier:
      return CollectSensorData(identifier.sensor_identifier(), store_,
                               start_time);
    // Add support for other telemetry types.
    default:
      return absl::InvalidArgumentError("Identifier is not supported");
  }
}

absl::StatusOr<std::vector<milotic_hft::Identifier>>
StoreHftAdapter::GetIdentifiersForResourceType(
    const milotic_hft::SubscriptionPolicy::ResourceType& resource_type) const {
  switch (resource_type) {
    case milotic_hft::SubscriptionPolicy::RESOURCE_TYPE_SENSOR:
      return GetSensorIdentifiers(store_);
    default:
      return absl::InvalidArgumentError("Resource type is not supported");
  }
}

absl::Status StoreHftAdapter::ConfigureBatchSize(
    const milotic_hft::Identifier& identifier, int max_batch_size) {
  switch (identifier.identifier_case()) {
    case milotic_hft::Identifier::kSensorIdentifier:
      return store_->ConfigureCollection(
          {.max_batch_size = max_batch_size,
           .key = identifier.sensor_identifier().name()},
          Collector::Type::kSensor);
    // Add support for other telemetry types.
    default:
      return absl::InvalidArgumentError(absl::Substitute(
          "Identifier $0 is not supported", identifier.identifier_case()));
  }
}

absl::Status StoreHftAdapter::ResetBatchSizeToDefault(
    const milotic_hft::Identifier& identifier) {
  switch (identifier.identifier_case()) {
    case milotic_hft::Identifier::kSensorIdentifier:
      return store_->ConfigureCollection(
          {.max_batch_size = 0,  // Setting 0 resets to default value.
           .key = identifier.sensor_identifier().name()},
          Collector::Type::kSensor);
    // Add support for other telemetry types.
    default:
      return absl::InvalidArgumentError(absl::Substitute(
          "Identifier $0 is not supported", identifier.identifier_case()));
  }
}

}  // namespace milotic_tlbmc