@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 92 lines 3.9 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "commands.h"
4 #include "../stream-sender-internals.h"
5 #include "plugins.d/pluginsd_internals.h"
6
7 void stream_send_rrddim_metrics_v2(RRDSET_STREAM_BUFFER *rsb, RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags) {
8 if(!rsb->wb || !rsb->v2 || !netdata_double_isnumber(n) || !does_storage_number_exist(flags))
9 return;
10
11 bool with_slots = stream_has_capability(rsb, STREAM_CAP_SLOTS) ? true : false;
12 NUMBER_ENCODING integer_encoding = stream_has_capability(rsb, STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_HEX;
13 NUMBER_ENCODING doubles_encoding = stream_has_capability(rsb, STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL;
14 BUFFER *wb = rsb->wb;
15 time_t point_end_time_s = (time_t)(point_end_time_ut / USEC_PER_SEC);
16 if(unlikely(rsb->last_point_end_time_s != point_end_time_s)) {
17
18 if(unlikely(rsb->begin_v2_added))
19 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
20
21 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2, sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1);
22
23 if(with_slots) {
24 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
25 buffer_print_uint64_encoded(wb, integer_encoding, rd->rrdset->stream.snd.chart_slot);
26 }
27
28 buffer_fast_strcat(wb, " '", 2);
29 buffer_fast_strcat(wb, rrdset_id(rd->rrdset), string_strlen(rd->rrdset->id));
30 buffer_fast_strcat(wb, "' ", 2);
31 buffer_print_uint64_encoded(wb, integer_encoding, rd->rrdset->update_every);
32 buffer_fast_strcat(wb, " ", 1);
33 buffer_print_uint64_encoded(wb, integer_encoding, point_end_time_s);
34 buffer_fast_strcat(wb, " ", 1);
35 if(point_end_time_s == rsb->wall_clock_time)
36 buffer_fast_strcat(wb, "#", 1);
37 else
38 buffer_print_uint64_encoded(wb, integer_encoding, rsb->wall_clock_time);
39 buffer_fast_strcat(wb, "\n", 1);
40
41 rsb->last_point_end_time_s = point_end_time_s;
42 rsb->begin_v2_added = true;
43 }
44
45 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2, sizeof(PLUGINSD_KEYWORD_SET_V2) - 1);
46
47 if(with_slots) {
48 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
49 buffer_print_uint64_encoded(wb, integer_encoding, rd->stream.snd.dim_slot);
50 }
51
52 buffer_fast_strcat(wb, " '", 2);
53 buffer_fast_strcat(wb, rrddim_id(rd), string_strlen(rd->id));
54 buffer_fast_strcat(wb, "' ", 2);
55 bool send_double_baseline = rrddim_is_float(rd) && stream_has_capability(rsb, STREAM_CAP_FLOAT_BASELINE);
56 NETDATA_DOUBLE baseline_d = rrddim_last_collected_as_double(rd);
57 int64_t baseline_i = rrddim_last_collected_raw_int(rd);
58
59 if(send_double_baseline)
60 buffer_print_netdata_double_encoded(wb, doubles_encoding, baseline_d);
61 else
62 buffer_print_int64_encoded(wb, integer_encoding, baseline_i);
63
64 buffer_fast_strcat(wb, " ", 1);
65
66 NETDATA_DOUBLE baseline_cmp = send_double_baseline ? baseline_d : (NETDATA_DOUBLE)baseline_i;
67
68 if(baseline_cmp == n)
69 buffer_fast_strcat(wb, "#", 1);
70 else
71 buffer_print_netdata_double_encoded(wb, doubles_encoding, n);
72
73 buffer_fast_strcat(wb, " ", 1);
74 buffer_print_sn_flags(wb, flags, true);
75 buffer_fast_strcat(wb, "\n", 1);
76 }
77
78 ALWAYS_INLINE void stream_send_rrdset_metrics_finished(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) {
79 if(!rsb->wb)
80 return;
81
82 if(rsb->v2 && rsb->begin_v2_added) {
83 if(unlikely(rsb->rrdset_flags & RRDSET_FLAG_UPSTREAM_SEND_VARIABLES))
84 rrdvar_print_to_streaming_custom_chart_variables(st, rsb->wb);
85
86 buffer_fast_strcat(rsb->wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
87 }
88
89 sender_commit(st->rrdhost->sender, rsb->wb, STREAM_TRAFFIC_TYPE_DATA);
90
91 *rsb = (RRDSET_STREAM_BUFFER){ .wb = NULL, };
92 }