MetricPlugin.hh

// MetricPlugin.hh: Metric Plugin Interface
// Author: Eric Flumerfelt
// Last Modified: 11/05/2014 (Created)
//
// Defines the interface that any ARTDAQ metric plugin must implement

#ifndef __METRIC_INTERFACE__
#define __METRIC_INTERFACE__

#include <chrono>
#include <string>
#include <unordered_map>
#include "fhiclcpp/ParameterSet.h"
#include "fhiclcpp/types/Atom.h"
#include "fhiclcpp/types/ConfigurationTable.h"
#include "fhiclcpp/types/Sequence.h"

#include "artdaq-utilities/Plugins/MetricData.hh"
#include "cetlib/compiler_macros.h"
#ifndef FALLTHROUGH
#define FALLTHROUGH while (0)
#endif

namespace artdaq {
class MetricPlugin
{
public:
    struct Config
    {
        fhicl::Atom<std::string> metricPluginType{fhicl::Name{"metricPluginType"}, fhicl::Comment{"The name of the metric plugin to load (may have additional configuration parameters"}};
        fhicl::Atom<size_t> level{fhicl::Name{"level"}, fhicl::Comment{"The verbosity level threshold for this plugin. sendMetric calls with verbosity level greater than this will not be sent to the plugin. OPTIONAL"}, 0};
        fhicl::Sequence<size_t> metric_levels{fhicl::Name{"metric_levels"}, fhicl::Comment{"A list of levels that should be enabled for this plugin. OPTIONAL"}, std::vector<size_t>()};
        fhicl::Atom<std::string> level_string{fhicl::Name{"level_string"}, fhicl::Comment{"A string containing a comma-separated list of levels to enable. Ranges are supported. Example: \"1,2,4-10,11\" OPTIONAL"}, ""};
        fhicl::Atom<double> reporting_interval{fhicl::Name{"reporting_interval"}, fhicl::Comment{"How often recorded metrics are sent to the underlying metric storage"}, 15.0};
    };
    using Parameters = fhicl::WrappedTable<Config>;

    explicit MetricPlugin(fhicl::ParameterSet const& ps, std::string const& app_name)
        : pset(ps)
        , app_name_(app_name)
        , inhibit_(false)
        , level_mask_(0ULL)
    {
        if (pset.has_key("level"))
        {
            for (size_t ii = 0; ii <= pset.get<size_t>("level"); ++ii)
            {
                level_mask_[ii] = true;
            }
        }
        if (pset.has_key("metric_levels"))
        {
            auto levels = pset.get<std::vector<size_t>>("metric_levels");
            for (auto& l : levels)
            {
                level_mask_[l] = true;
            }
        }
        if (pset.has_key("level_string"))
        {
            auto string = pset.get<std::string>("level_string");
            std::stringstream ss(string);
            std::string token;
            while (std::getline(ss, token, ','))
            {
                auto it = token.find("-");
                if (it == 0 || it == token.size() - 1) continue;

                if (it != std::string::npos)
                {
                    auto minStr = token.substr(0, it);
                    auto maxStr = token.substr(it + 1);
                    auto min = std::stoi(minStr);
                    auto max = std::stoi(maxStr);

                    if (min > max) std::swap(min, max);
                    if (min > 63) min = 63;
                    if (max > 63) max = 63;

                    for (int ii = min; ii <= max; ++ii)
                    {
                        level_mask_[ii] = true;
                    }
                }
                else
                {
                    auto level = std::stoi(token);
                    if (level >= 0 && level < 63) level_mask_[level] = true;
                }
            }
        }
        if (level_mask_.to_ullong() == 0)
        {
            throw cet::exception("Configuration Error") << "No levels were enabled for this plugin! Please specify at least one of the following Parameters: \"level\", \"metric_levels\", or \"level_string\"!";
        }
        accumulationTime_ = pset.get<double>("reporting_interval", 15.0);
    }

    virtual ~MetricPlugin() = default;

    //
    // Interface Functions: These should be reimplemented in plugin classes!
    //

    virtual std::string getLibName() const { return "ERROR"; }

protected:
    virtual void sendMetric_(const std::string& name, const std::string& value, const std::string& unit) = 0;

    virtual void sendMetric_(const std::string& name, const int& value, const std::string& unit) = 0;

    virtual void sendMetric_(const std::string& name, const double& value, const std::string& unit) = 0;

    virtual void sendMetric_(const std::string& name, const float& value, const std::string& unit) = 0;

    virtual void sendMetric_(const std::string& name, const long unsigned int& value, const std::string& unit) = 0;

    virtual void startMetrics_() = 0;

    virtual void stopMetrics_() = 0;

    //
    // Implementation Functions: These should be called from ARTDAQ code!
    //
public:
    void addMetricData(std::unique_ptr<MetricData> const& data)
    {
        if (data->Type == MetricType::StringMetric)
        {
            sendMetric_(data->Name, data->StringValue, data->Unit);
        }
        else
        {
            if (!metricRegistry_.count(data->Name))
            {
                metricRegistry_[data->Name] = *data;
            }
            metricData_[data->Name].push_back(*data);
            //sendMetrics();
        }
    }

    void sendMetrics(bool forceSend = false,
                     std::chrono::steady_clock::time_point interval_end = std::chrono::steady_clock::now())
    {
        for (auto metric : metricData_)
        {
            auto* metricName = &metric.first;
            if (readyToSend_(*metricName) || forceSend)
            {
                if (metricData_[*metricName].size() == 0 && metricRegistry_.count(*metricName))
                {
                    sendZero_(metricRegistry_[*metricName]);
                }
                else if (metricData_[*metricName].size() > 0)
                {
                    MetricData& data = metricData_[*metricName].front();
                    auto it = ++(metricData_[*metricName].begin());
                    while (it != metricData_[*metricName].end())
                    {
                        data.Add(*it);
                        it = metricData_[*metricName].erase(it);
                    }

                    std::bitset<32> modeSet(static_cast<uint32_t>(data.Mode));
                    bool useSuffix = true;
                    if (modeSet.count() <= 1) useSuffix = false;

                    if ((data.Mode & MetricMode::LastPoint) != MetricMode::None)
                    {
                        sendMetric_(data.Name + (useSuffix ? " - Last" : ""), data.Last, data.Unit, data.Type);
                    }
                    if ((data.Mode & MetricMode::Accumulate) != MetricMode::None)
                    {
                        sendMetric_(data.Name + (useSuffix ? " - Total" : ""), data.Value, data.Unit, data.Type);
                    }
                    if ((data.Mode & MetricMode::Average) != MetricMode::None)
                    {
                        double average = 0.0;
                        switch (data.Type)
                        {
                            case MetricType::DoubleMetric:
                                average = data.Value.d / static_cast<double>(data.DataPointCount);
                                        break;
                            case MetricType::FloatMetric:
                                average = data.Value.f / static_cast<double>(data.DataPointCount);
                                        break;
                            case MetricType::IntMetric:
                                average = data.Value.i / static_cast<double>(data.DataPointCount);
                                break;
                            case MetricType::UnsignedMetric:
                                average = data.Value.u / static_cast<double>(data.DataPointCount);
                                        break;
                                    default:
                                        break;
                                }
                        sendMetric_(data.Name + (useSuffix ? " - Average" : ""), average, data.Unit);
                    }
                    if ((data.Mode & MetricMode::Rate) != MetricMode::None)
                    {
                        double duration = std::chrono::duration_cast<std::chrono::duration<double, std::ratio<1>>>(
                                                      interval_end - interval_start_[*metricName])
                                                      .count();
                        double rate = 0.0;
                        switch (data.Type)
                        {
                            case MetricType::DoubleMetric:
                                rate = data.Value.d / duration;
                                        break;
                            case MetricType::FloatMetric:
                                rate = data.Value.f / duration;
                                        break;
                            case MetricType::IntMetric:
                                rate = data.Value.i / duration;
                                        break;
                            case MetricType::UnsignedMetric:
                                rate = data.Value.u / duration;
                                        break;
                                    default:
                                        break;
                                }
                        sendMetric_(data.Name + (useSuffix ? " - Rate" : ""), rate, data.Unit + "/s");
                    }
                    if ((data.Mode & MetricMode::Minimum) != MetricMode::None)
                    {
                        sendMetric_(data.Name + (useSuffix ? " - Min" : ""), data.Min, data.Unit, data.Type);
                    }
                    if ((data.Mode & MetricMode::Maximum) != MetricMode::None)
                    {
                        sendMetric_(data.Name + (useSuffix ? " - Max" : ""), data.Max, data.Unit, data.Type);
                        }

                        metricData_[*metricName].clear();
                    }
                interval_start_[*metricName] = interval_end;
            }
        }
    }

    void startMetrics() { startMetrics_(); }

    void stopMetrics()
    {
        inhibit_ = true;
        sendMetrics(true);
        for (auto metric : metricRegistry_)
        {
            sendZero_(metric.second);
        }
        stopMetrics_();
        inhibit_ = false;
    }

    bool IsLevelEnabled(int level)
    {
        if (level > 63) level = 63;
        if (level < 0) return true;
        return level_mask_[level];
    }

protected:
    fhicl::ParameterSet pset;     
    double accumulationTime_;     
    std::string app_name_;        
    bool inhibit_;                
    std::bitset<64> level_mask_;  

private:
    std::unordered_map<std::string, std::list<MetricData>> metricData_;
    std::unordered_map<std::string, MetricData> metricRegistry_;
    std::unordered_map<std::string, std::chrono::steady_clock::time_point> lastSendTime_;
    std::unordered_map<std::string, std::chrono::steady_clock::time_point> interval_start_;

    bool readyToSend_(std::string name)
    {
        auto now = std::chrono::steady_clock::now();
        if (std::chrono::duration_cast<std::chrono::duration<double, std::ratio<1>>>(now - lastSendTime_[name]).count() >=
            accumulationTime_)
        {
            lastSendTime_[name] = now;
            return true;
        }

        return false;
    }

    void sendZero_(MetricData data)
    {
        std::bitset<32> modeSet(static_cast<uint32_t>(data.Mode));
        bool useSuffix = true;
        if (modeSet.count() <= 1) useSuffix = false;

        MetricData::MetricDataValue zero;
        switch (data.Type)
        {
            case MetricType::DoubleMetric:
                zero.d = 0.0;
                break;
            case MetricType::FloatMetric:
                zero.f = 0.0f;
                break;
            case MetricType::IntMetric:
                zero.i = 0;
                break;
            case MetricType::UnsignedMetric:
                zero.u = 0;
                break;
            default:
                break;
        }

        if ((data.Mode & MetricMode::LastPoint) != MetricMode::None)
        {
            sendMetric_(data.Name + (useSuffix ? " - Last" : ""), zero, data.Unit, data.Type);
        }
        if ((data.Mode & MetricMode::Accumulate) != MetricMode::None)
        {
            sendMetric_(data.Name + (useSuffix ? " - Total" : ""), zero, data.Unit, data.Type);
        }
        if ((data.Mode & MetricMode::Average) != MetricMode::None)
        {
            sendMetric_(data.Name + (useSuffix ? " - Average" : ""), 0.0, data.Unit);
        }
        if ((data.Mode & MetricMode::Rate) != MetricMode::None)
        {
            sendMetric_(data.Name + (useSuffix ? " - Rate" : ""), 0.0, data.Unit + "/s");
        }
        if ((data.Mode & MetricMode::Minimum) != MetricMode::None)
        {
            sendMetric_(data.Name + (useSuffix ? " - Min" : ""), zero, data.Unit, data.Type);
        }
        if ((data.Mode & MetricMode::Maximum) != MetricMode::None)
        {
            sendMetric_(data.Name + (useSuffix ? " - Max" : ""), zero, data.Unit, data.Type);
        }
    }

    void sendMetric_(std::string name, MetricData::MetricDataValue data, std::string unit, MetricType type)
    {
        switch (type)
        {
            case MetricType::DoubleMetric:
                sendMetric_(name, data.d, unit);
                break;
            case MetricType::FloatMetric:
                sendMetric_(name, data.f, unit);
                break;
            case MetricType::IntMetric:
                sendMetric_(name, data.i, unit);
                break;
            case MetricType::UnsignedMetric:
                sendMetric_(name, data.u, unit);
                break;
            default:
                break;
        }
    }
};
}  //End namespace artdaq

#endif  //End ifndef __METRIC_INTERFACE__