Skip to content

Commit 33366fc

Browse files
Al2KlimovAl2Klimov
committed
Provide zones and endpoints (ex local) stats via /v1/status
1 parent b16d96d commit 33366fc

File tree

4 files changed

+236
-0
lines changed

4 files changed

+236
-0
lines changed

‎lib/remote/endpoint.cpp

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,12 @@
2222
#include "remote/apilistener.hpp"
2323
#include "remote/jsonrpcconnection.hpp"
2424
#include "remote/zone.hpp"
25+
#include "base/perfdatavalue.hpp"
2526
#include "base/configtype.hpp"
2627
#include "base/utility.hpp"
2728
#include "base/exception.hpp"
2829
#include "base/convert.hpp"
30+
#include "base/statsfunction.hpp"
2931

3032
using namespace icinga;
3133

@@ -34,6 +36,8 @@ REGISTER_TYPE(Endpoint);
3436
boost::signals2::signal<void(const Endpoint::Ptr&, const JsonRpcConnection::Ptr&)> Endpoint::OnConnected;
3537
boost::signals2::signal<void(const Endpoint::Ptr&, const JsonRpcConnection::Ptr&)> Endpoint::OnDisconnected;
3638

39+
REGISTER_STATSFUNCTION(Endpoint, &Endpoint::StatsFunc);
40+
3741
void Endpoint::OnAllConfigLoaded()
3842
{
3943
ObjectImpl<Endpoint>::OnAllConfigLoaded();
@@ -118,6 +122,70 @@ Endpoint::Ptr Endpoint::GetLocalEndpoint()
118122
return listener->GetLocalEndpoint();
119123
}
120124

125+
void Endpoint::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
126+
{
127+
auto localZone (Zone::GetLocalZone());
128+
auto parentZone (localZone->GetParent());
129+
auto unorderedZones (ConfigType::GetObjectsByType<Zone>());
130+
std::set<Zone::Ptr> zones (unorderedZones.begin(), unorderedZones.end());
131+
std::set<Endpoint::Ptr> endpoints;
132+
Dictionary::Ptr ourStatus = new Dictionary;
133+
134+
unorderedZones.clear();
135+
136+
for (auto zone (zones.begin()); zone != zones.end();) {
137+
if ((*zone)->GetParent() == localZone) {
138+
++zone;
139+
} else {
140+
zones.erase(zone++);
141+
}
142+
}
143+
144+
zones.emplace(localZone);
145+
146+
if (parentZone)
147+
zones.emplace(parentZone);
148+
149+
for (auto& zone : zones) {
150+
auto zoneEndpoints (zone->GetEndpoints());
151+
endpoints.insert(zoneEndpoints.begin(), zoneEndpoints.end());
152+
}
153+
154+
endpoints.erase(GetLocalEndpoint());
155+
156+
for (auto& endpoint : endpoints) {
157+
ourStatus->Set(endpoint->GetName(), new Dictionary({
158+
{"local_log_position", endpoint->GetLocalLogPosition()},
159+
{"remote_log_position", endpoint->GetRemoteLogPosition()},
160+
{"connecting", endpoint->GetConnecting()},
161+
{"syncing", endpoint->GetSyncing()},
162+
{"connected", endpoint->GetConnected()},
163+
{"last_message_sent", endpoint->GetLastMessageSent()},
164+
{"last_message_received", endpoint->GetLastMessageReceived()},
165+
{"messages_sent_per_second", endpoint->GetMessagesSentPerSecond()},
166+
{"messages_received_per_second", endpoint->GetMessagesReceivedPerSecond()},
167+
{"bytes_sent_per_second", endpoint->GetBytesSentPerSecond()},
168+
{"bytes_received_per_second", endpoint->GetBytesReceivedPerSecond()}
169+
}));
170+
}
171+
172+
{
173+
ObjectLock ourStatusLock (ourStatus);
174+
175+
for (auto& nameEndpointStatus : ourStatus) {
176+
Dictionary::Ptr endpointStatus = nameEndpointStatus.second;
177+
ObjectLock endpointStatusLock (endpointStatus);
178+
auto labelPrefix ("endpoint_" + nameEndpointStatus.first + "_");
179+
180+
for (auto& labelValue : endpointStatus) {
181+
perfdata->Add(new PerfdataValue(labelPrefix + labelValue.first, labelValue.second));
182+
}
183+
}
184+
}
185+
186+
status->Set("endpoint", ourStatus);
187+
}
188+
121189
void Endpoint::AddMessageSent(int bytes)
122190
{
123191
double time = Utility::GetTime();

‎lib/remote/endpoint.hpp

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222

2323
#include "remote/i2-remote.hpp"
2424
#include "remote/endpoint-ti.hpp"
25+
#include "base/array.hpp"
26+
#include "base/dictionary.hpp"
2527
#include "base/ringbuffer.hpp"
2628
#include <set>
2729

@@ -54,6 +56,7 @@ class Endpoint final : public ObjectImpl<Endpoint>
5456
bool GetConnected() const override;
5557

5658
static Endpoint::Ptr GetLocalEndpoint();
59+
static void StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata);
5760

5861
void SetCachedZone(const intrusive_ptr<Zone>& zone);
5962

‎lib/remote/zone.cpp

Lines changed: 162 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,18 @@
2020
#include "remote/zone.hpp"
2121
#include "remote/zone-ti.cpp"
2222
#include "remote/jsonrpcconnection.hpp"
23+
#include "base/perfdatavalue.hpp"
2324
#include "base/objectlock.hpp"
2425
#include "base/logger.hpp"
26+
#include "base/statsfunction.hpp"
27+
#include <limits>
2528

2629
using namespace icinga;
2730

2831
REGISTER_TYPE(Zone);
2932

33+
REGISTER_STATSFUNCTION(Zone, &Zone::StatsFunc);
34+
3035
void Zone::OnAllConfigLoaded()
3136
{
3237
ObjectImpl<Zone>::OnAllConfigLoaded();
@@ -147,6 +152,163 @@ Zone::Ptr Zone::GetLocalZone()
147152
return local->GetZone();
148153
}
149154

155+
static std::set<String> l_StatsFuncAggregateSum ({
156+
"messages_sent_per_second", "messages_received_per_second", "bytes_sent_per_second", "bytes_received_per_second"
157+
});
158+
159+
static std::set<String> l_StatsFuncAggregateCount ({
160+
"connecting", "syncing", "connected"
161+
});
162+
163+
static std::set<String> l_StatsFuncAggregateMin ({
164+
"last_message_sent", "last_message_received"
165+
});
166+
167+
void Zone::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
168+
{
169+
auto localZone (Zone::GetLocalZone());
170+
auto parentZone (localZone->GetParent());
171+
auto unorderedZones (ConfigType::GetObjectsByType<Zone>());
172+
std::set<Zone::Ptr> zones (unorderedZones.begin(), unorderedZones.end());
173+
Dictionary::Ptr ourStatus = new Dictionary;
174+
auto localEndpoint (Endpoint::GetLocalEndpoint());
175+
176+
unorderedZones.clear();
177+
178+
for (auto zone (zones.begin()); zone != zones.end();) {
179+
if ((*zone)->GetParent() == localZone) {
180+
++zone;
181+
} else {
182+
zones.erase(zone++);
183+
}
184+
}
185+
186+
zones.emplace(localZone);
187+
188+
if (parentZone)
189+
zones.emplace(parentZone);
190+
191+
for (auto& zone : zones) {
192+
Dictionary::Ptr endpointStats = new Dictionary({
193+
{"local_log_position", (Array::Ptr)new Array},
194+
{"remote_log_position", (Array::Ptr)new Array},
195+
{"connecting", (Array::Ptr)new Array},
196+
{"syncing", (Array::Ptr)new Array},
197+
{"connected", (Array::Ptr)new Array},
198+
{"last_message_sent", (Array::Ptr)new Array},
199+
{"last_message_received", (Array::Ptr)new Array},
200+
{"messages_sent_per_second", (Array::Ptr)new Array},
201+
{"messages_received_per_second", (Array::Ptr)new Array},
202+
{"bytes_sent_per_second", (Array::Ptr)new Array},
203+
{"bytes_received_per_second", (Array::Ptr)new Array}
204+
});
205+
206+
auto endpoints (zone->GetEndpoints());
207+
208+
endpoints.erase(localEndpoint);
209+
210+
if (endpoints.empty())
211+
continue;
212+
213+
for (auto& endpoint : endpoints) {
214+
((Array::Ptr)endpointStats->Get("local_log_position"))->Add(endpoint->GetLocalLogPosition());
215+
((Array::Ptr)endpointStats->Get("remote_log_position"))->Add(endpoint->GetRemoteLogPosition());
216+
((Array::Ptr)endpointStats->Get("connecting"))->Add(endpoint->GetConnecting());
217+
((Array::Ptr)endpointStats->Get("syncing"))->Add(endpoint->GetSyncing());
218+
((Array::Ptr)endpointStats->Get("connected"))->Add(endpoint->GetConnected());
219+
((Array::Ptr)endpointStats->Get("last_message_sent"))->Add(endpoint->GetLastMessageSent());
220+
((Array::Ptr)endpointStats->Get("last_message_received"))->Add(endpoint->GetLastMessageReceived());
221+
((Array::Ptr)endpointStats->Get("messages_sent_per_second"))->Add(endpoint->GetMessagesSentPerSecond());
222+
((Array::Ptr)endpointStats->Get("messages_received_per_second"))->Add(endpoint->GetMessagesReceivedPerSecond());
223+
((Array::Ptr)endpointStats->Get("bytes_sent_per_second"))->Add(endpoint->GetBytesSentPerSecond());
224+
((Array::Ptr)endpointStats->Get("bytes_received_per_second"))->Add(endpoint->GetBytesReceivedPerSecond());
225+
}
226+
227+
for (auto& label : l_StatsFuncAggregateSum) {
228+
auto sum (0.0);
229+
Array::Ptr values = endpointStats->Get(label);
230+
ObjectLock valuesLock (values);
231+
232+
for (auto& value : values) {
233+
sum += value.Get<double>();
234+
}
235+
236+
endpointStats->Set(label, sum);
237+
}
238+
239+
for (auto& label : l_StatsFuncAggregateCount) {
240+
uintmax_t count = 0;
241+
Array::Ptr values = endpointStats->Get(label);
242+
ObjectLock valuesLock (values);
243+
244+
for (auto& value : values) {
245+
if (value.Get<bool>()) {
246+
++count;
247+
}
248+
}
249+
250+
endpointStats->Set(label, count);
251+
}
252+
253+
for (auto& label : l_StatsFuncAggregateMin) {
254+
auto min (std::numeric_limits<double>::infinity());
255+
Array::Ptr values = endpointStats->Get(label);
256+
ObjectLock valuesLock (values);
257+
258+
for (auto& value : values) {
259+
auto number (value.Get<double>());
260+
261+
if (number < min) {
262+
min = number;
263+
}
264+
}
265+
266+
endpointStats->Set(label, min);
267+
}
268+
269+
{
270+
auto maxDiff (-std::numeric_limits<double>::infinity());
271+
Array::Ptr remoteLogPositions = endpointStats->Get("remote_log_position");
272+
ObjectLock remoteLogPositionLock (remoteLogPositions);
273+
auto remoteLogPosition (begin(remoteLogPositions));
274+
Array::Ptr localLogPositions = endpointStats->Get("local_log_position");
275+
ObjectLock localLogPositionLock (localLogPositions);
276+
277+
for (auto& localLogPosition : localLogPositions) {
278+
auto diff (localLogPosition - *remoteLogPosition);
279+
280+
if (diff > maxDiff) {
281+
maxDiff = diff;
282+
}
283+
284+
++remoteLogPosition;
285+
}
286+
287+
endpointStats->Set("client_log_lag", maxDiff);
288+
endpointStats->Remove("local_log_position");
289+
endpointStats->Remove("remote_log_position");
290+
}
291+
292+
ourStatus->Set(zone->GetName(), endpointStats);
293+
}
294+
295+
{
296+
ObjectLock ourStatusLock (ourStatus);
297+
298+
for (auto& nameZoneStatus : ourStatus) {
299+
Dictionary::Ptr zoneStatus = nameZoneStatus.second;
300+
ObjectLock zoneStatusLock (zoneStatus);
301+
auto labelPrefix ("zone_" + nameZoneStatus.first + "_");
302+
303+
for (auto& labelValue : zoneStatus) {
304+
perfdata->Add(new PerfdataValue(labelPrefix + labelValue.first, labelValue.second));
305+
}
306+
}
307+
}
308+
309+
status->Set("zone", ourStatus);
310+
}
311+
150312
void Zone::ValidateEndpointsRaw(const Lazy<Array::Ptr>& lvalue, const ValidationUtils& utils)
151313
{
152314
ObjectImpl<Zone>::ValidateEndpointsRaw(lvalue, utils);

‎lib/remote/zone.hpp

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
#ifndef ZONE_H
2121
#define ZONE_H
2222

23+
#include "base/array.hpp"
24+
#include "base/dictionary.hpp"
2325
#include "remote/i2-remote.hpp"
2426
#include "remote/zone-ti.hpp"
2527
#include "remote/endpoint.hpp"
@@ -48,6 +50,7 @@ class Zone final : public ObjectImpl<Zone>
4850
bool IsSingleInstance() const;
4951

5052
static Zone::Ptr GetLocalZone();
53+
static void StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata);
5154

5255
protected:
5356
void ValidateEndpointsRaw(const Lazy<Array::Ptr>& lvalue, const ValidationUtils& utils) override;

0 commit comments

Comments
 (0)