tlbmc/collector/sensor_collector.cc

#include "tlbmc/collector/sensor_collector.h"

#include <array>
#include <cstddef>
#include <cstring>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <utility>
#include <vector>

#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/status/status.h"
#include "absl/status/statusor.h"
#include "absl/strings/match.h"
#include "absl/strings/str_cat.h"
#include "absl/strings/string_view.h"
#include "absl/strings/substitute.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"
#include "absl/types/span.h"
#include "boost/asio.hpp"  //NOLINT: boost::asio is commonly used in BMC
#include "g3/macros.h"
#include "thread/thread.h"
#include "gbmc_sel_defs.h"
#include "gbmc_sel_pub.h"
#include <nlohmann/json.hpp>
#include "tlbmc/central_config/config.h"
#include "tlbmc/collector/monitoring_change_base.h"
#include "tlbmc/collector/peci_scanner.h"
#include "tlbmc/collector/power_control_collector.h"
#include "adc_sensor_config.pb.h"
#include "entity_common_config.pb.h"
#include "fan_controller_config.pb.h"
#include "fan_pwm_config.pb.h"
#include "fan_tach_config.pb.h"
#include "gpio_sensor_config.pb.h"
#include "hwmon_temp_sensor_config.pb.h"
#include "intel_cpu_sensor_config.pb.h"
#include "psu_sensor_config.pb.h"
#include "redfish_aggregated_batch_config.pb.h"
#include "redfish_aggregated_sensor_config.pb.h"
#include "shared_mem_sensor_config.pb.h"
#include "threshold_config.pb.h"
#include "virtual_sensor_config.pb.h"
#include "tlbmc/expression/expression.h"
#include "tlbmc/hal/nic_veeprom/interface.h"
#include "veeprom.pb.h"
#include "tlbmc/hal/sysfs/hwmon.h"
#include "tlbmc/host_state/power_control.h"
#include "resource.pb.h"
#include "sensor.pb.h"
#include "tlbmc/sensors/adc_sensor.h"
#include "tlbmc/sensors/amd_cpu_sensor.h"
#include "tlbmc/sensors/fan_controller.h"
#include "tlbmc/sensors/fan_pwm.h"
#include "tlbmc/sensors/fan_tach.h"
#include "tlbmc/sensors/gpio_sensor.h"
#include "tlbmc/sensors/hwmon_temp_sensor.h"
#include "tlbmc/sensors/intel_cpu_sensor.h"
#include "tlbmc/sensors/nic_sensor.h"
#include "tlbmc/sensors/psu_sensor.h"
#include "tlbmc/sensors/redfish_aggregated_batch.h"
#include "tlbmc/sensors/redfish_aggregated_sensor.h"
#include "tlbmc/sensors/sensor.h"
#include "tlbmc/sensors/shared_mem_based_sensor.h"
#include "tlbmc/sensors/virtual_sensor.h"
#include "tlbmc/time/time.h"

constexpr absl::Duration kDefaultSensorSamplingInterval =
    absl::Milliseconds(1000);

namespace milotic_tlbmc {

// Returns true if the status means the sensor never came up: creation failed
// outright, or is still pending because the sensor's FRU was not detected.
bool IsCreationIncomplete(Status status) {
  return status == STATUS_CREATION_FAILED || status == STATUS_CREATION_PENDING;
}

void ScheduleIndividualSensorRead(
    const std::shared_ptr<Sensor>& sensor,
    const SensorNotification* refresh_notification,
    std::optional<int> override_sensor_sampling_interval_ms,
    ThreadManager& thread_manager) {
  absl::Duration interval = kDefaultSensorSamplingInterval;
  if (override_sensor_sampling_interval_ms.has_value()) {
    LOG(INFO) << "Overriding sensor sampling interval to "
              << *override_sensor_sampling_interval_ms;
    interval = absl::Milliseconds(*override_sensor_sampling_interval_ms);
  } else {
    absl::Duration static_refresh_interval = DecodeGoogleApiProto(
        sensor->GetSensorAttributesStatic().static_refresh_interval());
    if (static_refresh_interval > absl::ZeroDuration()) {
      interval = static_refresh_interval;
    }
  }

  const Status status = sensor->GetSensorAttributesDynamic().state().status();
  if (IsCreationIncomplete(status)) {
    LOG(INFO) << "Skipping schedule sensor read for " << sensor->GetKey()
              << " because sensor initialization failed. Status: " << status;
    return;
  }
  LOG(INFO) << "Scheduling Sensor Read! Sensor key is " << sensor->GetKey()
            << " and interval is " << interval;

  {
    absl::MutexLock lock(thread_manager.key_to_task_id_mutex);
    auto it = thread_manager.sensor_key_to_task_id.find(sensor->GetKey());
    if (it != thread_manager.sensor_key_to_task_id.end()) {
      thread_manager.task_scheduler->Cancel(it->second);
    }
    int task_id = thread_manager.task_scheduler->RunAndScheduleAsync(
        [sensor = std::weak_ptr<Sensor>(sensor),
         refresh_notification =
             refresh_notification](absl::AnyInvocable<void()> on_done) {
          std::shared_ptr<Sensor> sensor_locked = sensor.lock();
          if (!sensor_locked) {
            return;
          }

          sensor_locked->RefreshOnceAsync(
              [refresh_notification = refresh_notification,
               on_done =
                   std::move(on_done)](const std::shared_ptr<const SensorValue>&
                                           sensor_data) mutable {
                if (refresh_notification != nullptr) {
                  refresh_notification->NotifyWithData(sensor_data);
                }
                on_done();
              });
        },
        interval);

    thread_manager.sensor_key_to_task_id[sensor->GetKey()] = task_id;
  }
}

void ScheduleAllSensorReads(
    const std::vector<std::shared_ptr<Sensor>>& sensors,
    const SensorNotification* refresh_notification,
    std::optional<int> override_sensor_sampling_interval_ms,
    ThreadManager& thread_manager) {
  LOG(INFO) << "Scheduling Sensor Read! Total sensor count is "
            << sensors.size();
  absl::flat_hash_map<absl::Duration, std::vector<std::weak_ptr<Sensor>>>
      sensors_by_interval;
  for (const auto& sensor : sensors) {
    ScheduleIndividualSensorRead(sensor, refresh_notification,
                                 override_sensor_sampling_interval_ms,
                                 thread_manager);
  }
}

void ScheduleIndividualBatchRead(
    const std::shared_ptr<RedfishAggregatedBatch>& batch,
    const SensorNotification* refresh_notification,
    std::optional<int> override_sensor_sampling_interval_ms,
    ThreadManager& thread_manager) {
  absl::Duration interval = kDefaultSensorSamplingInterval;
  if (override_sensor_sampling_interval_ms.has_value()) {
    interval = absl::Milliseconds(*override_sensor_sampling_interval_ms);
  }

  LOG(INFO) << "Scheduling Redfish Batch Read! Batch name is "
            << batch->GetBatchName() << " and interval is " << interval;

  {
    absl::MutexLock lock(thread_manager.key_to_task_id_mutex);
    auto it = thread_manager.sensor_key_to_task_id.find(batch->GetBatchName());
    if (it != thread_manager.sensor_key_to_task_id.end()) {
      thread_manager.task_scheduler->Cancel(it->second);
    }
    int task_id = thread_manager.task_scheduler->RunAndScheduleAsync(
        [batch = std::weak_ptr<RedfishAggregatedBatch>(batch),
         refresh_notification =
             refresh_notification](absl::AnyInvocable<void()> on_done) {
          std::shared_ptr<RedfishAggregatedBatch> batch_locked = batch.lock();
          if (!batch_locked) {
            return;
          }

          batch_locked->RefreshOnceAsync(
              [refresh_notification, on_done = std::move(on_done)]() mutable {
                if (refresh_notification != nullptr) {
                  refresh_notification->NotifyWithData(nullptr);
                }
                on_done();
              });
        },
        interval);

    thread_manager.sensor_key_to_task_id[batch->GetBatchName()] = task_id;
  }
}

void ScheduleAllBatchSensorReads(
    const std::vector<std::shared_ptr<RedfishAggregatedBatch>>&
        redfish_aggregated_batches,
    const SensorNotification* refresh_notification,
    std::optional<int> override_sensor_sampling_interval_ms,
    ThreadManager& thread_manager) {
  LOG(INFO)
      << "Scheduling Sensor Read! Total redfish aggregated batch count is "
      << redfish_aggregated_batches.size();
  for (const auto& batch : redfish_aggregated_batches) {
    ScheduleIndividualBatchRead(batch, refresh_notification,
                                override_sensor_sampling_interval_ms,
                                thread_manager);
  }
}

// This class is used to apply sampling interval and batch size changes to the
// SensorCollector in a transaction-like manner.
class SensorCollectorMonitoringChange : public MonitoringChangeBase {
 public:
  explicit SensorCollectorMonitoringChange(SensorCollector* sensor_collector)
      : sensor_collector_(sensor_collector) {}

  bool AddSensor(std::string_view sensor_key, Mutation mutation) & override {
    if (sensor_collector_->GetSensorBySensorKey(std::string(sensor_key)) !=
        nullptr) {
      auto& current = mutations_[sensor_key];
      current.MergeWith(mutation);
      return true;
    }
    return false;
  }

  void Apply() && override {
    for (const auto& [sensor_key, mutation] : mutations_) {
      sensor_collector_->ConfigureCollection(sensor_key, mutation)
          .IgnoreError();
    }
  }

 private:
  SensorCollector* sensor_collector_;
  absl::flat_hash_map<std::string, Mutation> mutations_;
};

namespace {

constexpr absl::string_view kDefaultHwmonContext = "DEFAULT_HWMON_CONTEXT";
constexpr absl::string_view kDefaultPsuContext = "DEFAULT_PSU_CONTEXT";
constexpr absl::string_view kDefaultFanContext = "DEFAULT_FAN_CONTEXT";
constexpr absl::string_view kDefaultVirtualSensorContext =
    "DEFAULT_VIRTUAL_SENSOR_CONTEXT";
constexpr absl::string_view kDefaultIntelCpuContext =
    "DEFAULT_INTEL_CPU_CONTEXT";
constexpr absl::string_view kDefaultRedfishAggregatedSensorContext =
    "DEFAULT_REDFISH_AGGREGATED_SENSOR_CONTEXT";
constexpr absl::string_view kDefaultRedfishAggregatedBatchSensorContext =
    "DEFAULT_REDFISH_AGGREGATED_BATCH_SENSOR_CONTEXT";
constexpr absl::string_view kDefaultNicTelemetryContext =
    "DEFAULT_NIC_TELEMETRY_CONTEXT";
constexpr absl::string_view kDefaultNicTelemetryRefreshContext =
    "DEFAULT_NIC_TELEMETRY_REFRESH_CONTEXT";
constexpr absl::string_view kDefaultGpioContext = "DEFAULT_GPIO_CONTEXT";
constexpr absl::string_view kDefaultAdcSensorContext =
    "DEFAULT_ADC_SENSOR_CONTEXT";

std::optional<gbmc_sel_framework::EventSourceComponent> GetEventSourceComponent(
    SensorUnit unit) {
  switch (unit) {
    case UNIT_DEGREE_CELSIUS:
      return gbmc_sel_framework::EventSourceComponent::
          EVENT_SOURCE_COMPONENT_TemperatureSensor;
    case UNIT_WATT:
      return gbmc_sel_framework::EventSourceComponent::
          EVENT_SOURCE_COMPONENT_PowerSensor;
    case UNIT_AMPERE:
      return gbmc_sel_framework::EventSourceComponent::
          EVENT_SOURCE_COMPONENT_CurrentSensor;
    case UNIT_VOLT:
      return gbmc_sel_framework::EventSourceComponent::
          EVENT_SOURCE_COMPONENT_VoltageSensor;
    case UNIT_REVOLUTION_PER_MINUTE:
      return gbmc_sel_framework::EventSourceComponent::
          EVENT_SOURCE_COMPONENT_FAN;
    default:
      return std::nullopt;
  }
}

gbmc_sel_framework::EventSeverity GetEventSeverity(
    ThresholdType threshold_type) {
  if (threshold_type == THRESHOLD_TYPE_UPPER_CRITICAL ||
      threshold_type == THRESHOLD_TYPE_LOWER_CRITICAL) {
    return gbmc_sel_framework::EventSeverity::EVENT_SEVERITY_ERROR;
  }
  return gbmc_sel_framework::EventSeverity::EVENT_SEVERITY_WARNING;
}

std::optional<absl::string_view> GetThresholdEventStr(
    ThresholdType threshold_type) {
  switch (threshold_type) {
    case THRESHOLD_TYPE_UPPER_CRITICAL:
      return "OverCriticalThreshold";
    case THRESHOLD_TYPE_UPPER_NON_CRITICAL:
      return "OverNonCriticalThreshold";
    case THRESHOLD_TYPE_LOWER_CRITICAL:
      return "LowerCriticalThreshold";
    case THRESHOLD_TYPE_LOWER_NON_CRITICAL:
      return "LowerNonCriticalThreshold";
    default:
      return std::nullopt;
  }
}

gbmc_sel_framework::EventSourceType GetEventSourceType(
    absl::string_view devpath) {
  for (const auto& [type, name] : gbmc_sel_framework::kEventSourceNames) {
    if (absl::StrContains(devpath, name)) {
      return type;
    }
  }
  return gbmc_sel_framework::EventSourceType::EVENT_SOURCE_UNSPECIFIED;
}

std::vector<std::string> BuildAdditionalData(
    absl::string_view sensor_name, double reading,
    absl::string_view threshold_event_str,
    std::optional<absl::string_view> devpath) {
  std::vector<std::string> additional_data = {
      "GBMC_SYSTEM_EVENT_TYPE=SensorEvent",
      absl::StrCat("GBMC_SENSOR_NAME=", sensor_name),
      absl::StrCat("GBMC_SENSOR_VALUE=", reading),
      absl::StrCat("GBMC_SENSOR_THRESHOLD_EVENT=", threshold_event_str)};
  if (devpath.has_value()) {
    additional_data.push_back(absl::StrCat("GBMC_SENSOR_DEVPATH=", *devpath));
  }
  return additional_data;
}

void ResizeBufferAndMetrics(const std::shared_ptr<Sensor>& sensor,
                            size_t buffer_size_from_config) {
  size_t default_buffer_size =
      sensor->GetSensorAttributesStatic().entity_common_config().queue_size();
  if (buffer_size_from_config > default_buffer_size) {
    sensor->ResizeBuffer(buffer_size_from_config);
  } else {
    // If the buffer size from the config is less than the default buffer size,
    // we need to resize the buffer to the default buffer size.
    // This covers both the cases where we need to reset the buffer size to the
    // default value and when a lower than default size is specified in the
    // config.
    sensor->ResizeBuffer(default_buffer_size);
  }
  sensor->ResetMetrics();
}

// Returns the io_context for the given sensor group. If the io_context does not
// exist, it will be created. SensorGroup should be a unique identifier for
// sensors that share the same i2c device or are otherwise related to optimize
// reading speed for HFT and minimize contention over the bus mutex. See
// b/434024387 for details.
std::shared_ptr<boost::asio::io_context> GetOrCreateIoContextForSensorGroup(
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context,
    absl::string_view sensor_group) {
  auto& group_io_context = sensor_group_to_io_context[sensor_group];
  if (group_io_context == nullptr) {
    group_io_context = std::make_shared<boost::asio::io_context>();
  }
  return group_io_context;
}

absl::Status CreateHwmonSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  std::vector<std::shared_ptr<Sensor>> hwmon_temp_sensors;
  size_t count_sensors = 0;
  for (const auto& config : params.sensor_configs.hwmon_temp_sensor_configs) {
    absl::string_view sensor_group =
        config.entity_common_config().has_sensor_group()
            ? config.entity_common_config().sensor_group()
            : kDefaultHwmonContext;
    ECCLESIA_ASSIGN_OR_RETURN(
        hwmon_temp_sensors,
        HwmonTempSensor::Create(config,
                                GetOrCreateIoContextForSensorGroup(
                                    sensor_group_to_io_context, sensor_group),
                                params.i2c_sysfs));
    // We want to use syslog to track device creation status
    LOG(INFO) << absl::Substitute("Created $0 HWmon sensors at $1",
                                  hwmon_temp_sensors.size(),
                                  config.hal_common_config());
    count_sensors += hwmon_temp_sensors.size();
    sensors.insert(sensors.end(), hwmon_temp_sensors.begin(),
                   hwmon_temp_sensors.end());
  }
  return absl::OkStatus();
}

absl::Status CreatePsuSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  size_t count_sensors = 0;
  for (const auto& config : params.sensor_configs.psu_sensor_configs) {
    absl::string_view sensor_group =
        config.entity_common_config().has_sensor_group()
            ? config.entity_common_config().sensor_group()
            : kDefaultPsuContext;
    // Default to use i2c
    HwmonSysfs* hwmon_sysfs = &params.i2c_sysfs;
    std::vector<std::shared_ptr<Sensor>> psu_sensors;
    if (AmdCpuSensor::IsAmdCpuSensor(config.type())) {
      ECCLESIA_ASSIGN_OR_RETURN(
          psu_sensors,
          AmdCpuSensor::Create(config,
                               GetOrCreateIoContextForSensorGroup(
                                   sensor_group_to_io_context, sensor_group),
                               params.i3c_sysfs));
    } else {
      // There's no I3C sensor other than AMD CPU sensors, use i2c by default.
      ECCLESIA_ASSIGN_OR_RETURN(
          psu_sensors,
          PsuSensor::Create(config,
                            GetOrCreateIoContextForSensorGroup(
                                sensor_group_to_io_context, sensor_group),
                            *hwmon_sysfs));
    }
    // We want to use syslog to track device creation status
    LOG(INFO) << absl::Substitute("Created $0 PSU sensors at $1",
                                  psu_sensors.size(),
                                  config.hal_common_config());
    count_sensors += psu_sensors.size();
    sensors.insert(sensors.end(), psu_sensors.begin(), psu_sensors.end());
  }
  return absl::OkStatus();
}

absl::Status CreateFanSensors(
    const SensorCollector::Params& params,
    absl::Span<const std::shared_ptr<FanController>> fan_controllers,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  int count = 0;
  for (const auto& config : params.sensor_configs.fan_pwm_configs) {
    for (const auto& fan_controller : fan_controllers) {
      if (!fan_controller->ControllerHasSensor(config.hal_common_config())) {
        continue;
      }
      absl::string_view sensor_group =
          config.entity_common_config().has_sensor_group()
              ? config.entity_common_config().sensor_group()
              : kDefaultFanContext;
      ECCLESIA_ASSIGN_OR_RETURN(
          std::shared_ptr<Sensor> fan_pwm,
          FanPwm::Create(config, *fan_controller,
                         GetOrCreateIoContextForSensorGroup(
                             sensor_group_to_io_context, sensor_group),
                         params.i2c_sysfs));
      sensors.push_back(std::move(fan_pwm));
      count++;
      break;
    }
  }
  for (const auto& config : params.sensor_configs.fan_tach_configs) {
    for (const auto& fan_controller : fan_controllers) {
      if (!fan_controller->ControllerHasSensor(config.hal_common_config())) {
        continue;
      }
      absl::string_view sensor_group =
          config.entity_common_config().has_sensor_group()
              ? config.entity_common_config().sensor_group()
              : kDefaultFanContext;
      ECCLESIA_ASSIGN_OR_RETURN(
          std::shared_ptr<Sensor> fan_tach,
          FanTachometer::Create(config, *fan_controller,
                                GetOrCreateIoContextForSensorGroup(
                                    sensor_group_to_io_context, sensor_group),
                                params.i2c_sysfs));
      sensors.push_back(std::move(fan_tach));
      count++;
      break;
    }
  }
  // We want to use syslog to track device creation status
  LOG(INFO) << absl::Substitute("Created $0 Fan PWM/Tach sensors", count);
  return absl::OkStatus();
}

absl::Status CreateGpioSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  if (params.gpio_collector == nullptr) {
    return absl::OkStatus();
  }
  for (const auto& config : params.sensor_configs.gpio_sensor_configs) {
    ECCLESIA_ASSIGN_OR_RETURN(
        std::shared_ptr<Sensor> gpio_sensor,
        GpioSensor::Create(config,
                           GetOrCreateIoContextForSensorGroup(
                               sensor_group_to_io_context, kDefaultGpioContext),
                           params.gpio_collector));
    // We want to use syslog to track device creation status
    LOG(INFO) << absl::Substitute("Created Gpio sensor: $0",
                                  config.instance_properties().name());
    sensors.push_back(std::move(gpio_sensor));
  }
  return absl::OkStatus();
}

absl::Status CreateSharedMemSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    ThreadManager& thread_manager) {
  auto io_context = std::make_shared<boost::asio::io_context>();
  boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
      work_guard(boost::asio::make_work_guard(*io_context));
  int count = 0;
  for (const auto& config : params.sensor_configs.shared_mem_sensor_configs) {
    ECCLESIA_ASSIGN_OR_RETURN(std::shared_ptr<Sensor> shared_mem_sensor,
                              SharedMemBasedSensor::Create(config, io_context));
    // We want to use syslog to track device creation status
    LOG(INFO) << absl::Substitute("Created SharedMem sensor: $0",
                                  config.instance_properties().name());
    sensors.push_back(std::move(shared_mem_sensor));
    count++;
  }
  if (count > 0) {
    thread_manager.work_guards.push_back(std::move(work_guard));
    thread_manager.io_contexts.push_back(std::move(io_context));
  }
  return absl::OkStatus();
}

absl::Status CreateVirtualSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  absl::flat_hash_map<std::string, std::shared_ptr<Sensor>> sensor_map;
  for (const std::shared_ptr<Sensor>& sensor : sensors) {
    sensor_map[sensor->GetKey()] = sensor;
  }

  // All virtual sensors are created in the same io_context.
  std::shared_ptr<boost::asio::io_context> io_context =
      GetOrCreateIoContextForSensorGroup(sensor_group_to_io_context,
                                         kDefaultVirtualSensorContext);

  for (const auto& config : params.sensor_configs.virtual_sensor_configs) {
    absl::flat_hash_map<std::string, std::shared_ptr<Sensor>>
        reference_sensors_map;
    ECCLESIA_ASSIGN_OR_RETURN(std::unique_ptr<Expression> expression,
                              Parse(config.expression()));
    absl::flat_hash_set<std::string> required_sensors;
    expression->GetRequiredVariables(required_sensors);
    for (const auto& sensor_key : required_sensors) {
      auto it = sensor_map.find(sensor_key);
      if (it == sensor_map.end()) {
        if (!GetTlbmcConfig()
                 .sensor_collector_module()
                 .allow_sensor_creation_failure()) {
          return absl::InvalidArgumentError(
              absl::StrCat("Reference sensor not found: ", sensor_key));
        }
        reference_sensors_map[sensor_key] = nullptr;
      } else {
        reference_sensors_map[sensor_key] = it->second;
      }
    }
    ECCLESIA_ASSIGN_OR_RETURN(
        std::shared_ptr<Sensor> virtual_sensor,
        VirtualSensor::Create(config, io_context, reference_sensors_map,
                              std::move(expression)));
    LOG(INFO) << absl::StrCat("Created Virtual sensor: ",
                              config.instance_properties().name());
    sensors.push_back(std::move(virtual_sensor));
  }
  return absl::OkStatus();
}

// Creates NIC sensors and adds them to the sensors vector.
// Also creates the accessors for the NICs per bus.
absl::Status CreateNicSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v1_accessor,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v2_accessor,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v4_accessor,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v5_accessor) {
  std::string i2c_sysfs_path = params.i2c_sysfs.GetI2cSysfsPath();
  std::shared_ptr<boost::asio::io_context> io_context =
      GetOrCreateIoContextForSensorGroup(sensor_group_to_io_context,
                                         kDefaultNicTelemetryContext);
  for (const auto& config : params.sensor_configs.nic_telemetry_configs) {
    int bus = static_cast<int>(config.hal_common_config().bus());
    // Existing assumption on
    // https://gbmc-internal.git.corp.google.com/diorite-hss/+/8c81357e247e55df1ee2e3799f32550d43124c46/find_diorite.cpp#223
    constexpr int kNicVirtualEepromAddress = 75;
    nic_veeprom::Accessor* accessor = nullptr;
    if (config.version() == nic_veeprom::NIC_TELEMETRY_VERSION_V1) {
      auto& v1_accessor = bus_to_v1_accessor[bus];
      if (v1_accessor == nullptr) {
        v1_accessor = params.nic_accessor_factory(config.version(), bus,
                                                  kNicVirtualEepromAddress);
      }
      accessor = v1_accessor.get();
    }
    if (config.version() == nic_veeprom::NIC_TELEMETRY_VERSION_V2) {
      auto& v2_accessor = bus_to_v2_accessor[bus];
      if (v2_accessor == nullptr) {
        v2_accessor = params.nic_accessor_factory(config.version(), bus,
                                                  kNicVirtualEepromAddress);
      }
      accessor = v2_accessor.get();
    }
    if (config.version() == nic_veeprom::NIC_TELEMETRY_VERSION_V4) {
      auto& v4_accessor = bus_to_v4_accessor[bus];
      if (v4_accessor == nullptr) {
        v4_accessor = params.nic_accessor_factory(config.version(), bus,
                                                  kNicVirtualEepromAddress);
      }
      accessor = v4_accessor.get();
    }

    if (config.version() == nic_veeprom::NIC_TELEMETRY_VERSION_V5) {
      auto& v5_accessor = bus_to_v5_accessor[bus];
      if (v5_accessor == nullptr) {
        v5_accessor = params.nic_accessor_factory(config.version(), bus,
                                                  kNicVirtualEepromAddress);
      }
      accessor = v5_accessor.get();
    }

    if (accessor == nullptr) {
      return absl::InternalError(
          "Failed to create accessor for NIC telemetry config: no accessor");
    }
    for (const auto& telemetry_instance : config.telemetry_instances()) {
      // For now, unit won't be set. Sensor unit can be derived from the sensor
      // name.
      ECCLESIA_ASSIGN_OR_RETURN(
          std::shared_ptr<Sensor> nic_sensor,
          NicSensor::Create(telemetry_instance, config.entity_common_config(),
                            io_context, *accessor));
      sensors.push_back(std::move(nic_sensor));
    }
  }
  return absl::OkStatus();
}

// Creates ADC sensors and adds them to the sensors vector.
absl::Status CreateAdcSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  std::shared_ptr<boost::asio::io_context> io_context =
      GetOrCreateIoContextForSensorGroup(sensor_group_to_io_context,
                                         kDefaultAdcSensorContext);

  for (const auto& config : params.sensor_configs.adc_sensor_configs) {
    for (const auto& channel_config : config.channels()) {
      ECCLESIA_ASSIGN_OR_RETURN(
          std::shared_ptr<Sensor> adc_sensor,
          AdcSensor::Create(
              {.device_name = config.device_name(),
               .channel_config = channel_config,
               .hal_common_config = config.hal_common_config(),
               .entity_common_config = config.entity_common_config(),
               .io_context = io_context,
               .type = config.type()},
              params.iio_sysfs, params.i2c_sysfs));
      sensors.push_back(std::move(adc_sensor));
    }
  }
  return absl::OkStatus();
}

absl::StatusOr<std::vector<std::shared_ptr<FanController>>>
CreateFanControllers(const SensorCollector::Params& params) {
  std::vector<std::shared_ptr<FanController>> fan_controllers;
  for (const auto& config : params.sensor_configs.fan_controller_configs) {
    ECCLESIA_ASSIGN_OR_RETURN(std::shared_ptr<FanController> fan_controller,
                              FanController::Create(config, params.i2c_sysfs));
    // We want to use syslog to track device creation status
    LOG(INFO) << absl::Substitute("Created 1 fan controller at $0",
                                  config.hal_common_config());
    fan_controllers.push_back(std::move(fan_controller));
  }
  return fan_controllers;
}

absl::Status CreateIntelCpuSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  for (const auto& config : params.sensor_configs.intel_cpu_sensor_configs) {
    absl::string_view sensor_group =
        config.entity_common_config().has_sensor_group()
            ? config.entity_common_config().sensor_group()
            : kDefaultIntelCpuContext;
    ECCLESIA_ASSIGN_OR_RETURN(
        std::vector<std::shared_ptr<Sensor>> intel_cpu_sensors,
        IntelCpuSensor::CreateInitialSensors(
            config,
            GetOrCreateIoContextForSensorGroup(sensor_group_to_io_context,
                                               sensor_group),
            params.peci_sysfs));
    // We want to use syslog to track device creation status
    LOG(INFO) << absl::Substitute("Created $0 Intel CPU sensors at $1",
                                  intel_cpu_sensors.size(),
                                  config.hal_common_config());
    sensors.insert(sensors.end(), intel_cpu_sensors.begin(),
                   intel_cpu_sensors.end());
  }

  return absl::OkStatus();
}

absl::Status CreateRedfishAggregatedSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  absl::string_view sensor_group = kDefaultRedfishAggregatedSensorContext;
  std::shared_ptr<boost::asio::io_context> io_context =
      GetOrCreateIoContextForSensorGroup(sensor_group_to_io_context,
                                         sensor_group);
  for (const auto& config :
       params.sensor_configs.redfish_aggregated_sensor_configs) {
    ECCLESIA_ASSIGN_OR_RETURN(
        std::shared_ptr<Sensor> redfish_aggregated_sensor,
        RedfishAggregatedSensor::Create(
            config, io_context, params.http_client_factory(*io_context)));
    sensors.push_back(std::move(redfish_aggregated_sensor));
  }
  return absl::OkStatus();
}

absl::Status CreateRedfishAggregatedBatchSensors(
    const SensorCollector::Params& params,
    std::vector<std::shared_ptr<Sensor>>& sensors,
    std::vector<std::shared_ptr<RedfishAggregatedBatch>>&
        redfish_aggregated_batches,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context) {
  absl::string_view sensor_group = kDefaultRedfishAggregatedBatchSensorContext;
  std::shared_ptr<boost::asio::io_context> io_context =
      GetOrCreateIoContextForSensorGroup(sensor_group_to_io_context,
                                         sensor_group);

  for (const auto& config :
       params.sensor_configs.redfish_aggregated_batch_configs) {
    std::vector<std::shared_ptr<RedfishAggregatedSensor>>
        redfish_aggregated_sensors;
    for (const auto& telemetry_config : config.telemetry_configs()) {
      ECCLESIA_ASSIGN_OR_RETURN(
          std::shared_ptr<RedfishAggregatedSensor> redfish_aggregated_sensor,
          RedfishAggregatedSensor::Create(
              telemetry_config, io_context,
              params.http_client_factory(*io_context)));
      redfish_aggregated_sensors.push_back(redfish_aggregated_sensor);
      sensors.push_back(std::move(redfish_aggregated_sensor));
    }
    ECCLESIA_ASSIGN_OR_RETURN(
        std::shared_ptr<RedfishAggregatedBatch> redfish_aggregated_batch,
        RedfishAggregatedBatch::Create(config, io_context,
                                       params.http_client_factory(*io_context),
                                       std::move(redfish_aggregated_sensors)));
    redfish_aggregated_batches.push_back(std::move(redfish_aggregated_batch));
  }
  return absl::OkStatus();
}

// Drops sensors that failed creation when a successfully created sensor with
// the same sensor key exists on another board. Supports platforms where one
// physical device is configured under every board it could belong to because
// the board cannot be identified at config time: only the config matching the
// real hardware initializes, and without this filter the winner of key-based
// lookups (GetSensorBySensorKey and everything built on it) is decided by
// hash-map iteration order. Sensors with unique keys are kept regardless of
// status, so ordinary failed sensors remain visible for diagnosis.
void DropFailedDuplicateSensors(std::vector<std::shared_ptr<Sensor>>& sensors) {
  absl::flat_hash_set<absl::string_view> healthy_keys;
  for (const std::shared_ptr<Sensor>& sensor : sensors) {
    if (!IsCreationIncomplete(
            sensor->GetSensorAttributesDynamic().state().status())) {
      healthy_keys.insert(sensor->GetKey());
    }
  }
  std::erase_if(sensors, [&healthy_keys](
                             const std::shared_ptr<Sensor>& sensor) {
    const Status status = sensor->GetSensorAttributesDynamic().state().status();
    const bool is_shadowed_duplicate =
        IsCreationIncomplete(status) && healthy_keys.contains(sensor->GetKey());
    if (!is_shadowed_duplicate) {
      return false;
    }
    LOG(INFO) << "Dropping sensor " << sensor->GetKey() << " on board "
              << sensor->GetBoardConfigKey()
              << " (status: " << Status_Name(status)
              << "): a healthy sensor with the same key exists "
                 "(drop_failed_duplicate_sensors).";
    return true;
  });
}

absl::flat_hash_map<std::string,
                    absl::flat_hash_map<std::string, std::shared_ptr<Sensor>>>
CreateSensorsTable(std::vector<std::shared_ptr<Sensor>>&& sensors) {
  absl::flat_hash_map<std::string,
                      absl::flat_hash_map<std::string, std::shared_ptr<Sensor>>>
      sensor_table;
  for (auto& sensor : sensors) {
    sensor_table[sensor->GetBoardConfigKey()][sensor->GetKey()] =
        std::move(sensor);
  }
  return sensor_table;
}

// NIC sensors won't be refreshed at the sensor level. Instead, we will have a
// dedicated task to refresh all NIC sensors together per accessor.
void ScheduleNicSensorAccessorRefresh(
    const absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v1_accessor,
    const absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v2_accessor,
    const absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v4_accessor,
    const absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&
        bus_to_v5_accessor,
    absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>&
        sensor_group_to_io_context,
    ThreadManager& thread_manager) {
  if (!bus_to_v1_accessor.empty() || !bus_to_v2_accessor.empty() ||
      !bus_to_v4_accessor.empty() || !bus_to_v5_accessor.empty()) {
    std::vector<nic_veeprom::Accessor*> accessors;
    auto& io_context =
        sensor_group_to_io_context[kDefaultNicTelemetryRefreshContext];
    io_context = std::make_shared<boost::asio::io_context>();
    for (const auto& [bus, accessor] : bus_to_v1_accessor) {
      accessors.push_back(accessor.get());
    }
    for (const auto& [bus, accessor] : bus_to_v2_accessor) {
      accessors.push_back(accessor.get());
    }
    for (const auto& [bus, accessor] : bus_to_v4_accessor) {
      accessors.push_back(accessor.get());
    }
    for (const auto& [bus, accessor] : bus_to_v5_accessor) {
      accessors.push_back(accessor.get());
    }
    // Now have a dedicated task to refresh the NIC sensors.
    thread_manager.task_scheduler->RunAndScheduleAsync(
        [accessors = std::move(accessors),
         io_context = io_context](absl::AnyInvocable<void()> on_done) mutable {
          for (auto* accessor : accessors) {
            boost::asio::post(*io_context,
                              [accessor]() { accessor->DoRefresh(); });
          }
          on_done();
        },
        absl::Seconds(1), absl::Seconds(10));
  }
}

}  // namespace

nlohmann::json SensorCollector::ToJson() const {
  nlohmann::json::object_t response;
  for (const auto& [board_config_key, key_to_sensor] : sensor_table_) {
    nlohmann::json::object_t sensors;
    for (const auto& [key, sensor] : key_to_sensor) {
      if (SharedMemBasedSensor::IsSharedMemSensorKey(key)) {
        continue;
      }
      absl::StatusOr<nlohmann::json> sensor_json = sensor->ToJson();
      if (!sensor_json.ok()) {
        LOG(ERROR) << "Failed to get sensor data for " << key << ": "
                   << sensor_json.status();
        continue;
      }
      sensors[key] = *std::move(sensor_json);
    }
    response[board_config_key] = sensors;
  }
  return response;
}

SensorCollector::SensorCollector(
    std::vector<std::shared_ptr<Sensor>>&& sensors,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&&
        bus_to_v1_accessor,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&&
        bus_to_v2_accessor,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&&
        bus_to_v4_accessor,
    absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>&&
        bus_to_v5_accessor,
    std::unique_ptr<ThreadManager> thread_manager,
    const SensorNotification* refresh_notification,
    std::optional<int> override_sensor_sampling_interval_ms,
    ecclesia::ThreadFactoryInterface* thread_factory,
    std::unique_ptr<PeciScanner> peci_scanner,
    std::vector<std::shared_ptr<RedfishAggregatedBatch>>&&
        redfish_aggregated_batches,
    bool enable_threshold_monitoring)
    : bus_to_v1_accessor_(std::move(bus_to_v1_accessor)),
      bus_to_v2_accessor_(std::move(bus_to_v2_accessor)),
      bus_to_v4_accessor_(std::move(bus_to_v4_accessor)),
      bus_to_v5_accessor_(std::move(bus_to_v5_accessor)),
      sensor_table_(CreateSensorsTable(std::move(sensors))),
      redfish_aggregated_batches_(std::move(redfish_aggregated_batches)),
      thread_manager_(std::move(thread_manager)),
      peci_scanner_(std::move(peci_scanner)),
      refresh_notification_(refresh_notification),
      override_sensor_sampling_interval_ms_(
          override_sensor_sampling_interval_ms),
      thread_factory_(thread_factory),
      enable_threshold_monitoring_(enable_threshold_monitoring) {
  // We only populate sensors_with_thresholds_ if threshold monitoring is
  // enabled to avoid unnecessary overhead.
  // We do not expect thresholds to be added dynamically at runtime.
  if (enable_threshold_monitoring_) {
    for (const auto& [_, key_to_sensor] : sensor_table_) {
      for (const auto& [_, sensor] : key_to_sensor) {
        if (!sensor->GetSensorAttributesDynamic()
                 .thresholds()
                 .threshold_configs()
                 .empty()) {
          sensors_with_thresholds_.push_back(sensor);
        }
      }
    }
  }
}

SensorCollector::~SensorCollector() {
  // In the case of EmptySensorCollector, avoid dereferencing nullptr.
  if (thread_manager_ == nullptr) {
    return;
  }
  // Stop the scheduler first.
  thread_manager_->task_scheduler->Stop();

  // Stop io_contexts.
  for (const std::shared_ptr<boost::asio::io_context>& io_context :
       thread_manager_->io_contexts) {
    io_context->stop();
  }

  // Finally join all threads.
  for (const std::unique_ptr<ecclesia::ThreadInterface>& thread :
       thread_manager_->threads) {
    thread->Join();
  }
  peci_scanner_.reset();
  thread_manager_.reset();
}

void SensorCollector::GetAllSensorKeysByConfigKey(
    const std::string& board_config_key,
    std::vector<std::string>& sensor_keys) const {
  if (auto it = sensor_table_.find(board_config_key);
      it != sensor_table_.end()) {
    for (const auto& [key, _] : it->second) {
      sensor_keys.push_back(key);
    }
  }
}

void SensorCollector::ReinitializeAndScheduleAllSensorsForConfigKey(
    const std::string& board_config_key) {
  auto it = sensor_table_.find(board_config_key);
  if (it == sensor_table_.end()) {
    LOG(WARNING) << "No sensors found for config key: " << board_config_key
                 << ". Skipping reinitialization.";
    return;
  }

  for (const auto& [key, sensor] : it->second) {
    sensor->Reinitialize([this, sensor, key](absl::Status status) {
      if (!status.ok()) {
        LOG(WARNING) << "Failed to reinitialize sensor: " << key
                     << " with status: " << status;
        return;
      }

      ScheduleIndividualSensorRead(sensor, refresh_notification_,
                                   override_sensor_sampling_interval_ms_,
                                   *thread_manager_);
    });
  }
}

void SensorCollector::StartCollection() {
  for (const auto& io_context : thread_manager_->io_contexts) {
    thread_manager_->threads.push_back(
        thread_factory_->New([io_context]() { io_context->run(); }));
  }
  if (enable_threshold_monitoring_) {
    ScheduleThresholdMonitoring();
  }
}

std::shared_ptr<const Sensor> SensorCollector::GetSensorByConfigKeyAndSensorKey(
    const std::string& board_config_key, const std::string& sensor_key) const {
  auto board_it = sensor_table_.find(board_config_key);
  if (board_it == sensor_table_.end()) {
    return nullptr;
  }
  auto sensor_it = board_it->second.find(sensor_key);
  if (sensor_it == board_it->second.end()) {
    return nullptr;
  }
  return sensor_it->second;
}

void SensorCollector::GetAllSensors(
    std::vector<std::shared_ptr<const Sensor>>& all_sensors) const {
  for (const auto& [_, key_to_sensor] : sensor_table_) {
    for (const auto& [_, sensor] : key_to_sensor) {
      all_sensors.push_back(sensor);
    }
  }
}

std::shared_ptr<const Sensor> SensorCollector::GetSensorBySensorKey(
    const std::string& sensor_key) const {
  for (const auto& [_, key_to_sensor] : sensor_table_) {
    auto it = key_to_sensor.find(sensor_key);
    if (it != key_to_sensor.end()) {
      return it->second;
    }
  }
  return nullptr;
}

absl::Status SensorCollector::WriteToSensor(const std::string& sensor_key,
                                            const SensorValue& value) {
  for (const auto& [_, key_to_sensor] : sensor_table_) {
    auto it = key_to_sensor.find(sensor_key);
    if (it != key_to_sensor.end()) {
      return it->second->WriteReading(value);
    }
  }
  return absl::NotFoundError(absl::StrCat("Sensor not found: ", sensor_key));
}

absl::Status SensorCollector::SetDevpathForSensor(absl::string_view sensor_key,
                                                  absl::string_view devpath) {
  for (const auto& [_, key_to_sensor] : sensor_table_) {
    auto it = key_to_sensor.find(sensor_key);
    if (it != key_to_sensor.end()) {
      return it->second->SetDevpath(devpath);
    }
  }
  return absl::NotFoundError(absl::StrCat("Sensor not found: ", sensor_key));
}

absl::Status SensorCollector::ConfigureCollection(std::string_view sensor_key,
                                                  Mutation mutation) const {
  for (auto& [_, key_to_sensor] : sensor_table_) {
    auto it = key_to_sensor.find(sensor_key);
    if (it != key_to_sensor.end()) {
      const std::shared_ptr<Sensor>& sensor = it->second;
      const Status status =
          sensor->GetSensorAttributesDynamic().state().status();
      if (IsCreationIncomplete(status)) {
        LOG(WARNING) << "Skipping configuration for sensor " << sensor_key
                     << " because sensor status is: " << Status_Name(status);
        return absl::OkStatus();
      }

      if (mutation.interval.has_value()) {
        int sampling_interval_ms = mutation.interval.value();
        absl::Duration interval;
        if (sampling_interval_ms > 0) {
          interval = absl::Milliseconds(sampling_interval_ms);
        } else {
          // If the sampling interval is not set in the config, then we use the
          // static refresh interval of the sensor if it is set. Otherwise, we
          // use the default sensor sampling interval.
          absl::Duration static_refresh_interval = DecodeGoogleApiProto(
              sensor->GetSensorAttributesStatic().static_refresh_interval());
          if (static_refresh_interval > absl::ZeroDuration()) {
            LOG(INFO) << "Using static refresh interval: "
                      << static_refresh_interval
                      << " for sensor: " << sensor->GetKey();
            interval = static_refresh_interval;
          } else {
            LOG(INFO) << "Using default sensor sampling interval: "
                      << kDefaultSensorSamplingInterval
                      << " for sensor: " << sensor->GetKey();
            interval = kDefaultSensorSamplingInterval;
          }
        }

        {
          absl::MutexLock lock(thread_manager_->key_to_task_id_mutex);
          thread_manager_->task_scheduler->UpdateTaskPeriod(
              thread_manager_->sensor_key_to_task_id.at(sensor_key), interval);
        }
        sensor->UpdateSensorSamplingInterval(interval);
      }

      if (mutation.batch_size.has_value()) {
        int batch_size = mutation.interval == kResetMonitoring
                             ? 0
                             : mutation.batch_size.value();
        ResizeBufferAndMetrics(sensor, batch_size);
      }
      return absl::OkStatus();
    }
  }

  return absl::NotFoundError("No sensor found for configuration");
}

std::unique_ptr<MonitoringChangeBase>
SensorCollector::CreateMonitoringChange() {
  return std::make_unique<SensorCollectorMonitoringChange>(this);
}

void SensorCollector::ReinitializeSensorByConfigKeyAndSensorKey(
    const std::string& board_config_key, const std::string& sensor_key) const {
  auto board_it = sensor_table_.find(board_config_key);
  if (board_it == sensor_table_.end()) {
    LOG(WARNING) << "No sensors found for config key: " << board_config_key
                 << ". Skipping reinitialization.";
    return;
  }

  auto sensor_it = board_it->second.find(sensor_key);
  if (sensor_it == board_it->second.end()) {
    LOG(WARNING) << "Failed to find sensor " << sensor_key << " in config "
                 << board_config_key;
    return;
  }

  sensor_it->second->Reinitialize(
      [this, sensor_it, sensor_key](absl::Status status) {
        if (!status.ok()) {
          LOG(WARNING) << "Failed to reinitialize sensor: " << sensor_key
                       << " with status: " << status;
          return;
        }

        ScheduleIndividualSensorRead(sensor_it->second, refresh_notification_,
                                     override_sensor_sampling_interval_ms_,
                                     *thread_manager_);
      });
}

absl::StatusOr<std::unique_ptr<SensorCollector>> SensorCollector::Create(
    const Params& params) {
  std::vector<std::shared_ptr<Sensor>> sensors;
  std::vector<std::shared_ptr<RedfishAggregatedBatch>>
      redfish_aggregated_batches;
  auto thread_manager = std::make_unique<ThreadManager>(params.clock);
  absl::flat_hash_map<std::string, std::shared_ptr<boost::asio::io_context>>
      sensor_group_to_io_context;

  ECCLESIA_RETURN_IF_ERROR(
      CreateHwmonSensors(params, sensors, sensor_group_to_io_context));

  ECCLESIA_RETURN_IF_ERROR(
      CreatePsuSensors(params, sensors, sensor_group_to_io_context));

  ECCLESIA_ASSIGN_OR_RETURN(
      std::vector<std::shared_ptr<FanController>> fan_controllers,
      CreateFanControllers(params));

  ECCLESIA_RETURN_IF_ERROR(CreateFanSensors(params, fan_controllers, sensors,
                                            sensor_group_to_io_context));

  ECCLESIA_RETURN_IF_ERROR(
      CreateSharedMemSensors(params, sensors, *thread_manager));

  ECCLESIA_RETURN_IF_ERROR(
      CreateIntelCpuSensors(params, sensors, sensor_group_to_io_context));

  ECCLESIA_RETURN_IF_ERROR(
      CreateAdcSensors(params, sensors, sensor_group_to_io_context));

  ECCLESIA_RETURN_IF_ERROR(CreateRedfishAggregatedBatchSensors(
      params, sensors, redfish_aggregated_batches, sensor_group_to_io_context));

  ECCLESIA_RETURN_IF_ERROR(CreateRedfishAggregatedSensors(
      params, sensors, sensor_group_to_io_context));

  absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>
      bus_to_v1_accessor;
  absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>
      bus_to_v2_accessor;
  absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>
      bus_to_v4_accessor;
  absl::flat_hash_map<int, std::unique_ptr<nic_veeprom::Accessor>>
      bus_to_v5_accessor;
  ECCLESIA_RETURN_IF_ERROR(CreateNicSensors(
      params, sensors, sensor_group_to_io_context, bus_to_v1_accessor,
      bus_to_v2_accessor, bus_to_v4_accessor, bus_to_v5_accessor));

  ECCLESIA_RETURN_IF_ERROR(
      CreateGpioSensors(params, sensors, sensor_group_to_io_context));

  // Create virtual sensors last to make sure all physical sensors can be used
  // in the formula for virtual sensors.
  ECCLESIA_RETURN_IF_ERROR(
      CreateVirtualSensors(params, sensors, sensor_group_to_io_context));

  // Must run after all sensors are created and before anything consumes the
  // sensor list (power callbacks, scheduling, the sensor table), so a dropped
  // sensor leaves no trace.
  if (GetTlbmcConfig()
          .sensor_collector_module()
          .drop_failed_duplicate_sensors()) {
    DropFailedDuplicateSensors(sensors);
  }

  ScheduleNicSensorAccessorRefresh(bus_to_v1_accessor, bus_to_v2_accessor,
                                   bus_to_v4_accessor, bus_to_v5_accessor,
                                   sensor_group_to_io_context, *thread_manager);

  for (const auto& [sensor_group, io_context] : sensor_group_to_io_context) {
    LOG(INFO) << "Starting io_context thread for sensor group: "
              << sensor_group;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
        work_guard(boost::asio::make_work_guard(*io_context));
    thread_manager->work_guards.push_back(std::move(work_guard));
    thread_manager->io_contexts.push_back(io_context);
  }

  for (const auto& sensor : sensors) {
    const EntityCommonConfig& entity_common_config =
        sensor->GetSensorAttributesStatic().entity_common_config();
    if (entity_common_config.power_state() ==
        SensorPowerRequirement::SENSOR_POWER_REQUIREMENT_ALWAYS) {
      continue;
    }
    if (params.power_control_collector == nullptr) {
      continue;
    }
    PowerControl* power_control =
        params.power_control_collector->GetPowerControl(
            entity_common_config.host_id());
    if (power_control == nullptr) {
      LOG(WARNING) << "No power control found for host: "
                   << entity_common_config.host_id();
      continue;
    }
    if (entity_common_config.power_state() ==
        SensorPowerRequirement::SENSOR_POWER_REQUIREMENT_CHASSIS_ON) {
      power_control->RegisterHostPowerOffToOnCallback(
          [sensor]([[maybe_unused]] bool setup_run) {
            sensor->PowerOffToOnCallback(setup_run);
          });
      power_control->RegisterHostPowerOnToOffCallback(
          [sensor]([[maybe_unused]] bool setup_run) {
            sensor->PowerOnToOffCallback(setup_run);
          });
    } else if (entity_common_config.power_state() ==
               SensorPowerRequirement::SENSOR_POWER_REQUIREMENT_BIOS_POST) {
      power_control->RegisterHostOSInactiveToStandbyCallback(
          [sensor]([[maybe_unused]] bool setup_run) {
            sensor->OSInactiveToStandbyCallback(setup_run);
          });
      power_control->RegisterHostOSStandbyToInactiveCallback(
          [sensor]([[maybe_unused]] bool setup_run) {
            sensor->OSStandbyToInactiveCallback(setup_run);
          });
    }
  }

  ScheduleAllSensorReads(sensors, params.refresh_notification,
                         params.override_sensor_sampling_interval_ms,
                         *thread_manager);
  ScheduleAllBatchSensorReads(
      redfish_aggregated_batches, params.refresh_notification,
      params.override_sensor_sampling_interval_ms, *thread_manager);

  // Create the SensorCollector instance first.
  std::unique_ptr<SensorCollector> collector(new SensorCollector(
      std::move(sensors), std::move(bus_to_v1_accessor),
      std::move(bus_to_v2_accessor), std::move(bus_to_v4_accessor),
      std::move(bus_to_v5_accessor), std::move(thread_manager),
      params.refresh_notification, params.override_sensor_sampling_interval_ms,
      params.thread_factory,
      /*peci_scanner=*/nullptr, std::move(redfish_aggregated_batches),
      params.enable_threshold_monitoring));

  if (!params.sensor_configs.intel_cpu_sensor_configs.empty()) {
    // Initialize the PeciScanner after the SensorCollector is created,
    // so the callback can capture 'this'.
    collector->InitializePeciScanner(params);
    collector->peci_scanner_->InitializeScanContexts();
  }

  return collector;
}

void SensorCollector::InitializePeciScanner(const Params& params) {
  peci_scanner_ = PeciScanner::Create(
      params.sensor_configs.intel_cpu_sensor_configs, params.peci_sysfs,
      params.peci_access, thread_manager_->task_scheduler.get(),
      [this](const std::string& board_config_key,
             const std::string& sensor_key) {
        this->ReinitializeSensorByConfigKeyAndSensorKey(board_config_key,
                                                        sensor_key);
      });
}

void SensorCollector::TriggerPeciScan() {
  if (peci_scanner_ != nullptr) {
    peci_scanner_->TriggerScan();
  }
}

void SensorCollector::ScheduleThresholdMonitoring() {
  thread_manager_->task_scheduler->RunAndScheduleAsync(
      [this](absl::AnyInvocable<void()> on_done) {
        for (const auto& sensor : sensors_with_thresholds_) {
          if (!sensor->IsReadyForRead() ||
              sensor->GetSensorAttributesDynamic().state().status() ==
                  STATUS_STALE) {
            continue;
          }
          std::vector<Sensor::ThresholdEvent> events =
              sensor->CheckSensorThresholds();
          if (events.empty()) {
            continue;
          }
          std::shared_ptr<const SensorValue> sensor_data =
              sensor->GetSensorData();
          if (sensor_data == nullptr) {
            continue;
          }
          for (const Sensor::ThresholdEvent& event : events) {
            PublishSensorThresholdEvent(sensor, sensor_data, event);
          }
        }
        on_done();
      },
      absl::Seconds(1));
}

void SensorCollector::PublishSensorThresholdEvent(
    const std::shared_ptr<const Sensor>& sensor,
    const std::shared_ptr<const SensorValue>& sensor_data,
    const Sensor::ThresholdEvent& threshold_event) {
  const SensorAttributesStatic& static_attributes =
      sensor->GetSensorAttributesStatic();

  std::optional<gbmc_sel_framework::EventSourceComponent>
      event_source_component =
          GetEventSourceComponent(static_attributes.unit());
  if (!event_source_component.has_value()) {
    // Not supported sensor unit. silent return.
    return;
  }

  std::optional<absl::string_view> threshold_event_str =
      GetThresholdEventStr(threshold_event.threshold_type);
  std::string sensor_name =
      GetTrimmedSensorName(static_attributes.attributes().key());
  if (!threshold_event_str.has_value()) {
    LOG(ERROR) << "Failed to find threshold event for threshold type "
               << threshold_event.threshold_type << " for sensor "
               << sensor_name;
    return;
  }

  gbmc_sel_framework::EventSeverity event_severity =
      GetEventSeverity(threshold_event.threshold_type);
  gbmc_sel_framework::EventAction event_action =
      threshold_event.is_assert
          ? gbmc_sel_framework::EventAction::EVENT_ACTION_ASSERT
          : gbmc_sel_framework::EventAction::EVENT_ACTION_DEASSERT;

  std::string event_name = absl::StrCat(
      "Sensor ", sensor_name, " reading ", sensor_data->reading(), " ",
      gbmc_sel_framework::ToString(event_action), " ", *threshold_event_str);

  absl::StatusOr<const std::string&> devpath = sensor->GetDevpath();
  absl::string_view devpath_str;
  if (devpath.ok()) {
    devpath_str = *devpath;
  }
  gbmc_sel_framework::EventSourceType event_source_type =
      GetEventSourceType(devpath_str);

  std::vector<std::string> additional_data = BuildAdditionalData(
      sensor_name, sensor_data->reading(), *threshold_event_str,
      devpath.ok() ? std::make_optional(*devpath) : std::nullopt);

  std::vector<absl::string_view> additional_data_view;
  additional_data_view.reserve(additional_data.size());
  for (const std::string& data : additional_data) {
    additional_data_view.push_back(data);
  }

  absl::Status status;
  if (journal_sender_ != nullptr) {
    status = gbmc_sel_framework::PublishGBMCSystemEvent(
        event_source_type, *event_source_component, event_severity,
        event_action, event_name, additional_data_view,
        [this](const struct iovec* iov, int iovcnt) {
          return journal_sender_(iov, iovcnt);
        });
  } else {
    status = gbmc_sel_framework::PublishGBMCSystemEvent(
        event_source_type, *event_source_component, event_severity,
        event_action, event_name, additional_data_view);
  }
  if (!status.ok()) {
    // use info log level to avoid flooding the log with threshold events.
    LOG(INFO) << "Failed to publish sensor threshold event: " << status;
  }
}

std::unique_ptr<EmptySensorCollector> EmptySensorCollector::Create() {
  return std::make_unique<EmptySensorCollector>();
}

// Returns the list of sensor names contained by the given config key.
void EmptySensorCollector::GetAllSensorKeysByConfigKey(
    const std::string& board_config_key,
    std::vector<std::string>& sensor_keys) const {
  LOG(WARNING) << "EmptySensorCollector::GetAllSensorKeysByConfigKey is "
                  "called. This will return an empty list.";
}

// Returns all the sensors.
void EmptySensorCollector::GetAllSensors(
    std::vector<std::shared_ptr<const Sensor>>& sensors) const {
  LOG(WARNING) << "EmptySensorCollector::GetAllSensorKeysByConfigKey is "
                  "called. This will return an empty list.";
}

// Returns the sensor for the given sensor key.
std::shared_ptr<const Sensor> EmptySensorCollector::GetSensorBySensorKey(
    const std::string& sensor_key) const {
  LOG(WARNING) << "EmptySensorCollector::GetSensorBySensorKey is "
                  "called. This will return a nullptr.";
  return nullptr;
}

std::unique_ptr<MonitoringChangeBase>
EmptySensorCollector::CreateMonitoringChange() {
  return std::make_unique<FailingMonitoringChange>();
}

absl::Status EmptySensorCollector::ConfigureCollection(
    std::string_view sensor_key, Mutation mutation) const {
  return absl::UnimplementedError(
      "EmptySensorCollector::ConfigureCollection is called. This is not "
      "implemented.");
}

std::shared_ptr<const Sensor>
EmptySensorCollector::GetSensorByConfigKeyAndSensorKey(
    const std::string& board_config_key, const std::string& sensor_key) const {
  LOG(WARNING) << "EmptySensorCollector::GetSensorByConfigKeyAndSensorKey is "
                  "called. This will return a nullptr.";
  return nullptr;
}

absl::Status EmptySensorCollector::WriteToSensor(const std::string& sensor_key,
                                                 const SensorValue& value) {
  LOG(WARNING) << "EmptySensorCollector::WriteToSensor is "
                  "called. This will return an error.";
  return absl::UnimplementedError(
      "EmptySensorCollector::WriteToSensor is called. This is not "
      "implemented.");
}

absl::Status EmptySensorCollector::SetDevpathForSensor(
    absl::string_view sensor_key, absl::string_view devpath) {
  LOG(WARNING) << "EmptySensorCollector::SetDevpathForSensor is "
                  "called. This will return an error.";
  return absl::UnimplementedError(
      "EmptySensorCollector::SetDevpathForSensor is called. This is not "
      "implemented.");
}

nlohmann::json EmptySensorCollector::GetSchedulerStats() const {
  return nlohmann::json::parse("{\"Warning\": \"EmptySensorCollector used.\"}");
}

nlohmann::json EmptySensorCollector::ToJson() const {
  return nlohmann::json::parse("{\"Warning\": \"EmptySensorCollector used.\"}");
}

void EmptySensorCollector::StartCollection() {
  LOG(WARNING) << "EmptySensorCollector::StartCollection is "
                  "called. This does nothing.";
}

}  // namespace milotic_tlbmc