@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
h 265 lines 8.57 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #ifndef NETDATA_PLUGINSD_PARSER_H
4 #define NETDATA_PLUGINSD_PARSER_H
5
6 #include "database/rrd.h"
7
8 #ifdef NETDATA_LOG_STREAM_RECEIVER
9 #include "streaming/stream-receiver-internals.h"
10 #endif
11
12 #define WORKER_PARSER_FIRST_JOB 35
13
14 // this has to be in-sync with the same at stream-thread.c
15 #define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION 24
16
17 // this controls the max response size of a function
18 #define PLUGINSD_MAX_DEFERRED_SIZE (100 * 1024 * 1024)
19
20 #define PLUGINSD_MIN_RRDSET_POINTERS_CACHE 1024
21 // Slots are cache indexes. Larger values are treated as uncached input to avoid sparse cache allocations.
22 #define PLUGINSD_CHART_SLOT_MAX 1000000
23 #define PLUGINSD_DIMENSION_SLOT_MAX 65535
24
25 // PARSER return codes
26 typedef enum __attribute__ ((__packed__)) parser_rc {
27 PARSER_RC_OK, // Callback was successful, go on
28 PARSER_RC_STOP, // Callback says STOP
29 PARSER_RC_ERROR // Callback failed (abort rest of callbacks)
30 } PARSER_RC;
31
32 typedef enum __attribute__ ((__packed__)) parser_input_type {
33 PARSER_INPUT_SPLIT = (1 << 1),
34 PARSER_DEFER_UNTIL_KEYWORD = (1 << 2),
35 } PARSER_INPUT_TYPE;
36
37 typedef enum __attribute__ ((__packed__)) {
38 PARSER_INIT_PLUGINSD = (1 << 1),
39 PARSER_INIT_STREAMING = (1 << 2),
40 PARSER_REP_REPLICATION = (1 << 3),
41 PARSER_REP_METADATA = (1 << 4),
42 PARSER_REP_DATA = (1 << 5),
43 } PARSER_REPERTOIRE;
44
45 struct parser;
46 typedef PARSER_RC (*keyword_function)(char **words, size_t num_words, struct parser *parser);
47
48 typedef struct parser_keyword {
49 char *keyword;
50 size_t id;
51 PARSER_REPERTOIRE repertoire;
52 size_t worker_job_id;
53 } PARSER_KEYWORD;
54
55 typedef struct parser_user_object {
56 bool cleanup_slots;
57 RRDSET *st;
58 RRDHOST *host;
59 void *opaque;
60 struct plugind *cd;
61 int trust_durations;
62 RRDLABELS *new_host_labels;
63 size_t clabel_count;
64 size_t data_collections_count;
65 int enabled;
66
67 #ifdef NETDATA_LOG_STREAM_RECEIVER
68 void *rpt;
69 #endif
70
71 STREAM_CAPABILITIES capabilities; // receiver capabilities
72
73 struct {
74 bool parsing_host;
75 uint32_t node_stale_after_seconds;
76 nd_uuid_t machine_guid;
77 char machine_guid_str[UUID_STR_LEN];
78 STRING *hostname;
79 RRDLABELS *rrdlabels;
80 } host_define;
81
82 struct parser_user_object_replay {
83 time_t start_time;
84 time_t end_time;
85
86 usec_t start_time_ut;
87 usec_t end_time_ut;
88
89 time_t wall_clock_time;
90
91 bool rset_enabled;
92 } replay;
93
94 struct parser_user_object_v2 {
95 bool locked_data_collection;
96 RRDSET_STREAM_BUFFER stream_buffer; // sender capabilities in this
97 time_t update_every;
98 time_t end_time;
99 time_t wall_clock_time;
100 bool ml_locked;
101 } v2;
102
103 struct {
104 Pvoid_t JudyL;
105 } vnodes;
106
107 } PARSER_USER_OBJECT;
108
109 typedef void (*parser_deferred_action_t)(struct parser *parser, void *action_data);
110 struct parser;
111 typedef ssize_t (*send_to_plugin_callback_t)(const char *txt, void *data, STREAM_TRAFFIC_TYPE type);
112
113 struct parser {
114 uint8_t version; // Parser version
115 PARSER_REPERTOIRE repertoire;
116 uint32_t flags;
117 int fd_input;
118 int fd_output;
119 ND_SOCK *sock;
120 send_to_plugin_callback_t send_to_plugin_cb;
121 void *send_to_plugin_data;
122
123 PARSER_USER_OBJECT user; // User defined structure to hold extra state between calls
124
125 struct buffered_reader reader;
126 struct line_splitter line;
127 const PARSER_KEYWORD *keyword;
128
129 struct {
130 const char *end_keyword;
131 BUFFER *response;
132 parser_deferred_action_t action;
133 void *action_data;
134 } defer;
135
136 struct {
137 DICTIONARY *functions;
138 usec_t smaller_monotonic_timeout_ut;
139 } inflight;
140
141 struct {
142 SPINLOCK spinlock;
143 } writer;
144 };
145
146 typedef struct parser PARSER;
147
148 PARSER *parser_init(struct parser_user_object *user, int fd_input, int fd_output, PARSER_INPUT_TYPE flags, ND_SOCK *sock);
149 void parser_init_repertoire(PARSER *parser, PARSER_REPERTOIRE repertoire);
150 void parser_destroy(PARSER *working_parser);
151 void pluginsd_cleanup_v2(PARSER *parser);
152 void pluginsd_keywords_init(PARSER *parser, PARSER_REPERTOIRE repertoire);
153 PARSER_RC parser_execute(PARSER *parser, const PARSER_KEYWORD *keyword, char **words, size_t num_words);
154
155 static inline int find_first_keyword(const char *src, char *dst, int dst_size, bool *isspace_map) {
156 const char *s = src, *keyword_start;
157
158 while (unlikely(isspace_map[(uint8_t)*s])) s++;
159 keyword_start = s;
160
161 while (likely(*s && !isspace_map[(uint8_t)*s]) && dst_size > 1) {
162 *dst++ = *s++;
163 dst_size--;
164 }
165 *dst = '\0';
166 return dst_size == 0 ? 0 : (int) (s - keyword_start);
167 }
168
169 const PARSER_KEYWORD *gperf_lookup_keyword(register const char *str, register size_t len);
170
171 static inline const PARSER_KEYWORD *parser_find_keyword(PARSER *parser, const char *command) {
172 const PARSER_KEYWORD *t = gperf_lookup_keyword(command, strlen(command));
173 if(t && (t->repertoire & parser->repertoire))
174 return t;
175
176 return NULL;
177 }
178
179 bool parser_reconstruct_node(BUFFER *wb, void *ptr);
180 bool parser_reconstruct_instance(BUFFER *wb, void *ptr);
181 bool parser_reconstruct_context(BUFFER *wb, void *ptr);
182
183 static inline int parser_action(PARSER *parser, char *input) {
184 #ifdef NETDATA_LOG_STREAM_RECEIVER
185 char line[1024];
186 strncpyz(line, input, sizeof(line) - 1);
187 #endif
188
189 parser->line.count++;
190
191 if(unlikely(parser->flags & PARSER_DEFER_UNTIL_KEYWORD)) {
192 char command[100 + 1];
193 bool has_keyword = find_first_keyword(input, command, 100, isspace_map_pluginsd);
194
195 if(!has_keyword || strcmp(command, parser->defer.end_keyword) != 0) {
196 if(parser->defer.response) {
197 buffer_strcat(parser->defer.response, input);
198 if(buffer_strlen(parser->defer.response) > PLUGINSD_MAX_DEFERRED_SIZE) {
199 // more than PLUGINSD_MAX_DEFERRED_SIZE of data,
200 // or a bad plugin that did not send the end_keyword
201 nd_log(NDLS_DAEMON, NDLP_ERR,
202 "PLUGINSD: deferred response is too big (%zu bytes, limit %zu bytes) "
203 "while waiting for keyword '%s' from plugin '%s' (transaction '%s'). "
204 "Stopping this plugin.",
205 buffer_strlen(parser->defer.response),
206 (size_t)PLUGINSD_MAX_DEFERRED_SIZE,
207 parser->defer.end_keyword ? parser->defer.end_keyword : "unknown",
208 parser->user.cd ? string2str(parser->user.cd->filename) : "unknown",
209 parser->defer.action_data ? string2str((STRING *)parser->defer.action_data) : "none");
210 return 1;
211 }
212 }
213 return 0;
214 }
215 else {
216 // call the action
217 parser->defer.action(parser, parser->defer.action_data);
218
219 // empty everything
220 parser->defer.action = NULL;
221 parser->defer.action_data = NULL;
222 parser->defer.end_keyword = NULL;
223 parser->defer.response = NULL;
224 parser->flags &= ~PARSER_DEFER_UNTIL_KEYWORD;
225 }
226 return 0;
227 }
228
229 parser->line.num_words = quoted_strings_splitter_pluginsd(input, parser->line.words, PLUGINSD_MAX_WORDS);
230 const char *command = get_word(parser->line.words, parser->line.num_words, 0);
231
232 if(unlikely(!command)) {
233 line_splitter_reset(&parser->line);
234 return 0;
235 }
236
237 PARSER_RC rc;
238 parser->keyword = parser_find_keyword(parser, command);
239 if(likely(parser->keyword)) {
240 worker_is_busy(parser->keyword->worker_job_id);
241
242 rc = parser_execute(parser, parser->keyword, parser->line.words, parser->line.num_words);
243 // rc = (*t->func)(words, num_words, parser);
244 worker_is_idle();
245 }
246 else
247 rc = PARSER_RC_ERROR;
248
249 if(rc == PARSER_RC_ERROR) {
250 CLEAN_BUFFER *wb = buffer_create(1024, NULL);
251 line_splitter_reconstruct_line(wb, &parser->line);
252 netdata_log_error("PLUGINSD: parser_action('%s') failed on line %zu: { %s } (quotes added to show parsing)",
253 command, parser->line.count, buffer_tostring(wb));
254 }
255
256 #ifdef NETDATA_LOG_STREAM_RECEIVER
257 if((parser->keyword->repertoire & PARSER_REP_REPLICATION) && !(parser->keyword->repertoire & PARSER_REP_DATA))
258 stream_receiver_log_payload(parser->user.rpt, line, STREAM_TRAFFIC_TYPE_REPLICATION, true);
259 #endif
260
261 line_splitter_reset(&parser->line);
262 return (rc == PARSER_RC_ERROR || rc == PARSER_RC_STOP);
263 }
264
265 #endif //NETDATA_PLUGINSD_PARSER_H