This repository has been archived by the owner on Aug 11, 2020. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 52
/
InfluxDB.cxx
128 lines (113 loc) · 3.15 KB
/
InfluxDB.cxx
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
///
/// \author Adam Wegrzynek <adam.wegrzynek@cern.ch>
///
#include "InfluxDB.h"
#include "InfluxDBException.h"
#include <iostream>
#include <memory>
#include <string>
#ifdef INFLUXDB_WITH_BOOST
#include <boost/lexical_cast.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>
#endif
namespace influxdb
{
InfluxDB::InfluxDB(std::unique_ptr<Transport> transport) :
mTransport(std::move(transport))
{
mBuffer = {};
mBuffering = false;
mBufferSize = 0;
mGlobalTags = {};
}
void InfluxDB::batchOf(const std::size_t size)
{
mBufferSize = size;
mBuffering = true;
}
void InfluxDB::flushBuffer() {
if (!mBuffering || mBuffer.empty()) {
return;
}
std::string stringBuffer{};
for (const auto &i : mBuffer) {
stringBuffer+= i + "\n";
}
mBuffer.clear();
transmit(std::move(stringBuffer));
}
void InfluxDB::addGlobalTag(std::string_view key, std::string_view value)
{
if (!mGlobalTags.empty()) mGlobalTags += ",";
mGlobalTags += key;
mGlobalTags += "=";
mGlobalTags += value;
}
InfluxDB::~InfluxDB()
{
if (mBuffering) {
flushBuffer();
}
}
void InfluxDB::transmit(std::string&& point)
{
mTransport->send(std::move(point));
}
void InfluxDB::write(Point&& metric)
{
if (mBuffering) {
mBuffer.emplace_back(metric.toLineProtocol());
if (mBuffer.size() >= mBufferSize) {
flushBuffer();
}
} else {
transmit(metric.toLineProtocol());
}
}
#ifdef INFLUXDB_WITH_BOOST
std::vector<Point> InfluxDB::query(const std::string& query)
{
auto response = mTransport->query(query);
std::stringstream ss;
ss << response;
std::vector<Point> points;
boost::property_tree::ptree pt;
boost::property_tree::read_json(ss, pt);
for (auto& result : pt.get_child("results")) {
auto isResultEmpty = result.second.find("series");
if (isResultEmpty == result.second.not_found()) return {};
for (auto& series : result.second.get_child("series")) {
auto columns = series.second.get_child("columns");
for (auto& values : series.second.get_child("values")) {
Point point{series.second.get<std::string>("name")};
auto iColumns = columns.begin();
auto iValues = values.second.begin();
for (; iColumns != columns.end() && iValues != values.second.end(); iColumns++, iValues++) {
auto value = iValues->second.get_value<std::string>();
auto column = iColumns->second.get_value<std::string>();
if (!column.compare("time")) {
std::tm tm = {};
std::stringstream ss;
ss << value;
ss >> std::get_time(&tm, "%FT%TZ");
point.setTimestamp(std::chrono::system_clock::from_time_t(std::mktime(&tm)));
continue;
}
// cast all values to double, if strings add to tags
try { point.addField(column, boost::lexical_cast<double>(value)); }
catch(...) { point.addTag(column, value); }
}
points.push_back(std::move(point));
}
}
}
return points;
}
#else
std::vector<Point> InfluxDB::query(const std::string& /*query*/)
{
throw InfluxDBException("InfluxDB::query", "Boost is required");
}
#endif
} // namespace influxdb