MonitorableObject.cpp

Go to the documentation of this file.

 
#include "logging/Logging.hpp"
#include <NullOpMonFacility.hpp>
#include <opmonlib/MonitorableObject.hpp>
#include <opmonlib/Utils.hpp>
 
#include <google/protobuf/util/time_util.h>
 
#include <chrono>

#define TRACE_NAME "MonitorableObject" // NOLINT
enum
{
  TLVL_MONITORING_STEPS = 10,
  TLVL_LEVEL_SUPPRESSION = 20,
};
 
using namespace dunedaq::opmonlib;
 
std::shared_ptr<OpMonFacility> MonitorableObject::s_default_facility = std::make_shared<NullOpMonFacility>();
 
void
MonitorableObject::register_node(ElementId name, NewNodePtr p)
{
 
  std::lock_guard<std::mutex> lock(m_node_mutex);
 
  // check if the name is already present to ensure uniqueness
  auto it = m_nodes.find(name);
  if (it != m_nodes.end()) {
    // This not desired because names are suppposed to be unique
    // But if the pointer is expired, there is no harm in override it
    if (it->second.expired()) {
      ers::warning(NonUniqueNodeName(ERS_HERE, name, to_string(get_opmon_id())));
    } else {
      throw NonUniqueNodeName(ERS_HERE, name, to_string(get_opmon_id()));
    }
  }
 
  m_nodes[name] = p;
 
  p->m_opmon_name = name;
  p->inherit_parent_properties(*this);
 
  TLOG() << "Node " << name << " registered to " << to_string(get_opmon_id());
}
 
void
MonitorableObject::publish(google::protobuf::Message&& m, CustomOrigin&& co, OpMonLevel l) const noexcept
{
 
  auto timestamp = google::protobuf::util::TimeUtil::GetCurrentTime();
 
  auto start_time = std::chrono::high_resolution_clock::now();
 
  if (!MonitorableObject::publishable_metric(l, get_opmon_level())) {
    TLOG_DEBUG(TLVL_LEVEL_SUPPRESSION) << "Metric " << m.GetTypeName() << " ignored because of the level";
    ++m_ignored_counter;
    return;
  }
 
  auto e = to_entry(m, co);
 
  if (e.data().empty()) {
    ers::warning(EntryWithNoData(ERS_HERE, e.measurement()));
    return;
  }
 
  *e.mutable_origin() = get_opmon_id();
 
  *e.mutable_time() = timestamp;
 
  // this pointer is always garanteed to be filled, even if with a null Facility.
  // But the facility can fail
  try {
    m_facility.load()->publish(std::move(e));
    ++m_published_counter;
  } catch (const OpMonPublishFailure& e) {
    ers::error(e);
    ++m_error_counter;
  }
 
  auto stop_time = std::chrono::high_resolution_clock::now();
 
  auto duration = std::chrono::duration_cast<std::chrono::microseconds>(stop_time - start_time);
  m_cpu_us_counter += duration.count();
}
 
opmon::MonitoringTreeInfo
MonitorableObject::collect() noexcept
{
 
  auto start_time = std::chrono::high_resolution_clock::now();
 
  TLOG_DEBUG(TLVL_MONITORING_STEPS) << "Collecting data from " << to_string(get_opmon_id());
 
  opmon::MonitoringTreeInfo info;
 
  info.set_n_invalid_links(0);
 
  try {
    generate_opmon_data();
  } catch (const ers::Issue& i) {
    ++m_error_counter;
    auto cause_ptr = i.cause();
    while (cause_ptr) {
      ++m_error_counter;
      cause_ptr = cause_ptr->cause();
    }
    ers::error(ErrorWhileCollecting(ERS_HERE, to_string(get_opmon_id()), i));
  } catch (const std::exception& e) {
    ++m_error_counter;
    ers::error(ErrorWhileCollecting(ERS_HERE, to_string(get_opmon_id()), e));
  } catch (...) {
    ++m_error_counter;
    ers::error(ErrorWhileCollecting(ERS_HERE, to_string(get_opmon_id())));
  }
 
  info.set_n_published_measurements(m_published_counter.exchange(0));
  info.set_n_ignored_measurements(m_ignored_counter.exchange(0));
  info.set_n_errors(m_error_counter.exchange(0));
  if (info.n_published_measurements() > 0) {
    info.set_n_publishing_nodes(1);
  }
  info.set_cpu_elapsed_time_us(m_cpu_us_counter.exchange(0));
 
  std::lock_guard<std::mutex> lock(m_node_mutex);
 
  info.set_n_registered_nodes(m_nodes.size());
 
  unsigned int n_invalid_links = 0;
 
  for (auto it = m_nodes.begin(); it != m_nodes.end();) {
 
    auto ptr = it->second.lock();
 
    if (ptr) {
      auto child_info = ptr->collect(); // MR: can we make this an async? There is no point to wait all done here
      info.set_n_registered_nodes(info.n_registered_nodes() + child_info.n_registered_nodes());
      info.set_n_publishing_nodes(info.n_publishing_nodes() + child_info.n_publishing_nodes());
      info.set_n_invalid_links(info.n_invalid_links() + child_info.n_invalid_links());
      info.set_n_published_measurements(info.n_published_measurements() + child_info.n_published_measurements());
      info.set_n_ignored_measurements(info.n_ignored_measurements() + child_info.n_ignored_measurements());
      info.set_n_errors(info.n_errors() + child_info.n_errors());
      info.set_cpu_elapsed_time_us(info.cpu_elapsed_time_us() + child_info.cpu_elapsed_time_us());
    }
 
    // prune the dead links
    if (it->second.expired()) {
      it = m_nodes.erase(it);
      ++n_invalid_links;
    } else {
      ++it;
    }
  }
 
  info.set_n_invalid_links(info.n_invalid_links() + n_invalid_links);
 
  auto stop_time = std::chrono::high_resolution_clock::now();
 
  auto duration = std::chrono::duration_cast<std::chrono::microseconds>(stop_time - start_time);
  info.set_clockwall_elapsed_time_us(duration.count());
 
  return info;
}
 
void
MonitorableObject::set_opmon_level(OpMonLevel l) noexcept
{
 
  m_opmon_level = l;
 
  std::lock_guard<std::mutex> lock(m_node_mutex);
  for (const auto& [key, wp] : m_nodes) {
    auto p = wp.lock();
    if (p) {
      p->set_opmon_level(l);
    }
  }
}
 
void
MonitorableObject::inherit_parent_properties(const MonitorableObject& parent)
{
 
  m_facility.store(parent.m_facility);
  m_parent_id = parent.get_opmon_id();
  m_opmon_level = parent.get_opmon_level();
 
  std::lock_guard<std::mutex> lock(m_node_mutex);
 
  for (const auto& [key, wp] : m_nodes) {
 
    auto p = wp.lock();
    if (p) {
      p->inherit_parent_properties(*this);
    }
  }
}