Virtual hosts for data collection (#14464)
* support multiple hosts at pluginsd structures * cleanup obsolete code * use a lookup hashtable to quickly find the keyword to execute, without traversing the whole linked list of keywords * more cleanup * move new hash function to inlined.h * minimize comparisons, eliminate a pre-parsing of the first keyword for each line * cleanup parser from old code * move parser into libnetdata * unique entries in parser keywords hashtable * move all hashing functions to inlined.h, name their sources, simple_hash() now defaults to FNV1a, it was FNV1 * small_hash() for parser * plugins.d now can switch hosts, and also create/update them * update hash function and hashtable size * updated message * unittest all hashing functions * reset the chart when setting a new host * remove host tags * enable archived hosts when a collector pushes host info * do not need localhost to swtich to localhost * disable ARAL and OWA with -DFSANITIZE_ADDRESS=1
Costa Tsaousis committed
Feb 9, 2023 at 20:27 UTC
414f416c5d290db3c3eed9073258c834fac7f2f7
30 files changed
+868
-617
Makefile.am
+2
-7
@@ -107,7 +107,6 @@ SUBDIRS += \
107
streaming \
108
web \
109
claim \
110
- parser \
110
spawn \
111
$(NULL)
112
@@ -160,6 +159,8 @@ LIBNETDATA_FILES = \
159
libnetdata/log/log.h \
160
libnetdata/onewayalloc/onewayalloc.c \
161
libnetdata/onewayalloc/onewayalloc.h \
162
+ libnetdata/parser/parser.c \
163
+ libnetdata/parser/parser.h \
164
libnetdata/popen/popen.c \
165
libnetdata/popen/popen.h \
166
libnetdata/procfile/procfile.c \
@@ -677,11 +678,6 @@ CLAIM_FILES = \
678
claim/claim.h \
679
$(NULL)
680
680
-PARSER_FILES = \
681
- parser/parser.c \
682
- parser/parser.h \
683
- $(NULL)
684
-
681
if ENABLE_ACLK
682
ACLK_FILES = \
683
aclk/aclk_util.c \
@@ -949,7 +945,6 @@ NETDATA_FILES = \
945
$(STATSD_PLUGIN_FILES) \
946
$(WEB_PLUGIN_FILES) \
947
$(CLAIM_FILES) \
952
- $(PARSER_FILES) \
948
$(ACLK_ALWAYS_BUILD_FILES) \
949
$(ACLK_FILES) \
950
$(SPAWN_PLUGIN_FILES) \
collectors/apps.plugin/apps_plugin.c
+2
-2
@@ -4336,7 +4336,7 @@ static void apps_plugin_function_processes(const char *transaction, char *functi
4336
struct pid_stat *p;
4337
4338
char *words[PLUGINSD_MAX_WORDS] = { NULL };
4339
- size_t num_words = pluginsd_split_words(function, words, PLUGINSD_MAX_WORDS, NULL, NULL, 0);
4339
+ size_t num_words = pluginsd_split_words(function, words, PLUGINSD_MAX_WORDS);
4340
4341
struct target *category = NULL, *user = NULL, *group = NULL;
4342
const char *process_name = NULL;
@@ -4809,7 +4809,7 @@ void *reader_main(void *arg __maybe_unused) {
4809
while(!apps_plugin_exit && (s = fgets(buffer, PLUGINSD_LINE_MAX, stdin))) {
4810
4811
char *words[PLUGINSD_MAX_WORDS] = { NULL };
4812
- size_t num_words = pluginsd_split_words(buffer, words, PLUGINSD_MAX_WORDS, NULL, NULL, 0);
4812
+ size_t num_words = pluginsd_split_words(buffer, words, PLUGINSD_MAX_WORDS);
4813
4814
const char *keyword = get_word(words, num_words, 0);
4815
collectors/plugins.d/README.md
+77
@@ -175,6 +175,83 @@ The plugin should output instructions for Netdata to its output (`stdout`). Sinc
175
176
`DISABLE` will disable this plugin. This will prevent Netdata from restarting the plugin. You can also exit with the value `1` to have the same effect.
177
178
+#### HOST_DEFINE
179
+
180
+`HOST_DEFINE` defines a new (or updates an existing) virtual host.
181
+
182
+The template is:
183
+
184
+> HOST_DEFINE machine_guid hostname
185
+
186
+where:
187
+
188
+- `machine_guid`
189
+
190
+ uniquely identifies the host, this is what will be needed to add charts to the host.
191
+
192
+- `hostname`
193
+
194
+ is the hostname of the virtual host
195
+
196
+#### HOST_LABEL
197
+
198
+`HOST_LABEL` adds a key-value pair to the virtual host labels. It has to be given between `HOST_DEFINE` and `HOST_DEFINE_END`.
199
+
200
+The template is:
201
+
202
+> HOST_LABEL key value
203
+
204
+where:
205
+
206
+- `key`
207
+
208
+ uniquely identifies the key of the label
209
+
210
+- `value`
211
+
212
+ is the value associated with this key
213
+
214
+There are a few special keys that are used to define the system information of the monitored system:
215
+
216
+- `_cloud_provider_type`
217
+- `_cloud_instance_type`
218
+- `_cloud_instance_region`
219
+- `_os_name`
220
+- `_os_version`
221
+- `_kernel_version`
222
+- `_system_cores`
223
+- `_system_cpu_freq`
224
+- `_system_ram_total`
225
+- `_system_disk_space`
226
+- `_architecture`
227
+- `_virtualization`
228
+- `_container`
229
+- `_container_detection`
230
+- `_virt_detection`
231
+- `_is_k8s_node`
232
+- `_install_type`
233
+- `_prebuilt_arch`
234
+- `_prebuilt_dist`
235
+
236
+#### HOST_DEFINE_END
237
+
238
+`HOST_DEFINE_END` commits the host information, creating a new host entity, or updating an existing one with the same `machine_guid`.
239
+
240
+#### HOST
241
+
242
+`HOST` switches data collection between hosts.
243
+
244
+The template is:
245
+
246
+> HOST machine_guid
247
+
248
+where:
249
+
250
+- `machine_guid`
251
+
252
+ is the UUID of the host to switch to. After this command, every other command following it is assumed to be associated with this host.
253
+ Setting machine_guid to `localhost` switches data collection to the local host.
254
+
255
#### CHART
256
257
`CHART` defines a new chart.
collectors/plugins.d/plugins_d.c
+49
-44
@@ -18,7 +18,7 @@ inline size_t pluginsd_initialize_plugin_directories()
18
}
19
20
// Parse it and store it to plugin directories
21
- return quoted_strings_splitter(plugins_dir_list, plugin_directories, PLUGINSD_MAX_DIRECTORIES, config_isspace, NULL, NULL, 0);
21
+ return quoted_strings_splitter(plugins_dir_list, plugin_directories, PLUGINSD_MAX_DIRECTORIES, config_isspace);
22
}
23
24
static inline void plugin_set_disabled(struct plugind *cd) {
@@ -51,6 +51,8 @@ static void pluginsd_worker_thread_cleanup(void *arg)
51
{
52
struct plugind *cd = (struct plugind *)arg;
53
54
+ worker_unregister();
55
+
56
netdata_spinlock_lock(&cd->unsafe.spinlock);
57
58
cd->unsafe.running = false;
@@ -62,74 +64,73 @@ static void pluginsd_worker_thread_cleanup(void *arg)
64
netdata_spinlock_unlock(&cd->unsafe.spinlock);
65
66
if (pid) {
65
- info("data collection thread exiting");
66
-
67
siginfo_t info;
68
- info("killing child process pid %d", pid);
68
+ info("PLUGINSD: 'host:%s', killing data collection child process with pid %d",
69
+ rrdhost_hostname(cd->host), pid);
70
+
71
if (killpid(pid) != -1) {
70
- info("waiting for child process pid %d to exit...", pid);
72
+ info("PLUGINSD: 'host:%s', waiting for data collection child process pid %d to exit...",
73
+ rrdhost_hostname(cd->host), pid);
74
+
75
waitid(P_PID, (id_t)pid, &info, WEXITED);
76
}
77
}
78
}
79
80
#define SERIAL_FAILURES_THRESHOLD 10
77
-static void pluginsd_worker_thread_handle_success(struct plugind *cd)
78
-{
81
+static void pluginsd_worker_thread_handle_success(struct plugind *cd) {
82
if (likely(cd->successful_collections)) {
83
sleep((unsigned int)cd->update_every);
84
return;
85
}
86
87
if (likely(cd->serial_failures <= SERIAL_FAILURES_THRESHOLD)) {
85
- info(
86
- "'%s' (pid %d) does not generate useful output but it reports success (exits with 0). %s.",
87
- cd->fullfilename, cd->unsafe.pid,
88
- plugin_is_enabled(cd) ? "Waiting a bit before starting it again." : "Will not start it again - it is now disabled.");
88
+ info("PLUGINSD: 'host:%s', '%s' (pid %d) does not generate useful output but it reports success (exits with 0). %s.",
89
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid,
90
+ plugin_is_enabled(cd) ? "Waiting a bit before starting it again." : "Will not start it again - it is now disabled.");
91
+
92
sleep((unsigned int)(cd->update_every * 10));
93
return;
94
}
95
96
if (cd->serial_failures > SERIAL_FAILURES_THRESHOLD) {
94
- error(
95
- "'%s' (pid %d) does not generate useful output, although it reports success (exits with 0)."
96
- "We have tried to collect something %zu times - unsuccessfully. Disabling it.",
97
- cd->fullfilename, cd->unsafe.pid, cd->serial_failures);
97
+ error("PLUGINSD: 'host:'%s', '%s' (pid %d) does not generate useful output, "
98
+ "although it reports success (exits with 0)."
99
+ "We have tried to collect something %zu times - unsuccessfully. Disabling it.",
100
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid, cd->serial_failures);
101
plugin_set_disabled(cd);
102
return;
103
}
104
}
105
103
-static void pluginsd_worker_thread_handle_error(struct plugind *cd, int worker_ret_code)
104
-{
106
+static void pluginsd_worker_thread_handle_error(struct plugind *cd, int worker_ret_code) {
107
if (worker_ret_code == -1) {
106
- info("'%s' (pid %d) was killed with SIGTERM. Disabling it.", cd->fullfilename, cd->unsafe.pid);
108
+ info("PLUGINSD: 'host:%s', '%s' (pid %d) was killed with SIGTERM. Disabling it.",
109
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid);
110
plugin_set_disabled(cd);
111
return;
112
}
113
114
if (!cd->successful_collections) {
112
- error(
113
- "'%s' (pid %d) exited with error code %d and haven't collected any data. Disabling it.", cd->fullfilename,
114
- cd->unsafe.pid, worker_ret_code);
115
+ error("PLUGINSD: 'host:%s', '%s' (pid %d) exited with error code %d and haven't collected any data. Disabling it.",
116
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid, worker_ret_code);
117
plugin_set_disabled(cd);
118
return;
119
}
120
121
if (cd->serial_failures <= SERIAL_FAILURES_THRESHOLD) {
120
- error(
121
- "'%s' (pid %d) exited with error code %d, but has given useful output in the past (%zu times). %s",
122
- cd->fullfilename, cd->unsafe.pid, worker_ret_code, cd->successful_collections,
123
- plugin_is_enabled(cd) ? "Waiting a bit before starting it again." : "Will not start it again - it is disabled.");
122
+ error("PLUGINSD: 'host:%s', '%s' (pid %d) exited with error code %d, but has given useful output in the past (%zu times). %s",
123
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid, worker_ret_code, cd->successful_collections,
124
+ plugin_is_enabled(cd) ? "Waiting a bit before starting it again." : "Will not start it again - it is disabled.");
125
sleep((unsigned int)(cd->update_every * 10));
126
return;
127
}
128
129
if (cd->serial_failures > SERIAL_FAILURES_THRESHOLD) {
129
- error(
130
- "'%s' (pid %d) exited with error code %d, but has given useful output in the past (%zu times)."
131
- "We tried to restart it %zu times, but it failed to generate data. Disabling it.",
132
- cd->fullfilename, cd->unsafe.pid, worker_ret_code, cd->successful_collections, cd->serial_failures);
130
+ error("PLUGINSD: 'host:%s', '%s' (pid %d) exited with error code %d, but has given useful output in the past (%zu times)."
131
+ "We tried to restart it %zu times, but it failed to generate data. Disabling it.",
132
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid, worker_ret_code,
133
+ cd->successful_collections, cd->serial_failures);
134
plugin_set_disabled(cd);
135
return;
136
}
@@ -137,8 +138,7 @@ static void pluginsd_worker_thread_handle_error(struct plugind *cd, int worker_r
138
139
#undef SERIAL_FAILURES_THRESHOLD
140
140
-static void *pluginsd_worker_thread(void *arg)
141
-{
141
+static void *pluginsd_worker_thread(void *arg) {
142
worker_register("PLUGINSD");
143
144
netdata_thread_cleanup_push(pluginsd_worker_thread_cleanup, arg);
@@ -151,14 +151,20 @@ static void *pluginsd_worker_thread(void *arg)
151
while (service_running(SERVICE_COLLECTORS)) {
152
FILE *fp_child_input = NULL;
153
FILE *fp_child_output = netdata_popen(cd->cmd, &cd->unsafe.pid, &fp_child_input);
154
+
155
if (unlikely(!fp_child_input || !fp_child_output)) {
155
- error("Cannot popen(\"%s\", \"r\").", cd->cmd);
156
+ error("PLUGINSD: 'host:%s', cannot popen(\"%s\", \"r\").", rrdhost_hostname(cd->host), cd->cmd);
157
break;
158
}
159
159
- info("connected to '%s' running on pid %d", cd->fullfilename, cd->unsafe.pid);
160
- count = pluginsd_process(localhost, cd, fp_child_input, fp_child_output, 0);
161
- error("'%s' (pid %d) disconnected after %zu successful data collections (ENDs).", cd->fullfilename, cd->unsafe.pid, count);
160
+ info("PLUGINSD: 'host:%s' connected to '%s' running on pid %d",
161
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid);
162
+
163
+ count = pluginsd_process(cd->host, cd, fp_child_input, fp_child_output, 0);
164
+
165
+ info("PLUGINSD: 'host:%s', '%s' (pid %d) disconnected after %zu successful data collections (ENDs).",
166
+ rrdhost_hostname(cd->host), cd->fullfilename, cd->unsafe.pid, count);
167
+
168
killpid(cd->unsafe.pid);
169
170
int worker_ret_code = netdata_pclose(fp_child_input, fp_child_output, cd->unsafe.pid);
@@ -172,29 +178,29 @@ static void *pluginsd_worker_thread(void *arg)
178
if (unlikely(!plugin_is_enabled(cd)))
179
break;
180
}
175
- worker_unregister();
181
182
netdata_thread_cleanup_pop(1);
183
return NULL;
184
}
185
181
-static void pluginsd_main_cleanup(void *data)
182
-{
186
+static void pluginsd_main_cleanup(void *data) {
187
struct netdata_static_thread *static_thread = (struct netdata_static_thread *)data;
188
static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
185
- info("cleaning up...");
189
+ info("PLUGINSD: cleaning up...");
190
191
struct plugind *cd;
192
for (cd = pluginsd_root; cd; cd = cd->next) {
193
netdata_spinlock_lock(&cd->unsafe.spinlock);
194
if (cd->unsafe.enabled && cd->unsafe.running && cd->unsafe.thread != 0) {
191
- info("stopping plugin thread: %s", cd->id);
195
+ info("PLUGINSD: 'host:%s', stopping plugin thread: %s",
196
+ rrdhost_hostname(cd->host), cd->id);
197
+
198
netdata_thread_cancel(cd->unsafe.thread);
199
}
200
netdata_spinlock_unlock(&cd->unsafe.spinlock);
201
}
202
197
- info("cleanup completed.");
203
+ info("PLUGINSD: cleanup completed.");
204
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
205
206
worker_unregister();
@@ -282,6 +288,7 @@ void *pluginsd_main(void *ptr)
288
strncpyz(cd->filename, file->d_name, FILENAME_MAX);
289
snprintfz(cd->fullfilename, FILENAME_MAX, "%s/%s", directory_name, cd->filename);
290
291
+ cd->host = localhost;
292
cd->unsafe.enabled = enabled;
293
cd->unsafe.running = false;
294
@@ -294,9 +301,7 @@ void *pluginsd_main(void *ptr)
301
config_get(cd->id, "command options", def));
302
303
// link it
297
- if (likely(pluginsd_root))
298
- cd->next = pluginsd_root;
299
- pluginsd_root = cd;
304
+ DOUBLE_LINKED_LIST_PREPEND_ITEM_UNSAFE(pluginsd_root, cd, prev, next);
305
306
if (plugin_is_enabled(cd)) {
307
char tag[NETDATA_THREAD_TAG_MAX + 1];
collectors/plugins.d/plugins_d.h
+8
-1
@@ -38,6 +38,11 @@
38
#define PLUGINSD_KEYWORD_SET_V2 "SET2"
39
#define PLUGINSD_KEYWORD_END_V2 "END2"
40
41
+#define PLUGINSD_KEYWORD_HOST_DEFINE "HOST_DEFINE"
42
+#define PLUGINSD_KEYWORD_HOST_DEFINE_END "HOST_DEFINE_END"
43
+#define PLUGINSD_KEYWORD_HOST_LABEL "HOST_LABEL"
44
+#define PLUGINSD_KEYWORD_HOST "HOST"
45
+
46
#define PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT 10 // seconds
47
48
#define PLUGINSD_LINE_MAX_SSL_READ 512
@@ -60,6 +65,7 @@ struct plugind {
65
size_t serial_failures; // the number of times the plugin started
66
// without collecting values
67
68
+ RRDHOST *host; // the host the plugin collects data for
69
int update_every; // the plugin default data collection frequency
70
71
struct {
@@ -71,7 +77,8 @@ struct plugind {
77
} unsafe;
78
79
time_t started_t;
74
- uint32_t capabilities; // follows the same principles as streaming capabilities
80
+
81
+ struct plugind *prev;
82
struct plugind *next;
83
};
84
collectors/plugins.d/pluginsd_parser.c
+281
-7
@@ -152,7 +152,7 @@ static inline RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, cons
152
if(likely(st->pluginsd.pos < st->pluginsd.used)) {
153
rda = st->pluginsd.rda[st->pluginsd.pos];
154
RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
155
- if (likely(rd && strcmp(rrddim_id(rd), dimension) == 0)) {
155
+ if (likely(rd && string_strcmp(rd->id, dimension) == 0)) {
156
st->pluginsd.pos++;
157
return rd;
158
}
@@ -297,6 +297,146 @@ PARSER_RC pluginsd_end(char **words, size_t num_words, void *user)
297
return PARSER_RC_OK;
298
}
299
300
+static void pluginsd_host_define_cleanup(void *user) {
301
+ PARSER_USER_OBJECT *u = user;
302
+
303
+ string_freez(u->host_define.hostname);
304
+ dictionary_destroy(u->host_define.rrdlabels);
305
+
306
+ u->host_define.hostname = NULL;
307
+ u->host_define.rrdlabels = NULL;
308
+ u->host_define.parsing_host = false;
309
+}
310
+
311
+static inline bool pluginsd_validate_machine_guid(const char *guid, uuid_t *uuid, char *output) {
312
+ if(uuid_parse(guid, *uuid))
313
+ return false;
314
+
315
+ uuid_unparse_lower(*uuid, output);
316
+
317
+ return true;
318
+}
319
+
320
+static PARSER_RC pluginsd_host_define(char **words, size_t num_words, void *user) {
321
+ PARSER_USER_OBJECT *u = user;
322
+
323
+ char *guid = get_word(words, num_words, 1);
324
+ char *hostname = get_word(words, num_words, 2);
325
+
326
+ if(unlikely(!guid || !*guid || !hostname || !*hostname))
327
+ return PLUGINSD_DISABLE_PLUGIN(user, PLUGINSD_KEYWORD_HOST_DEFINE, "missing parameters");
328
+
329
+ if(unlikely(u->host_define.parsing_host))
330
+ return PLUGINSD_DISABLE_PLUGIN(user, PLUGINSD_KEYWORD_HOST_DEFINE,
331
+ "another host definition is already open - did you send " PLUGINSD_KEYWORD_HOST_DEFINE_END "?");
332
+
333
+ if(!pluginsd_validate_machine_guid(guid, &u->host_define.machine_guid, u->host_define.machine_guid_str))
334
+ return PLUGINSD_DISABLE_PLUGIN(user, PLUGINSD_KEYWORD_HOST_DEFINE, "cannot parse MACHINE_GUID - is it a valid UUID?");
335
+
336
+ u->host_define.hostname = string_strdupz(hostname);
337
+ u->host_define.rrdlabels = rrdlabels_create();
338
+ u->host_define.parsing_host = true;
339
+
340
+ return PARSER_RC_OK;
341
+}
342
+
343
+static inline PARSER_RC pluginsd_host_dictionary(char **words, size_t num_words, void *user, DICTIONARY *dict, const char *keyword) {
344
+ PARSER_USER_OBJECT *u = user;
345
+
346
+ char *name = get_word(words, num_words, 1);
347
+ char *value = get_word(words, num_words, 2);
348
+
349
+ if(!name || !*name || !value)
350
+ return PLUGINSD_DISABLE_PLUGIN(user, keyword, "missing parameters");
351
+
352
+ if(!u->host_define.parsing_host || !dict)
353
+ return PLUGINSD_DISABLE_PLUGIN(user, keyword, "host is not defined, send " PLUGINSD_KEYWORD_HOST_DEFINE " before this");
354
+
355
+ rrdlabels_add(dict, name, value, RRDLABEL_SRC_CONFIG);
356
+
357
+ return PARSER_RC_OK;
358
+}
359
+
360
+static PARSER_RC pluginsd_host_labels(char **words, size_t num_words, void *user) {
361
+ PARSER_USER_OBJECT *u = user;
362
+ return pluginsd_host_dictionary(words, num_words, user, u->host_define.rrdlabels, PLUGINSD_KEYWORD_HOST_LABEL);
363
+}
364
+
365
+static PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, size_t num_words __maybe_unused, void *user) {
366
+ PARSER_USER_OBJECT *u = user;
367
+
368
+ if(!u->host_define.parsing_host)
369
+ return PLUGINSD_DISABLE_PLUGIN(user, PLUGINSD_KEYWORD_HOST_DEFINE_END, "missing initialization, send " PLUGINSD_KEYWORD_HOST_DEFINE " before this");
370
+
371
+ RRDHOST *host = rrdhost_find_or_create(
372
+ string2str(u->host_define.hostname),
373
+ string2str(u->host_define.hostname),
374
+ u->host_define.machine_guid_str,
375
+ "Netdata Virtual Host 1.0",
376
+ netdata_configured_timezone,
377
+ netdata_configured_abbrev_timezone,
378
+ netdata_configured_utc_offset,
379
+ NULL,
380
+ program_name,
381
+ program_version,
382
+ default_rrd_update_every,
383
+ default_rrd_history_entries,
384
+ default_rrd_memory_mode,
385
+ default_health_enabled,
386
+ default_rrdpush_enabled,
387
+ default_rrdpush_destination,
388
+ default_rrdpush_api_key,
389
+ default_rrdpush_send_charts_matching,
390
+ default_rrdpush_enable_replication,
391
+ default_rrdpush_seconds_to_replicate,
392
+ default_rrdpush_replication_step,
393
+ rrdhost_labels_to_system_info(u->host_define.rrdlabels),
394
+ false
395
+ );
396
+
397
+ if(host->rrdlabels) {
398
+ rrdlabels_migrate_to_these(host->rrdlabels, u->host_define.rrdlabels);
399
+ }
400
+ else {
401
+ host->rrdlabels = u->host_define.rrdlabels;
402
+ u->host_define.rrdlabels = NULL;
403
+ }
404
+
405
+ pluginsd_host_define_cleanup(user);
406
+
407
+ u->host = host;
408
+ pluginsd_set_chart_from_parent(user, NULL, PLUGINSD_KEYWORD_HOST_DEFINE_END);
409
+
410
+ rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
411
+ rrdcontext_host_child_connected(host);
412
+
413
+ return PARSER_RC_OK;
414
+}
415
+
416
+static PARSER_RC pluginsd_host(char **words, size_t num_words, void *user) {
417
+ PARSER_USER_OBJECT *u = user;
418
+
419
+ char *guid = get_word(words, num_words, 1);
420
+
421
+ if(!guid || !*guid || strcmp(guid, "localhost") == 0) {
422
+ u->host = localhost;
423
+ return PARSER_RC_OK;
424
+ }
425
+
426
+ uuid_t uuid;
427
+ char uuid_str[UUID_STR_LEN];
428
+ if(!pluginsd_validate_machine_guid(guid, &uuid, uuid_str))
429
+ return PLUGINSD_DISABLE_PLUGIN(user, PLUGINSD_KEYWORD_HOST, "cannot parse MACHINE_GUID - is it a valid UUID?");
430
+
431
+ RRDHOST *host = rrdhost_find_by_guid(uuid_str);
432
+ if(unlikely(!host))
433
+ return PLUGINSD_DISABLE_PLUGIN(user, PLUGINSD_KEYWORD_HOST, "cannot find a host with this machine guid - have you created it?");
434
+
435
+ u->host = host;
436
+
437
+ return PARSER_RC_OK;
438
+}
439
+
440
PARSER_RC pluginsd_chart(char **words, size_t num_words, void *user)
441
{
442
RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_CHART);
@@ -883,7 +1023,7 @@ PARSER_RC pluginsd_disable(char **words __maybe_unused, size_t num_words __maybe
1023
{
1024
info("PLUGINSD: plugin called DISABLE. Disabling it.");
1025
((PARSER_USER_OBJECT *) user)->enabled = 0;
886
- return PARSER_RC_ERROR;
1026
+ return PARSER_RC_STOP;
1027
}
1028
1029
PARSER_RC pluginsd_label(char **words, size_t num_words, void *user)
@@ -1654,8 +1794,8 @@ PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_words __maybe_
1794
static void pluginsd_process_thread_cleanup(void *ptr) {
1795
PARSER *parser = (PARSER *)ptr;
1796
1657
- if(parser->user_cleanup_cb)
1658
- parser->user_cleanup_cb(parser->user);
1797
+ pluginsd_cleanup_v2(parser->user);
1798
+ pluginsd_host_define_cleanup(parser->user);
1799
1800
rrd_collector_finished();
1801
parser_destroy(parser);
@@ -1695,7 +1835,10 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugi
1835
};
1836
1837
// fp_plugin_output = our input; fp_plugin_input = our output
1698
- PARSER *parser = parser_init(host, &user, NULL, fp_plugin_output, fp_plugin_input, -1, PARSER_INPUT_SPLIT, NULL);
1838
+ PARSER *parser = parser_init(&user, fp_plugin_output, fp_plugin_input, -1,
1839
+ PARSER_INPUT_SPLIT, NULL);
1840
+
1841
+ pluginsd_keywords_init(parser, PARSER_INIT_PLUGINSD);
1842
1843
rrd_collector_started();
1844
@@ -1704,9 +1847,10 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugi
1847
netdata_thread_cleanup_push(pluginsd_process_thread_cleanup, parser);
1848
1849
user.parser = parser;
1850
+ char buffer[PLUGINSD_LINE_MAX + 1];
1851
1708
- while (likely(!parser_next(parser))) {
1709
- if (unlikely(!service_running(SERVICE_COLLECTORS) || parser_action(parser, NULL)))
1852
+ while (likely(!parser_next(parser, buffer, PLUGINSD_LINE_MAX))) {
1853
+ if (unlikely(!service_running(SERVICE_COLLECTORS) || parser_action(parser, buffer)))
1854
break;
1855
}
1856
@@ -1725,3 +1869,133 @@ inline size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugi
1869
1870
return count;
1871
}
1872
+
1873
+static void pluginsd_keywords_init_internal(PARSER *parser, PLUGINSD_KEYWORDS types, void (*add_func)(PARSER *parser, char *keyword, keyword_function func)) {
1874
+
1875
+ if (types & PARSER_INIT_PLUGINSD) {
1876
+ add_func(parser, PLUGINSD_KEYWORD_FLUSH, pluginsd_flush);
1877
+ add_func(parser, PLUGINSD_KEYWORD_DISABLE, pluginsd_disable);
1878
+
1879
+ add_func(parser, PLUGINSD_KEYWORD_HOST_DEFINE, pluginsd_host_define);
1880
+ add_func(parser, PLUGINSD_KEYWORD_HOST_DEFINE_END, pluginsd_host_define_end);
1881
+ add_func(parser, PLUGINSD_KEYWORD_HOST_LABEL, pluginsd_host_labels);
1882
+ add_func(parser, PLUGINSD_KEYWORD_HOST, pluginsd_host);
1883
+ }
1884
+
1885
+ if (types & (PARSER_INIT_PLUGINSD | PARSER_INIT_STREAMING)) {
1886
+ // plugins.d plugins and streaming
1887
+ add_func(parser, PLUGINSD_KEYWORD_CHART, pluginsd_chart);
1888
+ add_func(parser, PLUGINSD_KEYWORD_DIMENSION, pluginsd_dimension);
1889
+ add_func(parser, PLUGINSD_KEYWORD_VARIABLE, pluginsd_variable);
1890
+ add_func(parser, PLUGINSD_KEYWORD_LABEL, pluginsd_label);
1891
+ add_func(parser, PLUGINSD_KEYWORD_OVERWRITE, pluginsd_overwrite);
1892
+ add_func(parser, PLUGINSD_KEYWORD_CLABEL_COMMIT, pluginsd_clabel_commit);
1893
+ add_func(parser, PLUGINSD_KEYWORD_CLABEL, pluginsd_clabel);
1894
+ add_func(parser, PLUGINSD_KEYWORD_FUNCTION, pluginsd_function);
1895
+ add_func(parser, PLUGINSD_KEYWORD_FUNCTION_RESULT_BEGIN, pluginsd_function_result_begin);
1896
+
1897
+ add_func(parser, PLUGINSD_KEYWORD_BEGIN, pluginsd_begin);
1898
+ add_func(parser, PLUGINSD_KEYWORD_SET, pluginsd_set);
1899
+ add_func(parser, PLUGINSD_KEYWORD_END, pluginsd_end);
1900
+
1901
+ inflight_functions_init(parser);
1902
+ }
1903
+
1904
+ if (types & PARSER_INIT_STREAMING) {
1905
+ add_func(parser, PLUGINSD_KEYWORD_CHART_DEFINITION_END, pluginsd_chart_definition_end);
1906
+
1907
+ // replication
1908
+ add_func(parser, PLUGINSD_KEYWORD_REPLAY_BEGIN, pluginsd_replay_begin);
1909
+ add_func(parser, PLUGINSD_KEYWORD_REPLAY_SET, pluginsd_replay_set);
1910
+ add_func(parser, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, pluginsd_replay_rrddim_collection_state);
1911
+ add_func(parser, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE, pluginsd_replay_rrdset_collection_state);
1912
+ add_func(parser, PLUGINSD_KEYWORD_REPLAY_END, pluginsd_replay_end);
1913
+
1914
+ // streaming metrics v2
1915
+ add_func(parser, PLUGINSD_KEYWORD_BEGIN_V2, pluginsd_begin_v2);
1916
+ add_func(parser, PLUGINSD_KEYWORD_SET_V2, pluginsd_set_v2);
1917
+ add_func(parser, PLUGINSD_KEYWORD_END_V2, pluginsd_end_v2);
1918
+ }
1919
+}
1920
+
1921
+void pluginsd_keywords_init(PARSER *parser, PLUGINSD_KEYWORDS types) {
1922
+ pluginsd_keywords_init_internal(parser, types, parser_add_keyword);
1923
+}
1924
+
1925
+struct pluginsd_user_unittest {
1926
+ size_t size;
1927
+ const char **hashtable;
1928
+ uint32_t (*hash)(const char *s);
1929
+ size_t collisions;
1930
+};
1931
+
1932
+void pluginsd_keyword_collision_check(PARSER *parser, char *keyword, keyword_function func __maybe_unused) {
1933
+ struct pluginsd_user_unittest *u = parser->user;
1934
+
1935
+ uint32_t hash = u->hash(keyword);
1936
+ uint32_t slot = hash % u->size;
1937
+
1938
+ if(u->hashtable[slot])
1939
+ u->collisions++;
1940
+
1941
+ u->hashtable[slot] = keyword;
1942
+}
1943
+
1944
+static struct {
1945
+ const char *name;
1946
+ uint32_t (*hash)(const char *s);
1947
+ size_t slots_needed;
1948
+} hashers[] = {
1949
+ { .name = "djb2_hash32(s)", djb2_hash32, .slots_needed = 0, },
1950
+ { .name = "fnv1_hash32(s)", fnv1_hash32, .slots_needed = 0, },
1951
+ { .name = "fnv1a_hash32(s)", fnv1a_hash32, .slots_needed = 0, },
1952
+ { .name = "larson_hash32(s)", larson_hash32, .slots_needed = 0, },
1953
+ { .name = "pluginsd_parser_hash32(s)", pluginsd_parser_hash32, .slots_needed = 0, },
1954
+
1955
+ // terminator
1956
+ { .name = NULL, NULL, .slots_needed = 0, },
1957
+};
1958
+
1959
+int pluginsd_parser_unittest(void) {
1960
+ PARSER *p;
1961
+ size_t slots_to_check = 1000;
1962
+ size_t i, h;
1963
+
1964
+ // check for hashtable collisions
1965
+ for(h = 0; hashers[h].name ;h++) {
1966
+ hashers[h].slots_needed = slots_to_check * 1000000;
1967
+
1968
+ for (i = 10; i < slots_to_check; i++) {
1969
+ struct pluginsd_user_unittest user = {
1970
+ .hash = hashers[h].hash,
1971
+ .size = i,
1972
+ .hashtable = callocz(i, sizeof(const char *)),
1973
+ .collisions = 0,
1974
+ };
1975
+
1976
+ p = parser_init(&user, NULL, NULL, -1, PARSER_INPUT_SPLIT, NULL);
1977
+ pluginsd_keywords_init_internal(p, PARSER_INIT_PLUGINSD | PARSER_INIT_STREAMING,
1978
+ pluginsd_keyword_collision_check);
1979
+ parser_destroy(p);
1980
+
1981
+ freez(user.hashtable);
1982
+
1983
+ if (!user.collisions) {
1984
+ hashers[h].slots_needed = i;
1985
+ break;
1986
+ }
1987
+ }
1988
+ }
1989
+
1990
+ for(h = 0; hashers[h].name ;h++) {
1991
+ if(hashers[h].slots_needed > 1000)
1992
+ info("PARSER: hash function '%s' cannot be used without collisions under %zu slots", hashers[h].name, slots_to_check);
1993
+ else
1994
+ info("PARSER: hash function '%s' needs PARSER_KEYWORDS_HASHTABLE_SIZE (in parser.h) set to %zu", hashers[h].name, hashers[h].slots_needed);
1995
+ }
1996
+
1997
+ p = parser_init(NULL, NULL, NULL, -1, PARSER_INPUT_SPLIT, NULL);
1998
+ pluginsd_keywords_init(p, PARSER_INIT_PLUGINSD | PARSER_INIT_STREAMING);
1999
+ parser_destroy(p);
2000
+ return 0;
2001
+}
collectors/plugins.d/pluginsd_parser.h
+16
-1
@@ -3,7 +3,12 @@
3
#ifndef NETDATA_PLUGINSD_PARSER_H
4
#define NETDATA_PLUGINSD_PARSER_H
5
6
-#include "parser/parser.h"
6
+#include "daemon/common.h"
7
+
8
+typedef enum __attribute__ ((__packed__)) {
9
+ PARSER_INIT_PLUGINSD = (1 << 1),
10
+ PARSER_INIT_STREAMING = (1 << 2),
11
+} PLUGINSD_KEYWORDS;
12
13
typedef struct parser_user_object {
14
PARSER *parser;
@@ -17,6 +22,14 @@ typedef struct parser_user_object {
22
size_t data_collections_count;
23
int enabled;
24
25
+ struct {
26
+ bool parsing_host;
27
+ uuid_t machine_guid;
28
+ char machine_guid_str[UUID_STR_LEN];
29
+ STRING *hostname;
30
+ DICTIONARY *rrdlabels;
31
+ } host_define;
32
+
33
struct parser_user_object_replay {
34
time_t start_time;
35
time_t end_time;
@@ -42,4 +55,6 @@ typedef struct parser_user_object {
55
PARSER_RC pluginsd_function(char **words, size_t num_words, void *user);
56
PARSER_RC pluginsd_function_result_begin(char **words, size_t num_words, void *user);
57
void inflight_functions_init(PARSER *parser);
58
+void pluginsd_keywords_init(PARSER *parser, PLUGINSD_KEYWORDS types);
59
+
60
#endif //NETDATA_PLUGINSD_PARSER_H
collectors/statsd.plugin/statsd.c
+1
-1
@@ -1480,7 +1480,7 @@ static int statsd_readfile(const char *filename, STATSD_APP *app, STATSD_APP_CHA
1480
else if (!strcmp(name, "dimension")) {
1481
// metric [name [type [multiplier [divisor]]]]
1482
char *words[10] = { NULL };
1483
- size_t num_words = pluginsd_split_words(value, words, 10, NULL, NULL, 0);
1483
+ size_t num_words = pluginsd_split_words(value, words, 10);
1484
1485
int pattern = 0;
1486
size_t i = 0;
configure.ac
+1
-1
@@ -1743,6 +1743,7 @@ AC_CONFIG_FILES([
1743
libnetdata/locks/Makefile
1744
libnetdata/log/Makefile
1745
libnetdata/onewayalloc/Makefile
1746
+ libnetdata/parser/Makefile
1747
libnetdata/popen/Makefile
1748
libnetdata/procfile/Makefile
1749
libnetdata/simple_pattern/Makefile
@@ -1791,7 +1792,6 @@ AC_CONFIG_FILES([
1792
web/server/static/Makefile
1793
claim/Makefile
1794
spawn/Makefile
1794
- parser/Makefile
1795
])
1796
1797
AC_OUTPUT
daemon/main.c
+4
@@ -1317,6 +1317,7 @@ void post_conf_load(char **user)
1317
int pgc_unittest(void);
1318
int mrg_unittest(void);
1319
int julytest(void);
1320
+int pluginsd_parser_unittest(void);
1321
1322
int main(int argc, char **argv) {
1323
// initialize the system clocks
@@ -1437,6 +1438,9 @@ int main(int argc, char **argv) {
1438
if(strcmp(optarg, "unittest") == 0) {
1439
unittest_running = true;
1440
1441
+ if (pluginsd_parser_unittest())
1442
+ return 1;
1443
+
1444
if (unit_test_static_threads())
1445
return 1;
1446
if (unit_test_buffer())
database/rrd.h
+3
-1
@@ -279,7 +279,7 @@ void rrdlabels_destroy(DICTIONARY *labels_dict);
279
void rrdlabels_add(DICTIONARY *dict, const char *name, const char *value, RRDLABEL_SRC ls);
280
void rrdlabels_add_pair(DICTIONARY *dict, const char *string, RRDLABEL_SRC ls);
281
void rrdlabels_get_value_to_buffer_or_null(DICTIONARY *labels, BUFFER *wb, const char *key, const char *quote, const char *null);
282
-void rrdlabels_get_value_to_char_or_null(DICTIONARY *labels, char **value, const char *key);
282
+void rrdlabels_get_value_strdup_or_null(DICTIONARY *labels, char **value, const char *key);
283
void rrdlabels_flush(DICTIONARY *labels_dict);
284
285
void rrdlabels_unmark_all(DICTIONARY *labels);
@@ -961,6 +961,8 @@ struct rrdhost_system_info {
961
int mc_version;
962
};
963
964
+struct rrdhost_system_info *rrdhost_labels_to_system_info(DICTIONARY *labels);
965
+
966
struct rrdhost {
967
char machine_guid[GUID_LEN + 1]; // the unique ID of this host
968
database/rrdcalc.c
+1
-1
@@ -100,7 +100,7 @@ static STRING *rrdcalc_replace_variables_with_rrdset_labels(const char *line, RR
100
label_val[i - RRDCALC_VAR_LABEL_LEN - 1] = '\0';
101
102
if(likely(rc->rrdset && rc->rrdset->rrdlabels)) {
103
- rrdlabels_get_value_to_char_or_null(rc->rrdset->rrdlabels, &lbl_value, label_val);
103
+ rrdlabels_get_value_strdup_or_null(rc->rrdset->rrdlabels, &lbl_value, label_val);
104
if (lbl_value) {
105
char *buf = find_and_replace(temp, var, lbl_value, m);
106
freez(temp);
database/rrdhost.c
+29
-4
@@ -1279,6 +1279,33 @@ void rrdhost_save_charts(RRDHOST *host) {
1279
rrdset_foreach_done(st);
1280
}
1281
1282
+struct rrdhost_system_info *rrdhost_labels_to_system_info(DICTIONARY *labels) {
1283
+ struct rrdhost_system_info *info = callocz(1, sizeof(struct rrdhost_system_info));
1284
+ info->hops = 1;
1285
+
1286
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->cloud_provider_type, "_cloud_provider_type");
1287
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->cloud_instance_type, "_cloud_instance_type");
1288
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->cloud_instance_region, "_cloud_instance_region");
1289
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->host_os_name, "_os_name");
1290
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->host_os_version, "_os_version");
1291
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->kernel_version, "_kernel_version");
1292
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->host_cores, "_system_cores");
1293
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->host_cpu_freq, "_system_cpu_freq");
1294
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->host_ram_total, "_system_ram_total");
1295
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->host_disk_space, "_system_disk_space");
1296
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->architecture, "_architecture");
1297
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->virtualization, "_virtualization");
1298
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->container, "_container");
1299
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->container_detection, "_container_detection");
1300
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->virt_detection, "_virt_detection");
1301
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->is_k8s_node, "_is_k8s_node");
1302
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->install_type, "_install_type");
1303
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->prebuilt_arch, "_prebuilt_arch");
1304
+ rrdlabels_get_value_strdup_or_null(labels, &localhost->system_info->prebuilt_dist, "_prebuilt_dist");
1305
+
1306
+ return info;
1307
+}
1308
+
1309
static void rrdhost_load_auto_labels(void) {
1310
DICTIONARY *labels = localhost->rrdlabels;
1311
@@ -1289,8 +1316,7 @@ static void rrdhost_load_auto_labels(void) {
1316
rrdlabels_add(labels, "_cloud_instance_type", localhost->system_info->cloud_instance_type, RRDLABEL_SRC_AUTO);
1317
1318
if (localhost->system_info->cloud_instance_region)
1292
- rrdlabels_add(
1293
- labels, "_cloud_instance_region", localhost->system_info->cloud_instance_region, RRDLABEL_SRC_AUTO);
1319
+ rrdlabels_add(labels, "_cloud_instance_region", localhost->system_info->cloud_instance_region, RRDLABEL_SRC_AUTO);
1320
1321
if (localhost->system_info->host_os_name)
1322
rrdlabels_add(labels, "_os_name", localhost->system_info->host_os_name, RRDLABEL_SRC_AUTO);
@@ -1354,8 +1380,7 @@ void rrdhost_set_is_parent_label(int count) {
1380
DICTIONARY *labels = localhost->rrdlabels;
1381
1382
if (count == 0 || count == 1) {
1357
- rrdlabels_add(
1358
- labels, "_is_parent", (count) ? "true" : "false", RRDLABEL_SRC_AUTO);
1383
+ rrdlabels_add(labels, "_is_parent", (count) ? "true" : "false", RRDLABEL_SRC_AUTO);
1384
1385
//queue a node info
1386
#ifdef ENABLE_ACLK
database/rrdlabels.c
+1
-1
@@ -654,7 +654,7 @@ void rrdlabels_get_value_to_buffer_or_null(DICTIONARY *labels, BUFFER *wb, const
654
// ----------------------------------------------------------------------------
655
// rrdlabels_get_value_to_char_or_null()
656
657
-void rrdlabels_get_value_to_char_or_null(DICTIONARY *labels, char **value, const char *key) {
657
+void rrdlabels_get_value_strdup_or_null(DICTIONARY *labels, char **value, const char *key) {
658
const DICTIONARY_ITEM *acquired_item = dictionary_get_and_acquire_item(labels, key);
659
RRDLABEL *lb = dictionary_acquired_item_value(acquired_item);
660
libnetdata/Makefile.am
+1
@@ -20,6 +20,7 @@ SUBDIRS = \
20
locks \
21
log \
22
onewayalloc \
23
+ parser \
24
popen \
25
procfile \
26
simple_pattern \
libnetdata/aral/aral.c
+8
@@ -465,6 +465,9 @@ static inline ARAL_PAGE *aral_acquire_a_free_slot(ARAL *ar TRACE_ALLOCATIONS_FUN
465
}
466
467
void *aral_mallocz_internal(ARAL *ar TRACE_ALLOCATIONS_FUNCTION_DEFINITION_PARAMS) {
468
+#ifdef FSANITIZE_ADDRESS
469
+ return mallocz(ar->config.requested_element_size);
470
+#endif
471
472
ARAL_PAGE *page = aral_acquire_a_free_slot(ar TRACE_ALLOCATIONS_FUNCTION_CALL_PARAMS);
473
@@ -614,6 +617,11 @@ static inline void aral_move_page_with_free_list___aral_lock_needed(ARAL *ar, AR
617
}
618
619
void aral_freez_internal(ARAL *ar, void *ptr TRACE_ALLOCATIONS_FUNCTION_DEFINITION_PARAMS) {
620
+#ifdef FSANITIZE_ADDRESS
621
+ freez(ptr);
622
+ return;
623
+#endif
624
+
625
if(unlikely(!ptr)) return;
626
627
// get the page pointer
libnetdata/inlined.h
+103
-10
@@ -21,25 +21,118 @@ typedef uint64_t kernel_uint_t;
21
// for faster execution, allow the compiler to inline
22
// these functions that are called thousands of times per second
23
24
-static inline uint32_t simple_hash(const char *name) {
24
+static inline uint32_t djb2_hash32(const char* name) {
25
unsigned char *s = (unsigned char *) name;
26
- uint32_t hval = 0x811c9dc5;
26
+ uint32_t hash = 5381;
27
+ while (*s)
28
+ hash = ((hash << 5) + hash) + (uint32_t) *s++; // hash * 33 + char
29
+ return hash;
30
+}
31
+
32
+static inline uint32_t pluginsd_parser_hash32(const char *name) {
33
+ unsigned char *s = (unsigned char *) name;
34
+ uint32_t hash = 0;
35
while (*s) {
28
- hval *= 16777619;
29
- hval ^= (uint32_t) *s++;
36
+ hash <<= 5;
37
+ hash += *s++ - ' ';
38
}
31
- return hval;
39
+ return hash;
40
}
41
34
-static inline uint32_t simple_uhash(const char *name) {
42
+// https://stackoverflow.com/a/107657
43
+static inline uint32_t larson_hash32(const char *name) {
44
unsigned char *s = (unsigned char *) name;
36
- uint32_t hval = 0x811c9dc5, c;
45
+ uint32_t hash = 0;
46
+ while (*s)
47
+ hash = hash * 101 + (uint32_t) *s++;
48
+ return hash;
49
+}
50
+
51
+// http://isthe.com/chongo/tech/comp/fnv/
52
+static inline uint32_t fnv1_hash32(const char *name) {
53
+ unsigned char *s = (unsigned char *) name;
54
+ uint32_t hash = 0x811c9dc5;
55
+ while (*s) {
56
+ hash *= 0x01000193; // 16777619
57
+ hash ^= (uint32_t) *s++;
58
+ }
59
+ return hash;
60
+}
61
+
62
+// http://isthe.com/chongo/tech/comp/fnv/
63
+static inline uint32_t fnv1a_hash32(const char *name) {
64
+ unsigned char *s = (unsigned char *) name;
65
+ uint32_t hash = 0x811c9dc5;
66
+ while (*s) {
67
+ hash ^= (uint32_t) *s++;
68
+ hash *= 0x01000193; // 16777619
69
+ }
70
+ return hash;
71
+}
72
+
73
+static inline uint32_t fnv1a_uhash32(const char *name) {
74
+ unsigned char *s = (unsigned char *) name;
75
+ uint32_t hash = 0x811c9dc5, c;
76
while ((c = *s++)) {
77
if (unlikely(c >= 'A' && c <= 'Z')) c += 'a' - 'A';
39
- hval *= 16777619;
40
- hval ^= c;
78
+ hash ^= c;
79
+ hash *= 0x01000193; // 16777619
80
}
42
- return hval;
81
+ return hash;
82
+}
83
+
84
+#define simple_hash(s) fnv1a_hash32(s)
85
+#define simple_uhash(s) fnv1a_uhash32(s)
86
+
87
+static inline size_t indexing_partition_old(Word_t ptr, Word_t modulo) {
88
+ size_t total = 0;
89
+
90
+ total += (ptr & 0xff) >> 0;
91
+ total += (ptr & 0xff00) >> 8;
92
+ total += (ptr & 0xff0000) >> 16;
93
+ total += (ptr & 0xff000000) >> 24;
94
+
95
+ if(sizeof(Word_t) > 4) {
96
+ total += (ptr & 0xff00000000) >> 32;
97
+ total += (ptr & 0xff0000000000) >> 40;
98
+ total += (ptr & 0xff000000000000) >> 48;
99
+ total += (ptr & 0xff00000000000000) >> 56;
100
+ }
101
+
102
+ return (total % modulo);
103
+}
104
+
105
+static uint32_t murmur32(uint32_t k) __attribute__((const));
106
+static inline uint32_t murmur32(uint32_t k) {
107
+ k ^= k >> 16;
108
+ k *= 0x85ebca6b;
109
+ k ^= k >> 13;
110
+ k *= 0xc2b2ae35;
111
+ k ^= k >> 16;
112
+
113
+ return k;
114
+}
115
+
116
+static uint64_t murmur64(uint64_t k) __attribute__((const));
117
+static inline uint64_t murmur64(uint64_t k) {
118
+ k ^= k >> 33;
119
+ k *= 0xff51afd7ed558ccdUL;
120
+ k ^= k >> 33;
121
+ k *= 0xc4ceb9fe1a85ec53UL;
122
+ k ^= k >> 33;
123
+
124
+ return k;
125
+}
126
+
127
+static inline size_t indexing_partition(Word_t ptr, Word_t modulo) __attribute__((const));
128
+static inline size_t indexing_partition(Word_t ptr, Word_t modulo) {
129
+#ifdef ENV64BIT
130
+ uint64_t hash = murmur64(ptr);
131
+ return hash % modulo;
132
+#else
133
+ uint32_t hash = murmur32(ptr);
134
+ return hash % modulo;
135
+#endif
136
}
137
138
static inline int str2i(const char *s) {
libnetdata/libnetdata.c
+11
-18
@@ -1883,17 +1883,20 @@ inline int config_isspace(char c)
1883
}
1884
1885
// split a text into words, respecting quotes
1886
-inline size_t quoted_strings_splitter(char *str, char **words, size_t max_words, int (*custom_isspace)(char), char *recover_input, char **recover_location, int max_recover)
1886
+inline size_t quoted_strings_splitter(char *str, char **words, size_t max_words, int (*custom_isspace)(char))
1887
{
1888
char *s = str, quote = 0;
1889
size_t i = 0;
1890
- int rec = 0;
1891
- char *recover = recover_input;
1890
1891
// skip all white space
1892
while (unlikely(custom_isspace(*s)))
1893
s++;
1894
1895
+ if(unlikely(!*s)) {
1896
+ words[i] = NULL;
1897
+ return 0;
1898
+ }
1899
+
1900
// check for quote
1901
if (unlikely(*s == '\'' || *s == '"')) {
1902
quote = *s; // remember the quote
@@ -1905,19 +1908,15 @@ inline size_t quoted_strings_splitter(char *str, char **words, size_t max_words,
1908
1909
// while we have something
1910
while (likely(*s)) {
1908
- // if it is escape
1911
+ // if it is an escape
1912
if (unlikely(*s == '\\' && s[1])) {
1913
s += 2;
1914
continue;
1915
}
1916
1914
- // if it is quote
1917
+ // if it is a quote
1918
else if (unlikely(*s == quote)) {
1919
quote = 0;
1917
- if (recover && rec < max_recover) {
1918
- recover_location[rec++] = s;
1919
- *recover++ = *s;
1920
- }
1920
*s = ' ';
1921
continue;
1922
}
@@ -1925,19 +1924,13 @@ inline size_t quoted_strings_splitter(char *str, char **words, size_t max_words,
1924
// if it is a space
1925
else if (unlikely(quote == 0 && custom_isspace(*s))) {
1926
// terminate the word
1928
- if (recover && rec < max_recover) {
1929
- if (!rec || recover_location[rec-1] != s) {
1930
- recover_location[rec++] = s;
1931
- *recover++ = *s;
1932
- }
1933
- }
1927
*s++ = '\0';
1928
1929
// skip all white space
1930
while (likely(custom_isspace(*s)))
1931
s++;
1932
1940
- // check for quote
1933
+ // check for a quote
1934
if (unlikely(*s == '\'' || *s == '"')) {
1935
quote = *s; // remember the quote
1936
s++; // skip the quote
@@ -1965,9 +1958,9 @@ inline size_t quoted_strings_splitter(char *str, char **words, size_t max_words,
1958
return i;
1959
}
1960
1968
-inline size_t pluginsd_split_words(char *str, char **words, size_t max_words, char *recover_input, char **recover_location, int max_recover)
1961
+inline size_t pluginsd_split_words(char *str, char **words, size_t max_words)
1962
{
1970
- return quoted_strings_splitter(str, words, max_words, pluginsd_space, recover_input, recover_location, max_recover);
1963
+ return quoted_strings_splitter(str, words, max_words, pluginsd_space);
1964
}
1965
1966
bool bitmap256_get_bit(BITMAP256 *ptr, uint8_t idx) {
libnetdata/libnetdata.h
+3
-54
@@ -485,8 +485,8 @@ void bitmap256_set_bit(BITMAP256 *ptr, uint8_t idx, bool value);
485
int config_isspace(char c);
486
int pluginsd_space(char c);
487
488
-size_t quoted_strings_splitter(char *str, char **words, size_t max_words, int (*custom_isspace)(char), char *recover_input, char **recover_location, int max_recover);
489
-size_t pluginsd_split_words(char *str, char **words, size_t max_words, char *recover_string, char **recover_location, int max_recover);
488
+size_t quoted_strings_splitter(char *str, char **words, size_t max_words, int (*custom_isspace)(char));
489
+size_t pluginsd_split_words(char *str, char **words, size_t max_words);
490
491
static inline char *get_word(char **words, size_t num_words, size_t index) {
492
if (index >= num_words)
@@ -547,6 +547,7 @@ extern char *netdata_configured_host_prefix;
547
#include "libnetdata/aral/aral.h"
548
#include "onewayalloc/onewayalloc.h"
549
#include "worker_utilization/worker_utilization.h"
550
+#include "parser/parser.h"
551
552
// BEWARE: Outside of the C code this also exists in alarm-notify.sh
553
#define DEFAULT_CLOUD_BASE_URL "https://api.netdata.cloud"
@@ -609,58 +610,6 @@ static inline PPvoid_t JudyLLastThenPrev(Pcvoid_t PArray, Word_t * PIndex, bool
610
return JudyLPrev(PArray, PIndex, PJE0);
611
}
612
612
-static inline size_t indexing_partition_old(Word_t ptr, Word_t modulo) {
613
- size_t total = 0;
614
-
615
- total += (ptr & 0xff) >> 0;
616
- total += (ptr & 0xff00) >> 8;
617
- total += (ptr & 0xff0000) >> 16;
618
- total += (ptr & 0xff000000) >> 24;
619
-
620
- if(sizeof(Word_t) > 4) {
621
- total += (ptr & 0xff00000000) >> 32;
622
- total += (ptr & 0xff0000000000) >> 40;
623
- total += (ptr & 0xff000000000000) >> 48;
624
- total += (ptr & 0xff00000000000000) >> 56;
625
- }
626
-
627
- return (total % modulo);
628
-}
629
-
630
-static uint32_t murmur32(uint32_t h) __attribute__((const));
631
-static inline uint32_t murmur32(uint32_t h) {
632
- h ^= h >> 16;
633
- h *= 0x85ebca6b;
634
- h ^= h >> 13;
635
- h *= 0xc2b2ae35;
636
- h ^= h >> 16;
637
-
638
- return h;
639
-}
640
-
641
-static uint64_t murmur64(uint64_t h) __attribute__((const));
642
-static inline uint64_t murmur64(uint64_t k) {
643
- k ^= k >> 33;
644
- k *= 0xff51afd7ed558ccdUL;
645
- k ^= k >> 33;
646
- k *= 0xc4ceb9fe1a85ec53UL;
647
- k ^= k >> 33;
648
-
649
- return k;
650
-}
651
-
652
-static inline size_t indexing_partition(Word_t ptr, Word_t modulo) __attribute__((const));
653
-static inline size_t indexing_partition(Word_t ptr, Word_t modulo) {
654
- if(sizeof(Word_t) == 8) {
655
- uint64_t hash = murmur64(ptr);
656
- return hash % modulo;
657
- }
658
- else {
659
- uint32_t hash = murmur32(ptr);
660
- return hash % modulo;
661
- }
662
-}
663
-
613
typedef enum {
614
TIMING_STEP_INTERNAL = 0,
615
libnetdata/onewayalloc/onewayalloc.c
+9
@@ -97,6 +97,10 @@ ONEWAYALLOC *onewayalloc_create(size_t size_hint) {
97
}
98
99
void *onewayalloc_mallocz(ONEWAYALLOC *owa, size_t size) {
100
+#ifdef FSANITIZE_ADDRESS
101
+ return mallocz(size);
102
+#endif
103
+
104
OWA_PAGE *head = (OWA_PAGE *)owa;
105
OWA_PAGE *page = head->last;
106
@@ -142,6 +146,11 @@ void *onewayalloc_memdupz(ONEWAYALLOC *owa, const void *src, size_t size) {
146
}
147
148
void onewayalloc_freez(ONEWAYALLOC *owa __maybe_unused, const void *ptr __maybe_unused) {
149
+#ifdef FSANITIZE_ADDRESS
150
+ freez((void *)ptr);
151
+ return;
152
+#endif
153
+
154
#ifdef NETDATA_INTERNAL_CHECKS
155
// allow the caller to call us for a mallocz() allocation
156
// so try to find it in our memory and if it is not there
libnetdata/parser/Makefile.am
renamed
libnetdata/parser/README.md
renamed
libnetdata/parser/parser.c
new
+225
@@ -0,0 +1,225 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "parser.h"
4
+#include "collectors/plugins.d/pluginsd_parser.h"
5
+
6
+static inline int find_first_keyword(const char *src, char *dst, int dst_size, int (*custom_isspace)(char)) {
7
+ const char *s = src, *keyword_start;
8
+
9
+ while (unlikely(custom_isspace(*s))) s++;
10
+ keyword_start = s;
11
+
12
+ while (likely(*s && !custom_isspace(*s)) && dst_size > 1) {
13
+ *dst++ = *s++;
14
+ dst_size--;
15
+ }
16
+ *dst = '\0';
17
+ return dst_size == 0 ? 0 : (int) (s - keyword_start);
18
+}
19
+
20
+/*
21
+ * Initialize a parser
22
+ * user : as defined by the user, will be shared across calls
23
+ * input : main input stream (auto detect stream -- file, socket, pipe)
24
+ * buffer : This is the buffer to be used (if null a buffer of size will be allocated)
25
+ * size : buffer size either passed or will be allocated
26
+ * If the buffer is auto allocated, it will auto freed when the parser is destroyed
27
+ *
28
+ *
29
+ */
30
+
31
+PARSER *parser_init(void *user, FILE *fp_input, FILE *fp_output, int fd,
32
+ PARSER_INPUT_TYPE flags, void *ssl __maybe_unused)
33
+{
34
+ PARSER *parser;
35
+
36
+ parser = callocz(1, sizeof(*parser));
37
+ parser->user = user;
38
+ parser->fd = fd;
39
+ parser->fp_input = fp_input;
40
+ parser->fp_output = fp_output;
41
+#ifdef ENABLE_HTTPS
42
+ parser->ssl_output = ssl;
43
+#endif
44
+ parser->flags = flags;
45
+ parser->worker_job_next_id = WORKER_PARSER_FIRST_JOB;
46
+
47
+ return parser;
48
+}
49
+
50
+
51
+static inline PARSER_KEYWORD *parser_find_keyword(PARSER *parser, const char *command) {
52
+ uint32_t hash = parser_hash_function(command);
53
+ uint32_t slot = hash % PARSER_KEYWORDS_HASHTABLE_SIZE;
54
+ PARSER_KEYWORD *t = parser->keywords.hashtable[slot];
55
+
56
+ if(likely(t && strcmp(t->keyword, command) == 0))
57
+ return t;
58
+
59
+ return NULL;
60
+}
61
+
62
+/*
63
+ * Add a keyword and the corresponding function that will be called
64
+ * Multiple functions may be added
65
+ * Input : keyword
66
+ * : callback function
67
+ * : flags
68
+ * Output: > 0 registered function number
69
+ * : 0 Error
70
+ */
71
+
72
+void parser_add_keyword(PARSER *parser, char *keyword, keyword_function func) {
73
+ if(unlikely(!parser || !keyword || !*keyword || !func))
74
+ fatal("PARSER: invalid parameters");
75
+
76
+ PARSER_KEYWORD *t = callocz(1, sizeof(*t));
77
+ t->worker_job_id = parser->worker_job_next_id++;
78
+ t->keyword = strdupz(keyword);
79
+ t->func = func;
80
+
81
+ uint32_t hash = parser_hash_function(keyword);
82
+ uint32_t slot = hash % PARSER_KEYWORDS_HASHTABLE_SIZE;
83
+
84
+ if(unlikely(parser->keywords.hashtable[slot]))
85
+ fatal("PARSER: hashtable collision between keyword '%s' and '%s' on slot %u. "
86
+ "Change the hashtable size and / or the hashing function. "
87
+ "Run the unit test to find the optimal values.",
88
+ parser->keywords.hashtable[slot]->keyword,
89
+ t->keyword,
90
+ slot
91
+ );
92
+
93
+ parser->keywords.hashtable[slot] = t;
94
+
95
+ worker_register_job_name(t->worker_job_id, t->keyword);
96
+}
97
+
98
+/*
99
+ * Cleanup a previously allocated parser
100
+ */
101
+
102
+void parser_destroy(PARSER *parser)
103
+{
104
+ if (unlikely(!parser))
105
+ return;
106
+
107
+ dictionary_destroy(parser->inflight.functions);
108
+
109
+ // Remove keywords
110
+ for(size_t i = 0 ; i < PARSER_KEYWORDS_HASHTABLE_SIZE; i++) {
111
+ PARSER_KEYWORD *t = parser->keywords.hashtable[i];
112
+ if (t) {
113
+ freez(t->keyword);
114
+ freez(t);
115
+ }
116
+ }
117
+
118
+ freez(parser);
119
+}
120
+
121
+
122
+/*
123
+ * Fetch the next line to process
124
+ *
125
+ */
126
+
127
+int parser_next(PARSER *parser, char *buffer, size_t buffer_size)
128
+{
129
+ char *tmp = fgets(buffer, (int)buffer_size, (FILE *)parser->fp_input);
130
+
131
+ if (unlikely(!tmp)) {
132
+ if (feof((FILE *)parser->fp_input))
133
+ error("PARSER: read failed: end of file");
134
+
135
+ else if (ferror((FILE *)parser->fp_input))
136
+ error("PARSER: read failed: input error");
137
+
138
+ else
139
+ error("PARSER: read failed: unknown error");
140
+
141
+ return 1;
142
+ }
143
+
144
+ return 0;
145
+}
146
+
147
+
148
+/*
149
+* Takes an initialized parser object that has an unprocessed entry (by calling parser_next)
150
+* and if it contains a valid keyword, it will execute all the callbacks
151
+*
152
+*/
153
+
154
+inline int parser_action(PARSER *parser, char *input)
155
+{
156
+ parser->line++;
157
+
158
+ if(unlikely(parser->flags & PARSER_DEFER_UNTIL_KEYWORD)) {
159
+ char command[PLUGINSD_LINE_MAX + 1];
160
+ bool has_keyword = find_first_keyword(input, command, PLUGINSD_LINE_MAX, pluginsd_space);
161
+
162
+ if(!has_keyword || strcmp(command, parser->defer.end_keyword) != 0) {
163
+ if(parser->defer.response) {
164
+ buffer_strcat(parser->defer.response, input);
165
+ if(buffer_strlen(parser->defer.response) > 10 * 1024 * 1024) {
166
+ // more than 10MB of data
167
+ // a bad plugin that did not send the end_keyword
168
+ internal_error(true, "PLUGINSD: deferred response is too big (%zu bytes). Stopping this plugin.", buffer_strlen(parser->defer.response));
169
+ return 1;
170
+ }
171
+ }
172
+ return 0;
173
+ }
174
+ else {
175
+ // call the action
176
+ parser->defer.action(parser, parser->defer.action_data);
177
+
178
+ // empty everything
179
+ parser->defer.action = NULL;
180
+ parser->defer.action_data = NULL;
181
+ parser->defer.end_keyword = NULL;
182
+ parser->defer.response = NULL;
183
+ parser->flags &= ~PARSER_DEFER_UNTIL_KEYWORD;
184
+ }
185
+ return 0;
186
+ }
187
+
188
+ char *words[PLUGINSD_MAX_WORDS];
189
+ size_t num_words = pluginsd_split_words(input, words, PLUGINSD_MAX_WORDS);
190
+ const char *command = get_word(words, num_words, 0);
191
+
192
+ if(unlikely(!command))
193
+ return 0;
194
+
195
+ PARSER_RC rc;
196
+ PARSER_KEYWORD *t = parser_find_keyword(parser, command);
197
+ if(likely(t)) {
198
+ worker_is_busy(t->worker_job_id);
199
+ rc = (*t->func)(words, num_words, parser->user);
200
+ worker_is_idle();
201
+ }
202
+ else
203
+ rc = PARSER_RC_ERROR;
204
+
205
+#ifdef NETDATA_INTERNAL_CHECKS
206
+ if(rc == PARSER_RC_ERROR) {
207
+ BUFFER *wb = buffer_create(PLUGINSD_LINE_MAX, NULL);
208
+ for(size_t i = 0; i < num_words ;i++) {
209
+ if(i) buffer_fast_strcat(wb, " ", 1);
210
+
211
+ buffer_fast_strcat(wb, "\"", 1);
212
+ const char *s = get_word(words, num_words, i);
213
+ buffer_strcat(wb, s?s:"");
214
+ buffer_fast_strcat(wb, "\"", 1);
215
+ }
216
+
217
+ internal_error(true, "PLUGINSD: parser_action('%s') failed on line %zu: { %s } (quotes added to show parsing)",
218
+ command, parser->line, buffer_tostring(wb));
219
+
220
+ buffer_free(wb);
221
+ }
222
+#endif
223
+
224
+ return (rc == PARSER_RC_ERROR || rc == PARSER_RC_STOP);
225
+}
libnetdata/parser/parser.h
renamed
+21
-51
@@ -3,79 +3,56 @@
3
#ifndef NETDATA_INCREMENTAL_PARSER_H
4
#define NETDATA_INCREMENTAL_PARSER_H 1
5
6
-#include "daemon/common.h"
6
+#include "../libnetdata.h"
7
8
-#define PARSER_MAX_CALLBACKS 20
9
-#define PARSER_MAX_RECOVER_KEYWORDS 128
8
#define WORKER_PARSER_FIRST_JOB 3
9
10
// this has to be in-sync with the same at receiver.c
11
#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 3)
12
13
+#define PARSER_KEYWORDS_HASHTABLE_SIZE 73 // unittest finds this magic number
14
+//#define parser_hash_function(s) djb2_hash32(s)
15
+//#define parser_hash_function(s) fnv1_hash32(s)
16
+//#define parser_hash_function(s) fnv1a_hash32(s)
17
+//#define parser_hash_function(s) larson_hash32(s)
18
+#define parser_hash_function(s) pluginsd_parser_hash32(s)
19
+
20
// PARSER return codes
16
-typedef enum parser_rc {
21
+typedef enum __attribute__ ((__packed__)) parser_rc {
22
PARSER_RC_OK, // Callback was successful, go on
23
PARSER_RC_STOP, // Callback says STOP
24
PARSER_RC_ERROR // Callback failed (abort rest of callbacks)
25
} PARSER_RC;
26
22
-typedef enum parser_input_type {
27
+typedef enum __attribute__ ((__packed__)) parser_input_type {
28
PARSER_INPUT_SPLIT = (1 << 1),
24
- PARSER_INPUT_KEEP_ORIGINAL = (1 << 2),
25
- PARSER_INPUT_PROCESSED = (1 << 3),
26
- PARSER_NO_PARSE_INIT = (1 << 4),
27
- PARSER_NO_ACTION_INIT = (1 << 5),
28
- PARSER_DEFER_UNTIL_KEYWORD = (1 << 6),
29
+ PARSER_DEFER_UNTIL_KEYWORD = (1 << 2),
30
} PARSER_INPUT_TYPE;
31
31
-#define PARSER_INPUT_FULL (PARSER_INPUT_SPLIT|PARSER_INPUT_ORIGINAL)
32
-
32
typedef PARSER_RC (*keyword_function)(char **words, size_t num_words, void *user_data);
33
34
typedef struct parser_keyword {
36
- size_t worker_job_id;
37
- char *keyword;
38
- uint32_t keyword_hash;
39
- int func_no;
40
- keyword_function func[PARSER_MAX_CALLBACKS+1];
41
- struct parser_keyword *next;
35
+ size_t worker_job_id;
36
+ char *keyword;
37
+ keyword_function func;
38
} PARSER_KEYWORD;
39
44
-typedef struct parser_data {
45
- char *line;
46
- struct parser_data *next;
47
-} PARSER_DATA;
48
-
49
-typedef void (*parser_cleanup_t)(void *user);
50
-
40
typedef struct parser {
41
size_t worker_job_next_id;
42
uint8_t version; // Parser version
54
- RRDHOST *host;
43
int fd; // Socket
44
FILE *fp_input; // Input source e.g. stream
45
FILE *fp_output; // Stream to send commands to plugin
46
#ifdef ENABLE_HTTPS
47
struct netdata_ssl *ssl_output;
48
#endif
61
- PARSER_DATA *data; // extra input
62
- PARSER_KEYWORD *keyword; // List of parse keywords and functions
63
- void *user; // User defined structure to hold extra state between calls
64
- parser_cleanup_t user_cleanup_cb;
49
+ void *user; // User defined structure to hold extra state between calls
50
uint32_t flags;
51
size_t line;
52
68
- char *(*read_function)(char *buffer, long unsigned int, void *input);
69
- int (*eof_function)(void *input);
70
- keyword_function unknown_function;
71
- char buffer[PLUGINSD_LINE_MAX];
72
- char *recover_location[PARSER_MAX_RECOVER_KEYWORDS+1];
73
- char recover_input[PARSER_MAX_RECOVER_KEYWORDS];
74
-#ifdef ENABLE_HTTPS
75
- int bytesleft;
76
- char tmpbuffer[PLUGINSD_LINE_MAX];
77
- char *readfrom;
78
-#endif
53
+ struct {
54
+ PARSER_KEYWORD *hashtable[PARSER_KEYWORDS_HASHTABLE_SIZE];
55
+ } keywords;
56
57
struct {
58
const char *end_keyword;
@@ -88,20 +65,13 @@ typedef struct parser {
65
DICTIONARY *functions;
66
usec_t smaller_timeout;
67
} inflight;
91
-
68
} PARSER;
69
94
-int find_first_keyword(const char *str, char *keyword, int max_size, int (*custom_isspace)(char));
95
-
96
-PARSER *parser_init(RRDHOST *host, void *user, parser_cleanup_t cleanup_cb, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl);
97
-int parser_add_keyword(PARSER *working_parser, char *keyword, keyword_function func);
98
-int parser_next(PARSER *working_parser);
70
+PARSER *parser_init(void *user, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl);
71
+void parser_add_keyword(PARSER *working_parser, char *keyword, keyword_function func);
72
+int parser_next(PARSER *working_parser, char *buffer, size_t buffer_size);
73
int parser_action(PARSER *working_parser, char *input);
100
-int parser_push(PARSER *working_parser, char *line);
74
void parser_destroy(PARSER *working_parser);
102
-int parser_recover_input(PARSER *working_parser);
103
-
104
-size_t pluginsd_process(RRDHOST *host, struct plugind *cd, FILE *fp_plugin_input, FILE *fp_plugin_output, int trust_durations);
75
76
PARSER_RC pluginsd_set(char **words, size_t num_words, void *user);
77
PARSER_RC pluginsd_begin(char **words, size_t num_words, void *user);
libnetdata/string/string.c
+6
-2
@@ -300,16 +300,20 @@ void string_freez(STRING *string) {
300
string_stats_atomic_increment(releases);
301
}
302
303
-size_t string_strlen(STRING *string) {
303
+inline size_t string_strlen(STRING *string) {
304
if(unlikely(!string)) return 0;
305
return string->length - 1;
306
}
307
308
-const char *string2str(STRING *string) {
308
+inline const char *string2str(STRING *string) {
309
if(unlikely(!string)) return "";
310
return string->str;
311
}
312
313
+int string_strcmp(STRING *string, const char *s) {
314
+ return strcmp(string2str(string), s);
315
+}
316
+
317
STRING *string_2way_merge(STRING *a, STRING *b) {
318
static STRING *X = NULL;
319
libnetdata/string/string.h
+1
@@ -13,6 +13,7 @@ STRING *string_dup(STRING *string);
13
void string_freez(STRING *string);
14
size_t string_strlen(STRING *string);
15
const char *string2str(STRING *string) NEVERNULL;
16
+int string_strcmp(STRING *string, const char *s);
17
18
// keep common prefix/suffix and replace everything else with [x]
19
STRING *string_2way_merge(STRING *a, STRING *b);
parser/parser.c
deleted
-396
@@ -1,396 +0,0 @@
1
-// SPDX-License-Identifier: GPL-3.0-or-later
2
-
3
-#include "parser.h"
4
-#include "collectors/plugins.d/pluginsd_parser.h"
5
-
6
-inline int find_first_keyword(const char *str, char *keyword, int max_size, int (*custom_isspace)(char))
7
-{
8
- const char *s = str, *keyword_start;
9
-
10
- while (unlikely(custom_isspace(*s))) s++;
11
- keyword_start = s;
12
-
13
- while (likely(*s && !custom_isspace(*s)) && max_size > 1) {
14
- *keyword++ = *s++;
15
- max_size--;
16
- }
17
- *keyword = '\0';
18
- return max_size == 0 ? 0 : (int) (s - keyword_start);
19
-}
20
-
21
-/*
22
- * Initialize a parser
23
- * user : as defined by the user, will be shared across calls
24
- * input : main input stream (auto detect stream -- file, socket, pipe)
25
- * buffer : This is the buffer to be used (if null a buffer of size will be allocated)
26
- * size : buffer size either passed or will be allocated
27
- * If the buffer is auto allocated, it will auto freed when the parser is destroyed
28
- *
29
- *
30
- */
31
-
32
-PARSER *parser_init(RRDHOST *host, void *user, parser_cleanup_t cleanup_cb, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl __maybe_unused)
33
-{
34
- PARSER *parser;
35
-
36
- parser = callocz(1, sizeof(*parser));
37
- parser->user = user;
38
- parser->user_cleanup_cb = cleanup_cb;
39
- parser->fd = fd;
40
- parser->fp_input = fp_input;
41
- parser->fp_output = fp_output;
42
-#ifdef ENABLE_HTTPS
43
- parser->ssl_output = ssl;
44
-#endif
45
- parser->flags = flags;
46
- parser->host = host;
47
- parser->worker_job_next_id = WORKER_PARSER_FIRST_JOB;
48
- inflight_functions_init(parser);
49
-
50
-#ifdef ENABLE_HTTPS
51
- parser->bytesleft = 0;
52
- parser->readfrom = NULL;
53
-#endif
54
-
55
- if (unlikely(!(flags & PARSER_NO_PARSE_INIT))) {
56
- parser_add_keyword(parser, PLUGINSD_KEYWORD_FLUSH, pluginsd_flush);
57
- parser_add_keyword(parser, PLUGINSD_KEYWORD_CHART, pluginsd_chart);
58
- parser_add_keyword(parser, PLUGINSD_KEYWORD_CHART_DEFINITION_END, pluginsd_chart_definition_end);
59
- parser_add_keyword(parser, PLUGINSD_KEYWORD_DIMENSION, pluginsd_dimension);
60
- parser_add_keyword(parser, PLUGINSD_KEYWORD_DISABLE, pluginsd_disable);
61
- parser_add_keyword(parser, PLUGINSD_KEYWORD_VARIABLE, pluginsd_variable);
62
- parser_add_keyword(parser, PLUGINSD_KEYWORD_LABEL, pluginsd_label);
63
- parser_add_keyword(parser, PLUGINSD_KEYWORD_OVERWRITE, pluginsd_overwrite);
64
- parser_add_keyword(parser, PLUGINSD_KEYWORD_END, pluginsd_end);
65
- parser_add_keyword(parser, PLUGINSD_KEYWORD_CLABEL_COMMIT, pluginsd_clabel_commit);
66
- parser_add_keyword(parser, PLUGINSD_KEYWORD_CLABEL, pluginsd_clabel);
67
- parser_add_keyword(parser, PLUGINSD_KEYWORD_BEGIN, pluginsd_begin);
68
- parser_add_keyword(parser, PLUGINSD_KEYWORD_SET, pluginsd_set);
69
-
70
- parser_add_keyword(parser, PLUGINSD_KEYWORD_FUNCTION, pluginsd_function);
71
- parser_add_keyword(parser, PLUGINSD_KEYWORD_FUNCTION_RESULT_BEGIN, pluginsd_function_result_begin);
72
-
73
- parser_add_keyword(parser, PLUGINSD_KEYWORD_REPLAY_BEGIN, pluginsd_replay_begin);
74
- parser_add_keyword(parser, PLUGINSD_KEYWORD_REPLAY_SET, pluginsd_replay_set);
75
- parser_add_keyword(parser, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, pluginsd_replay_rrddim_collection_state);
76
- parser_add_keyword(parser, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE, pluginsd_replay_rrdset_collection_state);
77
- parser_add_keyword(parser, PLUGINSD_KEYWORD_REPLAY_END, pluginsd_replay_end);
78
-
79
- parser_add_keyword(parser, PLUGINSD_KEYWORD_BEGIN_V2, pluginsd_begin_v2);
80
- parser_add_keyword(parser, PLUGINSD_KEYWORD_SET_V2, pluginsd_set_v2);
81
- parser_add_keyword(parser, PLUGINSD_KEYWORD_END_V2, pluginsd_end_v2);
82
- }
83
-
84
- return parser;
85
-}
86
-
87
-
88
-/*
89
- * Push a new line into the parsing stream
90
- *
91
- * This line will be the next one to process ie the next fetch will get this one
92
- *
93
- */
94
-
95
-int parser_push(PARSER *parser, char *line)
96
-{
97
- PARSER_DATA *tmp_parser_data;
98
-
99
- if (unlikely(!parser))
100
- return 1;
101
-
102
- if (unlikely(!line))
103
- return 0;
104
-
105
- tmp_parser_data = callocz(1, sizeof(*tmp_parser_data));
106
- tmp_parser_data->line = strdupz(line);
107
- tmp_parser_data->next = parser->data;
108
- parser->data = tmp_parser_data;
109
-
110
- return 0;
111
-}
112
-
113
-/*
114
- * Add a keyword and the corresponding function that will be called
115
- * Multiple functions may be added
116
- * Input : keyword
117
- * : callback function
118
- * : flags
119
- * Output: > 0 registered function number
120
- * : 0 Error
121
- */
122
-
123
-int parser_add_keyword(PARSER *parser, char *keyword, keyword_function func)
124
-{
125
- PARSER_KEYWORD *tmp_keyword;
126
-
127
- if (strcmp(keyword, "_read") == 0) {
128
- parser->read_function = (void *) func;
129
- return 0;
130
- }
131
-
132
- if (strcmp(keyword, "_eof") == 0) {
133
- parser->eof_function = (void *) func;
134
- return 0;
135
- }
136
-
137
- if (strcmp(keyword, "_unknown") == 0) {
138
- parser->unknown_function = (void *) func;
139
- return 0;
140
- }
141
-
142
- uint32_t keyword_hash = simple_hash(keyword);
143
-
144
- tmp_keyword = parser->keyword;
145
-
146
- while (tmp_keyword) {
147
- if (tmp_keyword->keyword_hash == keyword_hash && (!strcmp(tmp_keyword->keyword, keyword))) {
148
- if (tmp_keyword->func_no == PARSER_MAX_CALLBACKS)
149
- return 0;
150
- tmp_keyword->func[tmp_keyword->func_no++] = (void *) func;
151
- return tmp_keyword->func_no;
152
- }
153
- tmp_keyword = tmp_keyword->next;
154
- }
155
-
156
- tmp_keyword = callocz(1, sizeof(*tmp_keyword));
157
-
158
- tmp_keyword->worker_job_id = parser->worker_job_next_id++;
159
- tmp_keyword->keyword = strdupz(keyword);
160
- tmp_keyword->keyword_hash = keyword_hash;
161
- tmp_keyword->func[tmp_keyword->func_no++] = (void *) func;
162
-
163
- worker_register_job_name(tmp_keyword->worker_job_id, tmp_keyword->keyword);
164
-
165
- tmp_keyword->next = parser->keyword;
166
- parser->keyword = tmp_keyword;
167
- return tmp_keyword->func_no;
168
-}
169
-
170
-/*
171
- * Cleanup a previously allocated parser
172
- */
173
-
174
-void parser_destroy(PARSER *parser)
175
-{
176
- if (unlikely(!parser))
177
- return;
178
-
179
- dictionary_destroy(parser->inflight.functions);
180
-
181
- PARSER_KEYWORD *tmp_keyword, *tmp_keyword_next;
182
- PARSER_DATA *tmp_parser_data, *tmp_parser_data_next;
183
-
184
- // Remove keywords
185
- tmp_keyword = parser->keyword;
186
- while (tmp_keyword) {
187
- tmp_keyword_next = tmp_keyword->next;
188
- freez(tmp_keyword->keyword);
189
- freez(tmp_keyword);
190
- tmp_keyword = tmp_keyword_next;
191
- }
192
-
193
- // Remove pushed data if any
194
- tmp_parser_data = parser->data;
195
- while (tmp_parser_data) {
196
- tmp_parser_data_next = tmp_parser_data->next;
197
- freez(tmp_parser_data->line);
198
- freez(tmp_parser_data);
199
- tmp_parser_data = tmp_parser_data_next;
200
- }
201
-
202
- freez(parser);
203
-}
204
-
205
-
206
-/*
207
- * Fetch the next line to process
208
- *
209
- */
210
-
211
-int parser_next(PARSER *parser)
212
-{
213
- char *tmp = NULL;
214
-
215
- if (unlikely(!parser))
216
- return 1;
217
-
218
- parser->flags &= ~(PARSER_INPUT_PROCESSED);
219
-
220
- PARSER_DATA *tmp_parser_data = parser->data;
221
-
222
- if (unlikely(tmp_parser_data)) {
223
- strncpyz(parser->buffer, tmp_parser_data->line, PLUGINSD_LINE_MAX);
224
- parser->data = tmp_parser_data->next;
225
- freez(tmp_parser_data->line);
226
- freez(tmp_parser_data);
227
- return 0;
228
- }
229
-
230
- if (unlikely(parser->read_function))
231
- tmp = parser->read_function(parser->buffer, PLUGINSD_LINE_MAX, parser->fp_input);
232
- else if(likely(parser->fp_input))
233
- tmp = fgets(parser->buffer, PLUGINSD_LINE_MAX, (FILE *)parser->fp_input);
234
- else
235
- tmp = NULL;
236
-
237
- if (unlikely(!tmp)) {
238
- if (unlikely(parser->eof_function)) {
239
- int rc = parser->eof_function(parser->fp_input);
240
- error("read failed: user defined function returned %d", rc);
241
- }
242
- else {
243
- if (feof((FILE *)parser->fp_input))
244
- error("read failed: end of file");
245
- else if (ferror((FILE *)parser->fp_input))
246
- error("read failed: input error");
247
- else
248
- error("read failed: unknown error");
249
- }
250
- return 1;
251
- }
252
- return 0;
253
-}
254
-
255
-
256
-/*
257
-* Takes an initialized parser object that has an unprocessed entry (by calling parser_next)
258
-* and if it contains a valid keyword, it will execute all the callbacks
259
-*
260
-*/
261
-
262
-inline int parser_action(PARSER *parser, char *input)
263
-{
264
- parser->line++;
265
-
266
- PARSER_RC rc = PARSER_RC_OK;
267
- char *words[PLUGINSD_MAX_WORDS];
268
- char command[PLUGINSD_LINE_MAX + 1];
269
- keyword_function action_function;
270
- keyword_function *action_function_list = NULL;
271
-
272
- if (unlikely(!parser)) {
273
- internal_error(true, "parser is NULL");
274
- return 1;
275
- }
276
-
277
- parser->recover_location[0] = 0x0;
278
-
279
- // if not direct input check if we have reprocessed this
280
- if (unlikely(!input && parser->flags & PARSER_INPUT_PROCESSED))
281
- return 0;
282
-
283
- PARSER_KEYWORD *tmp_keyword = parser->keyword;
284
- if (unlikely(!tmp_keyword)) {
285
- internal_error(true, "called without a keyword");
286
- return 1;
287
- }
288
-
289
- if (unlikely(!input))
290
- input = parser->buffer;
291
-
292
- if(unlikely(parser->flags & PARSER_DEFER_UNTIL_KEYWORD)) {
293
- bool has_keyword = find_first_keyword(input, command, PLUGINSD_LINE_MAX, pluginsd_space);
294
-
295
- if(!has_keyword || strcmp(command, parser->defer.end_keyword) != 0) {
296
- if(parser->defer.response) {
297
- buffer_strcat(parser->defer.response, input);
298
- if(buffer_strlen(parser->defer.response) > 10 * 1024 * 1024) {
299
- // more than 10MB of data
300
- // a bad plugin that did not send the end_keyword
301
- internal_error(true, "PLUGINSD: deferred response is too big (%zu bytes). Stopping this plugin.", buffer_strlen(parser->defer.response));
302
- return 1;
303
- }
304
- }
305
- return 0;
306
- }
307
- else {
308
- // call the action
309
- parser->defer.action(parser, parser->defer.action_data);
310
-
311
- // empty everything
312
- parser->defer.action = NULL;
313
- parser->defer.action_data = NULL;
314
- parser->defer.end_keyword = NULL;
315
- parser->defer.response = NULL;
316
- parser->flags &= ~PARSER_DEFER_UNTIL_KEYWORD;
317
- }
318
- return 0;
319
- }
320
-
321
- if (unlikely(!find_first_keyword(input, command, PLUGINSD_LINE_MAX, pluginsd_space)))
322
- return 0;
323
-
324
- size_t num_words = 0;
325
- if ((parser->flags & PARSER_INPUT_KEEP_ORIGINAL) == PARSER_INPUT_KEEP_ORIGINAL)
326
- num_words = pluginsd_split_words(input, words, PLUGINSD_MAX_WORDS, parser->recover_input, parser->recover_location, PARSER_MAX_RECOVER_KEYWORDS);
327
- else
328
- num_words = pluginsd_split_words(input, words, PLUGINSD_MAX_WORDS, NULL, NULL, 0);
329
-
330
- uint32_t command_hash = simple_hash(command);
331
-
332
- size_t worker_job_id = WORKER_UTILIZATION_MAX_JOB_TYPES + 1; // set an invalid value by default
333
- while(tmp_keyword) {
334
- if (command_hash == tmp_keyword->keyword_hash && (!strcmp(command, tmp_keyword->keyword))) {
335
- action_function_list = &tmp_keyword->func[0];
336
- worker_job_id = tmp_keyword->worker_job_id;
337
- break;
338
- }
339
- tmp_keyword = tmp_keyword->next;
340
- }
341
-
342
- if (unlikely(!action_function_list)) {
343
- if (unlikely(parser->unknown_function))
344
- rc = parser->unknown_function(words, num_words, parser->user);
345
- else
346
- rc = PARSER_RC_ERROR;
347
- }
348
- else {
349
- worker_is_busy(worker_job_id);
350
- while ((action_function = *action_function_list) != NULL) {
351
- rc = action_function(words, num_words, parser->user);
352
- if (unlikely(rc == PARSER_RC_ERROR || rc == PARSER_RC_STOP))
353
- break;
354
-
355
- action_function_list++;
356
- }
357
- worker_is_idle();
358
- }
359
-
360
- if (likely(input == parser->buffer))
361
- parser->flags |= PARSER_INPUT_PROCESSED;
362
-
363
-#ifdef NETDATA_INTERNAL_CHECKS
364
- if(rc == PARSER_RC_ERROR) {
365
- BUFFER *wb = buffer_create(PLUGINSD_LINE_MAX, NULL);
366
- for(size_t i = 0; i < num_words ;i++) {
367
- if(i) buffer_fast_strcat(wb, " ", 1);
368
-
369
- buffer_fast_strcat(wb, "\"", 1);
370
- const char *s = get_word(words, num_words, i);
371
- buffer_strcat(wb, s?s:"");
372
- buffer_fast_strcat(wb, "\"", 1);
373
- }
374
-
375
- internal_error(true, "PLUGINSD: parser_action('%s') failed on line %zu: { %s } (quotes added to show parsing)",
376
- command, parser->line, buffer_tostring(wb));
377
-
378
- buffer_free(wb);
379
- }
380
-#endif
381
-
382
- return (rc == PARSER_RC_ERROR);
383
-}
384
-
385
-inline int parser_recover_input(PARSER *parser)
386
-{
387
- if (unlikely(!parser))
388
- return 1;
389
-
390
- for(int i=0; i < PARSER_MAX_RECOVER_KEYWORDS && parser->recover_location[i]; i++)
391
- *(parser->recover_location[i]) = parser->recover_input[i];
392
-
393
- parser->recover_location[0] = 0x0;
394
-
395
- return 0;
396
-}
streaming/receiver.c
+4
-12
@@ -1,7 +1,6 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
#include "rrdpush.h"
4
-#include "parser/parser.h"
4
5
// IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly
6
#define WORKER_RECEIVER_JOB_BYTES_READ (WORKER_PARSER_FIRST_JOB - 1)
@@ -332,10 +331,6 @@ static void streaming_parser_thread_cleanup(void *ptr) {
331
332
bool plugin_is_enabled(struct plugind *cd);
333
335
-void streaming_parser_cleanup(void *user) {
336
- pluginsd_cleanup_v2(user);
337
-}
338
-
334
static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) {
335
size_t result;
336
@@ -347,7 +342,10 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, i
342
.trust_durations = 1
343
};
344
350
- PARSER *parser = parser_init(rpt->host, &user, streaming_parser_cleanup, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl);
345
+ PARSER *parser = parser_init(&user, NULL, NULL, fd,
346
+ PARSER_INPUT_SPLIT, ssl);
347
+
348
+ pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);
349
350
rrd_collector_started();
351
@@ -728,16 +726,12 @@ static int rrdpush_receive(struct receiver_state *rpt)
726
727
struct plugind cd = {
728
.update_every = default_rrd_update_every,
731
- .serial_failures = 0,
732
- .successful_collections = 0,
729
.unsafe = {
730
.spinlock = NETDATA_SPINLOCK_INITIALIZER,
731
.running = true,
732
.enabled = true,
733
},
734
.started_t = now_realtime_sec(),
739
- .next = NULL,
740
- .capabilities = 0,
735
};
736
737
// put the client IP and port into the buffers used by plugins.d
@@ -808,8 +802,6 @@ static int rrdpush_receive(struct receiver_state *rpt)
802
803
rrdpush_receive_log_status(rpt, "ready to receive data", "CONNECTED");
804
811
- cd.capabilities = rpt->capabilities;
812
-
805
#ifdef ENABLE_ACLK
806
// in case we have cloud connection we inform cloud
807
// new child connected
streaming/rrdpush.c
-1
@@ -1,7 +1,6 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
#include "rrdpush.h"
4
-#include "parser/parser.h"
4
5
/*
6
* rrdpush
streaming/sender.c
+1
-2
@@ -1,7 +1,6 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
#include "rrdpush.h"
4
-#include "parser/parser.h"
4
5
#define WORKER_SENDER_JOB_CONNECT 0
6
#define WORKER_SENDER_JOB_PIPE_READ 1
@@ -937,7 +936,7 @@ void execute_commands(struct sender_state *s) {
936
// internal_error(true, "STREAM %s [send to %s] received command over connection: %s", rrdhost_hostname(s->host), s->connected_to, start);
937
938
char *words[PLUGINSD_MAX_WORDS] = { NULL };
940
- size_t num_words = pluginsd_split_words(start, words, PLUGINSD_MAX_WORDS, NULL, NULL, 0);
939
+ size_t num_words = pluginsd_split_words(start, words, PLUGINSD_MAX_WORDS);
940
941
const char *keyword = get_word(words, num_words, 0);
942