disruption_manager.cc

#include "disruption_manager.h"

#include <utility>

#include "callback_manager.h"
#include "daemon_context.h"
#include "absl/functional/any_invocable.h"
#include "absl/log/check.h"
#include "absl/log/log.h"
#include "absl/status/status.h"
#include "absl/strings/string_view.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"
#include "source_location"
#include "bmc/status_macros.h"

namespace safepower_agent {

static constexpr absl::string_view kEndCallbacksTaskName =
    "disruption_end_callbacks";
static constexpr absl::string_view kStartCallbacksTaskName =
    "disruption_start_callbacks";

absl::Status DisruptionManager::RunCallbacks(CallbackManager& callback_manager,
                                             absl::string_view name,
                                             absl::Duration delay,
                                             std::source_location location) {
  return DaemonContext::Get().scheduler().DelayCall(
      [&callback_manager, location = std::move(location), name]() {
        LOG(INFO).AtLocation(location.file_name(), location.line())
            << "Running " << name;
        callback_manager.RunCallbacks();
        LOG(INFO).AtLocation(location.file_name(), location.line())
            << "Done running " << name;
      },
      delay, name);
}

absl::Status DisruptionManager::CancelDisruption() {
  absl::MutexLock lock(disruption_mutex_);
  if (DaemonContext::Get().now() >= expect_disruption_until_) {
    return absl::FailedPreconditionError("Not expecting a disruption");
  }
  if (absl::Status status =
          DaemonContext::Get().scheduler().CancelCall(kEndCallbacksTaskName);
      !status.ok()) {
    LOG(DFATAL) << "Failed to cancel disruption end callbacks: " << status;
  } else if (absl::Status status =
                 RunCallbacks(disruption_end_callbacks_, kEndCallbacksTaskName,
                              absl::ZeroDuration());
             !status.ok()) {
    LOG(DFATAL) << "Failed to run disruption end callbacks: " << status;
  }
  LOG(INFO) << "Disruption canceled";
  expect_disruption_until_ = absl::InfinitePast();
  return absl::OkStatus();
}

absl::Status DisruptionManager::ExpectDisruptionIn(absl::Duration timeout) {
  absl::MutexLock lock(disruption_mutex_);
  if (expect_disruption_until_ >= DaemonContext::Get().now()) {
    return absl::FailedPreconditionError(
        "Disruption already expected by " +
        absl::FormatTime(expect_disruption_until_));
  }
  RETURN_IF_ERROR(
      RunCallbacks(disruption_end_callbacks_, kEndCallbacksTaskName, timeout));
  absl::Status status =
      RunCallbacks(disruption_start_callbacks_, kStartCallbacksTaskName,
                   absl::ZeroDuration());
  if (!status.ok()) {
    absl::Status cancel_status =
        DaemonContext::Get().scheduler().CancelCall(kStartCallbacksTaskName);
    if (!cancel_status.ok()) {
      LOG(DFATAL) << "Failed to cancel disruption start callbacks: "
                  << cancel_status;
    }
    return status;
  }
  expect_disruption_until_ = DaemonContext::Get().now() + timeout;
  LOG(INFO) << "Disruption expected by "
            << absl::FormatTime(expect_disruption_until_);
  return absl::OkStatus();
}

absl::Duration DisruptionManager::PendingDisruption() const {
  absl::MutexLock lock(disruption_mutex_);
  absl::Time now = DaemonContext::Get().now();
  if (expect_disruption_until_ > now) {
    return expect_disruption_until_ - now;
  }
  return absl::ZeroDuration();
}

DisruptionManager::CallbackHandle DisruptionManager::OnDisruptionStart(
    DisruptionCallback callback) {
  return disruption_start_callbacks_.RunLast(std::move(callback));
}

DisruptionManager::CallbackHandle DisruptionManager::OnDisruptionEnd(
    DisruptionCallback callback) {
  return disruption_end_callbacks_.RunFirst(std::move(callback));
}

}  // namespace safepower_agent