Metric Class Reference

#include "Metric.h"

Public Member Functions
	Metric ()
	~Metric ()
void	Initialize (std::string const &realmName, Acore::Asio::IoContext &ioContext, std::function< void()> overallStatusLogger)
void	LoadFromConfigs ()
void	Update ()
bool	ShouldLog (std::string const &category, int64 value) const
template<class T >
void	LogValue (std::string const &category, T value, std::vector< MetricTag > tags)
void	LogEvent (std::string const &category, std::string const &title, std::string const &description)
void	Unload ()
bool	IsEnabled () const

Static Public Member Functions
static Metric *	instance ()

Private Member Functions
std::iostream &	GetDataStream ()
bool	Connect ()
void	SendBatch ()
void	ScheduleSend ()
void	ScheduleOverallStatusLog ()

Static Private Member Functions
static std::string	FormatInfluxDBValue (bool value)
template<class T >
static std::string	FormatInfluxDBValue (T value)
static std::string	FormatInfluxDBValue (std::string const &value)
static std::string	FormatInfluxDBValue (char const *value)
static std::string	FormatInfluxDBValue (double value)
static std::string	FormatInfluxDBValue (float value)
static std::string	FormatInfluxDBValue (std::chrono::nanoseconds value)
static std::string	FormatInfluxDBTagValue (std::string const &value)

Private Attributes
std::unique_ptr< std::iostream >	_dataStream
MPSCQueue< MetricData >	_queuedData
std::unique_ptr< Acore::Asio::DeadlineTimer >	_batchTimer
std::unique_ptr< Acore::Asio::DeadlineTimer >	_overallStatusTimer
int32	_updateInterval = 0
int32	_overallStatusTimerInterval = 0
bool	_enabled = false
bool	_overallStatusTimerTriggered = false
std::string	_hostname
std::string	_port
std::string	_databaseName
std::function< void()>	_overallStatusLogger
std::string	_realmName
std::unordered_map< std::string, int64 >	_thresholds

Detailed Description

Constructor & Destructor Documentation

Metric()

Metric::Metric

(

)

Todo:

: should format TagKey and FieldKey too in the same way as TagValue

{
}

~Metric()

Metric::~Metric

(

)

{
}

Member Function Documentation

Connect()

bool Metric::Connect ( )

private

{
    auto& stream = static_cast<boost::asio::ip::tcp::iostream&>(GetDataStream());
    stream.connect(_hostname, _port);
 
    auto error = stream.error();
    if (error)
    {
        LOG_ERROR("metric", "Error connecting to '{}:{}', disabling Metric. Error message: {}",
            _hostname, _port, error.message());
 
        _enabled = false;
        return false;
    }
 
    stream.clear();
    return true;
}

References _enabled, _hostname, _port, GetDataStream(), and LOG_ERROR.

Referenced by LoadFromConfigs(), and SendBatch().

FormatInfluxDBTagValue()

std::string Metric::FormatInfluxDBTagValue ( std::string const &  value )

staticprivate

Todo:

: should handle '=' and ',' characters too

{
    return boost::replace_all_copy(value, " ", "\\ ");
}

Referenced by Initialize(), and SendBatch().

FormatInfluxDBValue() [1/7]

std::string Metric::FormatInfluxDBValue ( bool  value )

staticprivate

{
    return value ? "t" : "f";
}

Referenced by FormatInfluxDBValue().

FormatInfluxDBValue() [2/7]

std::string Metric::FormatInfluxDBValue ( char const *  value )

staticprivate

{
    return FormatInfluxDBValue(std::string(value));
}

References FormatInfluxDBValue().

FormatInfluxDBValue() [3/7]

std::string Metric::FormatInfluxDBValue ( double  value )

staticprivate

{
    return std::to_string(value);
}

FormatInfluxDBValue() [4/7]

std::string Metric::FormatInfluxDBValue ( float  value )

staticprivate

{
    return FormatInfluxDBValue(double(value));
}

References FormatInfluxDBValue().

FormatInfluxDBValue() [5/7]

std::string Metric::FormatInfluxDBValue ( std::chrono::nanoseconds  value )

staticprivate

{
    return FormatInfluxDBValue(std::chrono::duration_cast<Milliseconds>(value).count());
}

References FormatInfluxDBValue().

FormatInfluxDBValue() [6/7]

std::string Metric::FormatInfluxDBValue ( std::string const &  value )

staticprivate

{
    return '"' + boost::replace_all_copy(value, "\"", "\\\"") + '"';
}

FormatInfluxDBValue() [7/7]

template<class T >

std::string Metric::FormatInfluxDBValue ( T  value )

staticprivate

{
    return std::to_string(value) + 'i';
}

GetDataStream()

std::iostream & Metric::GetDataStream ( )

inlineprivate

{ return *_dataStream; }

Referenced by Connect(), ScheduleSend(), and SendBatch().

Initialize()

void Metric::Initialize	(	std::string const &	realmName,
		Acore::Asio::IoContext &	ioContext,
		std::function< void()>	overallStatusLogger
	)

{
    _dataStream = std::make_unique<boost::asio::ip::tcp::iostream>();
    _realmName = FormatInfluxDBTagValue(realmName);
    _batchTimer = std::make_unique<Acore::Asio::DeadlineTimer>(ioContext);
    _overallStatusTimer = std::make_unique<Acore::Asio::DeadlineTimer>(ioContext);
    _overallStatusLogger = overallStatusLogger;
    LoadFromConfigs();
}

References _batchTimer, _dataStream, _overallStatusLogger, _overallStatusTimer, _realmName, FormatInfluxDBTagValue(), and LoadFromConfigs().

instance()

Metric * Metric::instance ( )

static

{
    static Metric instance;
    return &instance;
}

References instance().

Referenced by instance().

IsEnabled()

bool Metric::IsEnabled ( ) const

inline

{ return _enabled; }

LoadFromConfigs()

void Metric::LoadFromConfigs

(

)

{
    bool previousValue = _enabled;
    _enabled = sConfigMgr->GetOption<bool>("Metric.Enable", false);
    _updateInterval = sConfigMgr->GetOption<int32>("Metric.Interval", 1);
 
    if (_updateInterval < 1)
    {
        LOG_ERROR("metric", "'Metric.Interval' config set to {}, overriding to 1.", _updateInterval);
        _updateInterval = 1;
    }
 
    _overallStatusTimerInterval = sConfigMgr->GetOption<int32>("Metric.OverallStatusInterval", 1);
    if (_overallStatusTimerInterval < 1)
    {
        LOG_ERROR("metric", "'Metric.OverallStatusInterval' config set to {}, overriding to 1.", _overallStatusTimerInterval);
        _overallStatusTimerInterval = 1;
    }
 
    _thresholds.clear();
    std::vector<std::string> thresholdSettings = sConfigMgr->GetKeysByString("Metric.Threshold.");
    for (std::string const& thresholdSetting : thresholdSettings)
    {
        int64 thresholdValue = sConfigMgr->GetOption<int64>(thresholdSetting, 0);
        std::string thresholdName = thresholdSetting.substr(strlen("Metric.Threshold."));
        _thresholds[thresholdName] = thresholdValue;
    }
 
    // Schedule a send at this point only if the config changed from Disabled to Enabled.
    // Cancel any scheduled operation if the config changed from Enabled to Disabled.
    if (_enabled && !previousValue)
    {
        std::string connectionInfo = sConfigMgr->GetOption<std::string>("Metric.ConnectionInfo", "");
        if (connectionInfo.empty())
        {
            LOG_ERROR("metric", "'Metric.ConnectionInfo' not specified in configuration file.");
            return;
        }
 
        std::vector<std::string_view> tokens = Acore::Tokenize(connectionInfo, ';', true);
        if (tokens.size() != 3)
        {
            LOG_ERROR("metric", "'Metric.ConnectionInfo' specified with wrong format in configuration file.");
            return;
        }
 
        _hostname.assign(tokens[0]);
        _port.assign(tokens[1]);
        _databaseName.assign(tokens[2]);
        Connect();
 
        ScheduleSend();
        ScheduleOverallStatusLog();
    }
}

References _databaseName, _enabled, _hostname, _overallStatusTimerInterval, _port, _thresholds, _updateInterval, Connect(), LOG_ERROR, ScheduleOverallStatusLog(), ScheduleSend(), sConfigMgr, and Acore::Tokenize().

Referenced by Initialize().

LogEvent()

void Metric::LogEvent	(	std::string const &	category,
		std::string const &	title,
		std::string const &	description
	)

{
    using namespace std::chrono;
 
    MetricData* data = new MetricData;
    data->Category = category;
    data->Timestamp = system_clock::now();
    data->Type = METRIC_DATA_EVENT;
    data->Title = title;
    data->Text = description;
 
    _queuedData.Enqueue(data);
}

References _queuedData, MetricData::Category, METRIC_DATA_EVENT, MetricData::Text, MetricData::Timestamp, MetricData::Title, and MetricData::Type.

LogValue()

template<class T >

void Metric::LogValue ( std::string const &  category, T  value, std::vector< MetricTag >  tags  )

inline

    {
        using namespace std::chrono;
 
        MetricData* data = new MetricData;
        data->Category = category;
        data->Timestamp = system_clock::now();
        data->Type = METRIC_DATA_VALUE;
        data->Value = FormatInfluxDBValue(value);
        data->Tags = std::move(tags);
 
        _queuedData.Enqueue(data);
    }

References MetricData::Category, METRIC_DATA_VALUE, MetricData::Tags, MetricData::Timestamp, MetricData::Type, and MetricData::Value.

ScheduleOverallStatusLog()

void Metric::ScheduleOverallStatusLog ( )

private

{
    if (_enabled)
    {
        _overallStatusTimer->expires_from_now(boost::posix_time::seconds(_overallStatusTimerInterval));
        _overallStatusTimer->async_wait([this](const boost::system::error_code&)
        {
            _overallStatusTimerTriggered = true;
            ScheduleOverallStatusLog();
        });
    }
}

References _enabled, _overallStatusTimer, _overallStatusTimerInterval, _overallStatusTimerTriggered, and ScheduleOverallStatusLog().

Referenced by LoadFromConfigs(), and ScheduleOverallStatusLog().

ScheduleSend()

void Metric::ScheduleSend ( )

private

{
    if (_enabled)
    {
        _batchTimer->expires_from_now(boost::posix_time::seconds(_updateInterval));
        _batchTimer->async_wait(std::bind(&Metric::SendBatch, this));
    }
    else
    {
        static_cast<boost::asio::ip::tcp::iostream&>(GetDataStream()).close();
        MetricData* data;
 
        // Clear the queue
        while (_queuedData.Dequeue(data))
        {
            delete data;
        }
    }
}

References _batchTimer, _enabled, _queuedData, _updateInterval, GetDataStream(), and SendBatch().

Referenced by LoadFromConfigs(), and SendBatch().

SendBatch()

void Metric::SendBatch ( )

private

{
    using namespace std::chrono;
 
    std::stringstream batchedData;
    MetricData* data;
    bool firstLoop = true;
 
    while (_queuedData.Dequeue(data))
    {
        if (!firstLoop)
            batchedData << "\n";
 
        batchedData << data->Category;
        if (!_realmName.empty())
            batchedData << ",realm=" << _realmName;
 
        for (MetricTag const& tag : data->Tags)
            batchedData << "," << tag.first << "=" << FormatInfluxDBTagValue(tag.second);
 
        batchedData << " ";
 
        switch (data->Type)
        {
            case METRIC_DATA_VALUE:
                batchedData << "value=" << data->Value;
                break;
            case METRIC_DATA_EVENT:
                batchedData << "title=\"" << data->Title << "\",text=\"" << data->Text << "\"";
                break;
        }
 
        batchedData << " " << std::to_string(duration_cast<nanoseconds>(data->Timestamp.time_since_epoch()).count());
 
        firstLoop = false;
        delete data;
    }
 
    // Check if there's any data to send
    if (batchedData.tellp() == std::streampos(0))
    {
        ScheduleSend();
        return;
    }
 
    if (!GetDataStream().good() && !Connect())
        return;
 
    GetDataStream() << "POST " << "/write?db=" << _databaseName << " HTTP/1.1\r\n";
    GetDataStream() << "Host: " << _hostname << ":" << _port << "\r\n";
    GetDataStream() << "Accept: */*\r\n";
    GetDataStream() << "Content-Type: application/octet-stream\r\n";
    GetDataStream() << "Content-Transfer-Encoding: binary\r\n";
 
    GetDataStream() << "Content-Length: " << std::to_string(batchedData.tellp()) << "\r\n\r\n";
    GetDataStream() << batchedData.rdbuf();
 
    std::string http_version;
    GetDataStream() >> http_version;
    unsigned int status_code = 0;
    GetDataStream() >> status_code;
 
    if (status_code != 204)
    {
        LOG_ERROR("metric", "Error sending data, returned HTTP code: {}", status_code);
    }
 
    // Read and ignore the status description
    std::string status_description;
    std::getline(GetDataStream(), status_description);
 
    // Read headers
    std::string header;
 
    while (std::getline(GetDataStream(), header) && header != "\r")
    {
        if (header == "Connection: close\r")
        {
            static_cast<boost::asio::ip::tcp::iostream&>(GetDataStream()).close();
        }
    }
 
    ScheduleSend();
}

References _databaseName, _hostname, _port, _queuedData, _realmName, MetricData::Category, Connect(), FormatInfluxDBTagValue(), GetDataStream(), LOG_ERROR, METRIC_DATA_EVENT, METRIC_DATA_VALUE, ScheduleSend(), MetricData::Tags, MetricData::Text, MetricData::Timestamp, MetricData::Title, MetricData::Type, and MetricData::Value.

Referenced by ScheduleSend(), and Unload().

ShouldLog()

bool Metric::ShouldLog	(	std::string const &	category,
		int64	value
	)		const

{
    auto threshold = _thresholds.find(category);
 
    if (threshold == _thresholds.end())
    {
        return false;
    }
 
    return value >= threshold->second;
}

References _thresholds.

Unload()

void Metric::Unload

(

)

{
    // Send what's queued only if IoContext is stopped (so only on shutdown)
    if (_enabled && Acore::Asio::get_io_context(*_batchTimer).stopped())
    {
        _enabled = false;
        SendBatch();
    }
 
    _batchTimer->cancel();
    _overallStatusTimer->cancel();
}

References _batchTimer, _enabled, _overallStatusTimer, Acore::Asio::get_io_context(), and SendBatch().

Update()

void Metric::Update

(

)

{
    if (_overallStatusTimerTriggered)
    {
        _overallStatusTimerTriggered = false;
        _overallStatusLogger();
    }
}

References _overallStatusLogger, and _overallStatusTimerTriggered.

Member Data Documentation

_batchTimer

std::unique_ptr<Acore::Asio::DeadlineTimer> Metric::_batchTimer

private

Referenced by Initialize(), ScheduleSend(), and Unload().

_databaseName

std::string Metric::_databaseName

private

Referenced by LoadFromConfigs(), and SendBatch().

_dataStream

std::unique_ptr<std::iostream> Metric::_dataStream

private

Referenced by Initialize().

_enabled

bool Metric::_enabled = false

private

Referenced by Connect(), LoadFromConfigs(), ScheduleOverallStatusLog(), ScheduleSend(), and Unload().

_hostname

std::string Metric::_hostname

private

Referenced by Connect(), LoadFromConfigs(), and SendBatch().

_overallStatusLogger

std::function<void()> Metric::_overallStatusLogger

private

Referenced by Initialize(), and Update().

_overallStatusTimer

std::unique_ptr<Acore::Asio::DeadlineTimer> Metric::_overallStatusTimer

private

Referenced by Initialize(), ScheduleOverallStatusLog(), and Unload().

_overallStatusTimerInterval

int32 Metric::_overallStatusTimerInterval = 0

private

Referenced by LoadFromConfigs(), and ScheduleOverallStatusLog().

_overallStatusTimerTriggered

bool Metric::_overallStatusTimerTriggered = false

private

Referenced by ScheduleOverallStatusLog(), and Update().

_port

std::string Metric::_port

private

Referenced by Connect(), LoadFromConfigs(), and SendBatch().

_queuedData

MPSCQueue<MetricData> Metric::_queuedData

private

Referenced by LogEvent(), ScheduleSend(), and SendBatch().

_realmName

std::string Metric::_realmName

private

Referenced by Initialize(), and SendBatch().

_thresholds

std::unordered_map<std::string, int64> Metric::_thresholds

private

Referenced by LoadFromConfigs(), and ShouldLog().

_updateInterval

int32 Metric::_updateInterval = 0

private

Referenced by LoadFromConfigs(), and ScheduleSend().