GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/telemetry_aggregator_influx.cc
Date: 2025-11-09 02:35:23
Exec Total Coverage
Lines: 97 134 72.4%
Branches: 104 212 49.1%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #include "telemetry_aggregator_influx.h"
6
7 #include <netinet/in.h>
8 #include <pthread.h>
9 #include <stdint.h>
10 #include <sys/socket.h>
11 #include <sys/types.h>
12
13 #include <cstring>
14 #include <map>
15 #include <string>
16 #include <vector>
17
18 #include "util/logging.h"
19 #include "util/posix.h"
20 #include "util/string.h"
21
22 namespace perf {
23
24 154 TelemetryAggregatorInflux::TelemetryAggregatorInflux(
25 Statistics *statistics,
26 int send_rate_sec,
27 OptionsManager *options_mgr,
28 MountPoint *mount_point,
29 154 const std::string &fqrn)
30 : TelemetryAggregator(statistics, send_rate_sec, mount_point, fqrn)
31
1/2
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
 154 , influx_extra_fields_("")
32
1/2
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
 154 , influx_extra_tags_("")
33 154 , send_delta_(true)
34 154 , socket_fd_(-1)
35 154 , res_(NULL) {
36 154 int params = 0;
37
38
4/6
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 154 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 132 times.
✓ Branch 10 taken 22 times.
 154 if (options_mgr->GetValue("CVMFS_INFLUX_HOST", &influx_host_)) {
39
1/2
✓ Branch 1 taken 132 times.
✗ Branch 2 not taken.
 132 if (influx_host_.size() > 1) {
40 132 params++;
41 } else {
42 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
43 "No value given for CVMFS_INFLUX_HOST");
44 }
45 }
46
47 154 std::string opt;
48
3/6
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 154 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 154 times.
✗ Branch 10 not taken.
 154 if (options_mgr->GetValue("CVMFS_INFLUX_PORT", &opt)) {
49
2/4
✓ Branch 3 taken 154 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 154 times.
✗ Branch 7 not taken.
 154 influx_port_ = static_cast<int>(String2Int64(opt.c_str()));
50
2/4
✓ Branch 0 taken 154 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
 154 if (influx_port_ > 0 && influx_port_ < 65536) {
51 154 params++;
52 } else {
53 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
54 "Invalid value for CVMFS_INFLUX_PORT [%s]", opt.c_str());
55 }
56 }
57
58
3/6
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 154 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 154 times.
✗ Branch 10 not taken.
 154 if (options_mgr->GetValue("CVMFS_INFLUX_METRIC_NAME", &influx_metric_name_)) {
59 154 params++;
60 }
61
62
4/6
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 154 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 110 times.
✓ Branch 10 taken 44 times.
 154 if (!options_mgr->GetValue("CVMFS_INFLUX_EXTRA_TAGS", &influx_extra_tags_)) {
63
1/2
✓ Branch 1 taken 110 times.
✗ Branch 2 not taken.
 110 influx_extra_tags_ = "";
64 }
65
66
4/6
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 154 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 110 times.
✓ Branch 10 taken 44 times.
 154 if (!options_mgr->GetValue("CVMFS_INFLUX_EXTRA_FIELDS",
67 &influx_extra_fields_)) {
68
1/2
✓ Branch 1 taken 110 times.
✗ Branch 2 not taken.
 110 influx_extra_fields_ = "";
69 }
70
71 // Read CVMFS_INFLUX_SEND_DELTA option, default to true (ON)
72 154 std::string send_delta_str;
73
4/6
✓ Branch 2 taken 154 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 154 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 44 times.
✓ Branch 10 taken 110 times.
 154 if (options_mgr->GetValue("CVMFS_INFLUX_SEND_DELTA", &send_delta_str)) {
74
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 send_delta_ = !options_mgr->IsOff(send_delta_str);
75 }
76
77
2/2
✓ Branch 0 taken 132 times.
✓ Branch 1 taken 22 times.
 154 if (params == 3) {
78 132 is_zombie_ = false;
79
1/2
✓ Branch 5 taken 132 times.
✗ Branch 6 not taken.
 132 LogCvmfs(kLogTelemetry, kLogDebug,
80 "Enabling influx metrics. "
81 "Send to [%s:%d] metric [%s]. Extra tags"
82 "[%s]. Extra fields [%s].",
83 influx_host_.c_str(), influx_port_, influx_metric_name_.c_str(),
84 influx_extra_tags_.c_str(), influx_extra_fields_.c_str());
85
1/2
✓ Branch 1 taken 132 times.
✗ Branch 2 not taken.
 132 const TelemetryReturn ret = OpenSocket();
86
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 132 times.
 132 if (ret != kTelemetrySuccess) {
87 ✗ is_zombie_ = true;
88 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
89 "Not enabling influx metrics. Error while open socket. %d", ret);
90 }
91 } else {
92 22 is_zombie_ = true;
93
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogWarn,
94 "Not enabling influx metrics. Not all mandatory variables set: "
95 "CVMFS_INFLUX_METRIC_NAME, CVMFS_INFLUX_HOST, CVMFS_INFLUX_PORT");
96 }
97 154 }
98
99 308 TelemetryAggregatorInflux::~TelemetryAggregatorInflux() {
100
2/2
✓ Branch 0 taken 132 times.
✓ Branch 1 taken 22 times.
 308 if (socket_fd_ >= 0) {
101 264 close(socket_fd_);
102 }
103
2/2
✓ Branch 0 taken 132 times.
✓ Branch 1 taken 22 times.
 308 if (res_) {
104 264 freeaddrinfo(res_);
105 }
106 }
107
108 /**
109 * Creates a string in the influx data format containing the absolute values
110 * of the counters. Counters are only included if their absolute value is > 0.
111 *
112 * Influx dataformat
113 * ( https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/
114 )
115 * Syntax
116 <measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]]
117 <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]
118 *
119 * Example
120 myMeasurement,tag1=value1,tag2=value2 fieldKey="fieldValue"
121 1556813561098000000
122 */
123 176 std::string TelemetryAggregatorInflux::MakePayload() {
124 // measurement and tags
125
1/2
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
 176 std::string metric_name = influx_metric_name_;
126
2/2
✓ Branch 0 taken 154 times.
✓ Branch 1 taken 22 times.
 176 if (send_delta_) {
127
1/2
✓ Branch 1 taken 154 times.
✗ Branch 2 not taken.
 154 metric_name += "_absolute";
128 }
129
2/4
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 176 times.
✗ Branch 5 not taken.
 176 std::string ret = metric_name + ",repo=" + fqrn_;
130
131
3/4
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 88 times.
✓ Branch 4 taken 88 times.
 176 if (influx_extra_tags_ != "") {
132
2/4
✓ Branch 1 taken 88 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 88 times.
✗ Branch 5 not taken.
 88 ret += "," + influx_extra_tags_;
133 }
134
135 // fields
136
1/2
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
 176 ret += " ";
137 176 bool add_token = false;
138 352 for (std::map<std::string, int64_t>::iterator it = counters_.begin(),
139 176 iEnd = counters_.end();
140
2/2
✓ Branch 1 taken 528 times.
✓ Branch 2 taken 176 times.
 704 it != iEnd;
141 528 it++) {
142
2/2
✓ Branch 1 taken 242 times.
✓ Branch 2 taken 286 times.
 528 if (it->second != 0) {
143
2/2
✓ Branch 0 taken 132 times.
✓ Branch 1 taken 110 times.
 242 if (add_token) {
144
1/2
✓ Branch 1 taken 132 times.
✗ Branch 2 not taken.
 132 ret += ",";
145 }
146
4/8
✓ Branch 2 taken 242 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 242 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 242 times.
✗ Branch 10 not taken.
✓ Branch 12 taken 242 times.
✗ Branch 13 not taken.
 242 ret += it->first + "=" + StringifyInt(it->second);
147 242 add_token = true;
148 }
149 }
150
3/4
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 88 times.
✓ Branch 4 taken 88 times.
 176 if (influx_extra_fields_ != "") {
151
2/2
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 44 times.
 88 if (add_token) {
152
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 ret += ",";
153 }
154
1/2
✓ Branch 1 taken 88 times.
✗ Branch 2 not taken.
 88 ret += influx_extra_fields_;
155 88 add_token = true;
156 }
157
158 // timestamp
159
2/2
✓ Branch 0 taken 154 times.
✓ Branch 1 taken 22 times.
 176 if (add_token) {
160
1/2
✓ Branch 1 taken 154 times.
✗ Branch 2 not taken.
 154 ret += " ";
161 }
162
2/4
✓ Branch 1 taken 176 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 176 times.
✗ Branch 5 not taken.
 176 ret += StringifyUint(timestamp_);
163
164 352 return ret;
165 176 }
166
167 /**
168 * Creates a string in the influx data format containing the delta between
169 * 2 measurements of the same counter. Counters are only included if their
170 * absolute value is > 0 (delta can be 0).
171 *
172 * NOTE: As influx_extra_fields_ are static, they are excluded of this
173 * delta format
174 *
175 * Influx dataformat
176 * ( https://docs.influxdata.com/influxdb/cloud/reference/syntax/line-protocol/
177 )
178 * Syntax
179 <measurement>[,<tag_key>=<tag_value>[,<tag_key>=<tag_value>]]
180 <field_key>=<field_value>[,<field_key>=<field_value>] [<timestamp>]
181 *
182 * Example
183 myMeasurement,tag1=value1,tag2=value2 fieldKey="fieldValue"
184 1556813561098000000
185 */
186 44 std::string TelemetryAggregatorInflux::MakeDeltaPayload() {
187 // measurement and tags
188
2/4
✓ Branch 2 taken 44 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 44 times.
✗ Branch 6 not taken.
 88 std::string ret = "" + influx_metric_name_ + "_delta,repo=" + fqrn_;
189
190
2/4
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 44 times.
✗ Branch 4 not taken.
 44 if (influx_extra_tags_ != "") {
191
2/4
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 44 times.
✗ Branch 5 not taken.
 44 ret += "," + influx_extra_tags_;
192 }
193
194 // fields
195
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 ret += " ";
196 44 bool add_token = false;
197 88 for (std::map<std::string, int64_t>::iterator it = counters_.begin(),
198 44 iEnd = counters_.end();
199
2/2
✓ Branch 1 taken 132 times.
✓ Branch 2 taken 44 times.
 176 it != iEnd;
200 132 it++) {
201 132 const int64_t value = it->second;
202
2/2
✓ Branch 0 taken 110 times.
✓ Branch 1 taken 22 times.
 132 if (value != 0) {
203 int64_t old_value;
204 try {
205
1/2
✓ Branch 2 taken 110 times.
✗ Branch 3 not taken.
 110 old_value = old_counters_.at(it->first);
206 ✗ } catch (const std::out_of_range &ex) {
207 ✗ old_value = 0;
208 }
209
2/2
✓ Branch 0 taken 66 times.
✓ Branch 1 taken 44 times.
 110 if (add_token) {
210
1/2
✓ Branch 1 taken 66 times.
✗ Branch 2 not taken.
 66 ret += ",";
211 }
212
4/8
✓ Branch 1 taken 110 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 110 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 110 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 110 times.
✗ Branch 12 not taken.
 110 ret += it->first + "=" + StringifyInt(value - old_value);
213 110 add_token = true;
214 }
215 }
216
217 // timestamp
218
1/2
✓ Branch 0 taken 44 times.
✗ Branch 1 not taken.
 44 if (add_token) {
219
1/2
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
 44 ret += " ";
220 }
221
2/4
✓ Branch 1 taken 44 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 44 times.
✗ Branch 5 not taken.
 44 ret += StringifyUint(timestamp_);
222
223 44 return ret;
224 }
225
226 132 TelemetryReturn TelemetryAggregatorInflux::OpenSocket() {
227 132 const char *hostname = influx_host_.c_str();
228 struct addrinfo hints;
229 int err;
230
231 132 memset(&hints, 0, sizeof(hints));
232 132 hints.ai_family = AF_INET;
233 132 hints.ai_socktype = SOCK_DGRAM;
234
235
1/2
✓ Branch 2 taken 132 times.
✗ Branch 3 not taken.
 132 err = getaddrinfo(influx_host_.c_str(), NULL, &hints, &res_);
236
2/4
✓ Branch 0 taken 132 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 132 times.
 132 if (err != 0 || res_ == NULL) {
237 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr,
238 ✗ "Failed to resolve influx server [%s]. errno=%d", hostname, errno);
239 ✗ return kTelemetryFailHostAddress;
240 }
241
242 132 socket_fd_ = socket(AF_INET, SOCK_DGRAM, 0);
243
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 132 times.
 132 if (socket_fd_ < 0) {
244 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr, "Failed to open socket");
245 ✗ freeaddrinfo(res_);
246 ✗ return kTelemetryFailSocket;
247 }
248
249 132 return kTelemetrySuccess;
250 }
251
252 ✗ TelemetryReturn TelemetryAggregatorInflux::SendToInflux(
253 const std::string &payload) {
254 ✗ struct sockaddr_in *dest_addr = NULL;
255 ✗ dest_addr = reinterpret_cast<sockaddr_in *>(res_->ai_addr);
256 ✗ dest_addr->sin_port = htons(influx_port_);
257
258 ✗ const ssize_t num_bytes_sent = sendto(
259 ✗ socket_fd_, payload.data(), payload.size(), 0,
260 reinterpret_cast<struct sockaddr *>(dest_addr),
261 sizeof(struct sockaddr_in));
262
263 ✗ if (num_bytes_sent < 0) {
264 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr,
265 ✗ "Failed to send to influx. errno=%d", errno);
266 ✗ return kTelemetryFailSend;
267 ✗ } else if (static_cast<size_t>(num_bytes_sent) != payload.size()) {
268 ✗ LogCvmfs(kLogTelemetry, kLogDebug | kLogSyslogErr,
269 "Incomplete send. Bytes transferred: %zd. Bytes expected %lu",
270 num_bytes_sent, payload.size());
271 ✗ return kTelemetryFailSend;
272 }
273
274 ✗ LogCvmfs(kLogTelemetry, kLogDebug, "INFLUX: POSTING [%s]", payload.c_str());
275 ✗ return kTelemetrySuccess;
276 }
277
278
279 ✗ void TelemetryAggregatorInflux::PushMetrics() {
280 // create payload
281 ✗ std::string payload = MakePayload();
282 ✗ std::string delta_payload = "";
283 ✗ if (send_delta_ && old_counters_.size() > 0) {
284 ✗ delta_payload = MakeDeltaPayload();
285 ✗ payload = payload + "\n" + delta_payload;
286 }
287 ✗ payload += "\n";
288
289 // send to influx
290 ✗ SendToInflux(payload);
291
292 // current counter is now old counter (only if delta sending is enabled)
293 ✗ if (send_delta_) {
294 ✗ counters_.swap(old_counters_);
295 }
296 }
297
298 } // namespace perf
299