4
5
#define LOG_FUNCTIONS false
6
7
-PARSER_RC pluginsd_set(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
7
+static int send_to_plugin(const char *txt, void *data) {
8
+ PARSER *parser = data;
9
+
10
+ if(!txt || !*txt)
11
+ return 0;
12
+
13
+#ifdef ENABLE_HTTPS
14
+ struct netdata_ssl *ssl = parser->ssl_output;
15
+ if(ssl) {
16
+ if(ssl->conn && ssl->flags == NETDATA_SSL_HANDSHAKE_COMPLETE) {
17
+ size_t size = strlen(txt);
18
+ return SSL_write(ssl->conn, txt, (int)size);
19
+ }
20
+
21
+ error("cannot write to SSL connection - connection is not ready.");
22
+ return -1;
23
+ }
24
+#endif
25
+
26
+ FILE *fp = parser->output;
27
+ int ret = fprintf(fp, "%s", txt);
28
+ fflush(fp);
29
+ return ret;
30
+}
31
+
32
+PARSER_RC pluginsd_set(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
33
{
9
- char *dimension = words[1];
10
- char *value = words[2];
34
+ char *dimension = get_word(words, num_words, 1);
35
+ char *value = get_word(words, num_words, 2);
36
37
RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
38
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
72
return PARSER_RC_ERROR;
73
}
74
50
-PARSER_RC pluginsd_begin(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
75
+PARSER_RC pluginsd_begin(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
76
{
52
- char *id = words[1];
53
- char *microseconds_txt = words[2];
77
+ char *id = get_word(words, num_words, 1);
78
+ char *microseconds_txt = get_word(words, num_words, 2);
79
80
RRDSET *st = NULL;
81
RRDHOST *host = ((PARSER_USER_OBJECT *)user)->host;
111
return PARSER_RC_ERROR;
112
}
113
89
-PARSER_RC pluginsd_end(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
114
+PARSER_RC pluginsd_end(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
115
{
116
UNUSED(words);
117
+ UNUSED(num_words);
118
+
119
RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
120
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
121
134
return PARSER_RC_OK;
135
}
136
110
-PARSER_RC pluginsd_chart(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
137
+PARSER_RC pluginsd_chart(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
138
{
139
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
140
if (unlikely(!host && !((PARSER_USER_OBJECT *) user)->host_exists)) {
142
return PARSER_RC_OK;
143
}
144
118
- char *type = words[1];
119
- char *name = words[2];
120
- char *title = words[3];
121
- char *units = words[4];
122
- char *family = words[5];
123
- char *context = words[6];
124
- char *chart = words[7];
125
- char *priority_s = words[8];
126
- char *update_every_s = words[9];
127
- char *options = words[10];
128
- char *plugin = words[11];
129
- char *module = words[12];
145
+ char *type = get_word(words, num_words, 1);
146
+ char *name = get_word(words, num_words, 2);
147
+ char *title = get_word(words, num_words, 3);
148
+ char *units = get_word(words, num_words, 4);
149
+ char *family = get_word(words, num_words, 5);
150
+ char *context = get_word(words, num_words, 6);
151
+ char *chart = get_word(words, num_words, 7);
152
+ char *priority_s = get_word(words, num_words, 8);
153
+ char *update_every_s = get_word(words, num_words, 9);
154
+ char *options = get_word(words, num_words, 10);
155
+ char *plugin = get_word(words, num_words, 11);
156
+ char *module = get_word(words, num_words, 12);
157
158
// parse the id from type
159
char *id = NULL;
258
return PARSER_RC_OK;
259
}
260
234
-PARSER_RC pluginsd_dimension(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
261
+PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action)
262
{
236
- char *id = words[1];
237
- char *name = words[2];
238
- char *algorithm = words[3];
239
- char *multiplier_s = words[4];
240
- char *divisor_s = words[5];
241
- char *options = words[6];
263
+ UNUSED(plugins_action);
264
+
265
+ long first_entry_child = str2l(get_word(words, num_words, 1));
266
+ long last_entry_child = str2l(get_word(words, num_words, 2));
267
+
268
+ PARSER_USER_OBJECT *user_object = (PARSER_USER_OBJECT *) user;
269
+
270
+ RRDHOST *host = user_object->host;
271
+ RRDSET *st = user_object->st;
272
+ if(unlikely(!host || !st)) {
273
+ error("REPLAY: received " PLUGINSD_KEYWORD_CHART_DEFINITION_END " command without a chart. Disabling it.");
274
+ return PARSER_RC_ERROR;
275
+ }
276
+
277
+ rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
278
+
279
+ bool ok = replicate_chart_request(send_to_plugin, user_object->parser, host, st, first_entry_child, last_entry_child, 0, 0);
280
+ return ok ? PARSER_RC_OK : PARSER_RC_ERROR;
281
+}
282
+
283
+PARSER_RC pluginsd_dimension(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
284
+{
285
+ char *id = get_word(words, num_words, 1);
286
+ char *name = get_word(words, num_words, 2);
287
+ char *algorithm = get_word(words, num_words, 3);
288
+ char *multiplier_s = get_word(words, num_words, 4);
289
+ char *divisor_s = get_word(words, num_words, 5);
290
+ char *options = get_word(words, num_words, 6);
291
292
RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
293
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
390
struct inflight_function *pf = func;
391
392
PARSER *parser = parser_ptr;
344
- FILE *fp = parser->output;
393
394
// leave this code as default, so that when the dictionary is destroyed this will be sent back to the caller
395
pf->code = HTTP_RESP_GATEWAY_TIMEOUT;
396
397
+ char buffer[2048 + 1];
398
+ snprintfz(buffer, 2048, "FUNCTION %s %d \"%s\"\n",
399
+ dictionary_acquired_item_name(item),
400
+ pf->timeout,
401
+ string2str(pf->function));
402
+
403
// send the command to the plugin
350
- int ret = fprintf(fp, "FUNCTION %s %d \"%s\"\n",
351
- dictionary_acquired_item_name(item),
352
- pf->timeout,
353
- string2str(pf->function));
404
+ int ret = send_to_plugin(buffer, parser);
405
406
pf->sent_ut = now_realtime_usec();
407
410
rrd_call_function_error(pf->destination_wb, "Failed to communicate with collector", HTTP_RESP_BACKEND_FETCH_FAILED);
411
}
412
else {
362
- fflush(fp);
363
-
413
internal_error(LOG_FUNCTIONS,
365
- "FUNCTION '%s' with transaction '%s' sent to collector (%d bytes, fd %d, in %llu usec)",
366
- string2str(pf->function), dictionary_acquired_item_name(item), ret, fileno(fp),
414
+ "FUNCTION '%s' with transaction '%s' sent to collector (%d bytes, in %llu usec)",
415
+ string2str(pf->function), dictionary_acquired_item_name(item), ret,
416
pf->sent_ut - pf->started_ut);
417
}
418
}
510
return HTTP_RESP_OK;
511
}
512
464
-PARSER_RC pluginsd_function(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
513
+PARSER_RC pluginsd_function(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
514
{
515
bool global = false;
467
- int i = 1;
468
- if(strcmp(words[i], "GLOBAL") == 0) {
516
+ size_t i = 1;
517
+ if(num_words >= 2 && strcmp(get_word(words, num_words, 1), "GLOBAL") == 0) {
518
i++;
519
global = true;
520
}
521
473
- char *name = words[i++];
474
- char *timeout_s = words[i++];
475
- char *help = words[i++];
522
+ char *name = get_word(words, num_words, i++);
523
+ char *timeout_s = get_word(words, num_words, i++);
524
+ char *help = get_word(words, num_words, i++);
525
526
RRDSET *st = (global)?NULL:((PARSER_USER_OBJECT *) user)->st;
527
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
557
string_freez(key);
558
}
559
511
-PARSER_RC pluginsd_function_result_begin(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
560
+PARSER_RC pluginsd_function_result_begin(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
561
{
513
- char *key = words[1];
514
- char *status = words[2];
515
- char *format = words[3];
516
- char *expires = words[4];
562
+ char *key = get_word(words, num_words, 1);
563
+ char *status = get_word(words, num_words, 2);
564
+ char *format = get_word(words, num_words, 3);
565
+ char *expires = get_word(words, num_words, 4);
566
567
if (unlikely(!key || !*key || !status || !*status || !format || !*format || !expires || !*expires)) {
568
error("got a " PLUGINSD_KEYWORD_FUNCTION_RESULT_BEGIN " without providing the required data (key = '%s', status = '%s', format = '%s', expires = '%s')."
613
614
// ----------------------------------------------------------------------------
615
567
-PARSER_RC pluginsd_variable(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
616
+PARSER_RC pluginsd_variable(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
617
{
569
- char *name = words[1];
570
- char *value = words[2];
618
+ char *name = get_word(words, num_words, 1);
619
+ char *value = get_word(words, num_words, 2);
620
NETDATA_DOUBLE v;
621
622
RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
627
if (name && *name) {
628
if ((strcmp(name, "GLOBAL") == 0 || strcmp(name, "HOST") == 0)) {
629
global = 1;
581
- name = words[2];
582
- value = words[3];
630
+ name = get_word(words, num_words, 2);
631
+ value = get_word(words, num_words, 3);
632
} else if ((strcmp(name, "LOCAL") == 0 || strcmp(name, "CHART") == 0)) {
633
global = 0;
585
- name = words[2];
586
- value = words[3];
634
+ name = get_word(words, num_words, 2);
635
+ value = get_word(words, num_words, 3);
636
}
637
}
638
690
return PARSER_RC_OK;
691
}
692
644
-PARSER_RC pluginsd_flush(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
693
+PARSER_RC pluginsd_flush(char **words __maybe_unused, size_t num_words __maybe_unused, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
694
{
646
- UNUSED(words);
695
debug(D_PLUGINSD, "requested a FLUSH");
696
((PARSER_USER_OBJECT *) user)->st = NULL;
697
+ ((PARSER_USER_OBJECT *) user)->replay.start_time = 0;
698
+ ((PARSER_USER_OBJECT *) user)->replay.end_time = 0;
699
+ ((PARSER_USER_OBJECT *) user)->replay.start_time_ut = 0;
700
+ ((PARSER_USER_OBJECT *) user)->replay.end_time_ut = 0;
701
return PARSER_RC_OK;
702
}
703
652
-PARSER_RC pluginsd_disable(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
704
+PARSER_RC pluginsd_disable(char **words __maybe_unused, size_t num_words __maybe_unused, void *user __maybe_unused, PLUGINSD_ACTION *plugins_action __maybe_unused)
705
{
654
- UNUSED(user);
655
- UNUSED(words);
656
-
706
info("called DISABLE. Disabling it.");
707
((PARSER_USER_OBJECT *) user)->enabled = 0;
708
return PARSER_RC_ERROR;
709
}
710
662
-PARSER_RC pluginsd_label(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
711
+PARSER_RC pluginsd_label(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
712
{
664
- char *store;
713
+ const char *name = get_word(words, num_words, 1);
714
+ const char *label_source = get_word(words, num_words, 2);
715
+ const char *value = get_word(words, num_words, 3);
716
666
- if (!words[1] || !words[2] || !words[3]) {
717
+ if (!name || !label_source || !value) {
718
error("Ignoring malformed or empty LABEL command.");
719
return PARSER_RC_OK;
720
}
670
- if (!words[4])
671
- store = words[3];
672
- else {
673
- store = callocz(PLUGINSD_LINE_MAX + 1, sizeof(char));
721
+
722
+ char *store = (char *)value;
723
+ bool allocated_store = false;
724
+
725
+ if(unlikely(num_words > 4)) {
726
+ allocated_store = true;
727
+ store = mallocz(PLUGINSD_LINE_MAX + 1);
728
size_t remaining = PLUGINSD_LINE_MAX;
729
char *move = store;
676
- int i = 3;
677
- while (i < PLUGINSD_MAX_WORDS) {
678
- size_t length = strlen(words[i]);
679
- if ((length + 1) >= remaining)
680
- break;
681
-
682
- remaining -= (length + 1);
683
- memcpy(move, words[i], length);
730
+ char *word;
731
+ for(size_t i = 3; i < num_words && remaining > 2 && (word = get_word(words, num_words, i)) ;i++) {
732
+ if(i > 3) {
733
+ *move++ = ' ';
734
+ *move = '\0';
735
+ remaining--;
736
+ }
737
+
738
+ size_t length = strlen(word);
739
+ if (length > remaining)
740
+ length = remaining;
741
+
742
+ remaining -= length;
743
+ memcpy(move, word, length);
744
move += length;
685
- *move++ = ' ';
686
-
687
- i++;
688
- if (!words[i])
689
- break;
745
+ *move = '\0';
746
}
747
}
748
749
if(unlikely(!((PARSER_USER_OBJECT *) user)->new_host_labels))
750
((PARSER_USER_OBJECT *) user)->new_host_labels = rrdlabels_create();
751
696
- rrdlabels_add(((PARSER_USER_OBJECT *)user)->new_host_labels, words[1], store, strtol(words[2], NULL, 10));
752
+ rrdlabels_add(((PARSER_USER_OBJECT *)user)->new_host_labels,
753
+ name,
754
+ store,
755
+ str2l(label_source));
756
698
- if (store != words[3])
757
+ if (allocated_store)
758
freez(store);
759
+
760
return PARSER_RC_OK;
761
}
762
703
-PARSER_RC pluginsd_overwrite(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
763
+PARSER_RC pluginsd_overwrite(char **words __maybe_unused, size_t num_words __maybe_unused, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
764
{
705
- UNUSED(words);
706
-
765
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
766
debug(D_PLUGINSD, "requested to OVERWRITE host labels");
767
777
}
778
779
722
-PARSER_RC pluginsd_clabel(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
780
+PARSER_RC pluginsd_clabel(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
781
{
724
- if (!words[1] || !words[2] || !words[3]) {
782
+ const char *name = get_word(words, num_words, 1);
783
+ const char *value = get_word(words, num_words, 2);
784
+ const char *label_source = get_word(words, num_words, 3);
785
+
786
+ if (!name || !value || !*label_source) {
787
error("Ignoring malformed or empty CHART LABEL command.");
788
return PARSER_RC_OK;
789
}
793
rrdlabels_unmark_all(((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily);
794
}
795
734
- rrdlabels_add(((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily, words[1], words[2], strtol(words[3], NULL, 10));
796
+ rrdlabels_add(((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily,
797
+ name, value, str2l(label_source));
798
799
return PARSER_RC_OK;
800
}
801
739
-PARSER_RC pluginsd_clabel_commit(char **words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
802
+PARSER_RC pluginsd_clabel_commit(char **words __maybe_unused, size_t num_words __maybe_unused, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
803
{
741
- UNUSED(words);
742
-
804
RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
805
RRDSET *st = ((PARSER_USER_OBJECT *)user)->st;
806
823
return PARSER_RC_OK;
824
}
825
765
-PARSER_RC pluginsd_guid(char **words, void *user, PLUGINSD_ACTION *plugins_action)
826
+PARSER_RC pluginsd_guid(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action)
827
{
767
- char *uuid_str = words[1];
828
+ char *uuid_str = get_word(words, num_words, 1);
829
uuid_t uuid;
830
831
if (unlikely(!uuid_str)) {
845
return PARSER_RC_OK;
846
}
847
787
-PARSER_RC pluginsd_context(char **words, void *user, PLUGINSD_ACTION *plugins_action)
848
+PARSER_RC pluginsd_context(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action)
849
{
789
- char *uuid_str = words[1];
850
+ char *uuid_str = get_word(words, num_words, 1);
851
uuid_t uuid;
852
853
if (unlikely(!uuid_str)) {
867
return PARSER_RC_OK;
868
}
869
809
-PARSER_RC pluginsd_tombstone(char **words, void *user, PLUGINSD_ACTION *plugins_action)
870
+PARSER_RC pluginsd_tombstone(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action)
871
{
811
- char *uuid_str = words[1];
872
+ char *uuid_str = get_word(words, num_words, 1);
873
uuid_t uuid;
874
875
if (unlikely(!uuid_str)) {
889
return PARSER_RC_OK;
890
}
891
831
-PARSER_RC metalog_pluginsd_host(char **words, void *user, PLUGINSD_ACTION *plugins_action)
892
+PARSER_RC metalog_pluginsd_host(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action)
893
{
833
- char *machine_guid = words[1];
834
- char *hostname = words[2];
835
- char *registry_hostname = words[3];
836
- char *update_every_s = words[4];
837
- char *os = words[5];
838
- char *timezone = words[6];
839
- char *tags = words[7];
894
+ char *machine_guid = get_word(words, num_words, 1);
895
+ char *hostname = get_word(words, num_words, 2);
896
+ char *registry_hostname = get_word(words, num_words, 3);
897
+ char *update_every_s = get_word(words, num_words, 4);
898
+ char *os = get_word(words, num_words, 5);
899
+ char *timezone = get_word(words, num_words, 6);
900
+ char *tags = get_word(words, num_words, 7);
901
902
int update_every = 1;
903
if (likely(update_every_s && *update_every_s))
916
return PARSER_RC_OK;
917
}
918
919
+PARSER_RC pluginsd_replay_rrdset_begin(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
920
+{
921
+ char *id = get_word(words, num_words, 1);
922
+ char *start_time_str = get_word(words, num_words, 2);
923
+ char *end_time_str = get_word(words, num_words, 3);
924
+
925
+ RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
926
+ RRDHOST *host = ((PARSER_USER_OBJECT *)user)->host;
927
+
928
+ if (unlikely(!id || (!st && !*id))) {
929
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_BEGIN " without a chart id for host '%s'. Disabling it.", rrdhost_hostname(host));
930
+ goto disable;
931
+ }
932
+
933
+ if(*id) {
934
+ st = rrdset_find(host, id);
935
+ if (unlikely(!st)) {
936
+ error("requested a " PLUGINSD_KEYWORD_REPLAY_BEGIN " on chart '%s', which does not exist on host '%s'. Disabling it.",
937
+ id, rrdhost_hostname(host));
938
+ goto disable;
939
+ }
940
+
941
+ ((PARSER_USER_OBJECT *) user)->st = st;
942
+ ((PARSER_USER_OBJECT *) user)->replay.start_time = 0;
943
+ ((PARSER_USER_OBJECT *) user)->replay.end_time = 0;
944
+ ((PARSER_USER_OBJECT *) user)->replay.start_time_ut = 0;
945
+ ((PARSER_USER_OBJECT *) user)->replay.end_time_ut = 0;
946
+ }
947
+
948
+ if(start_time_str && end_time_str) {
949
+ time_t start_time = strtol(start_time_str, NULL, 0);
950
+ time_t end_time = strtol(end_time_str, NULL, 0);
951
+
952
+ if(start_time && end_time) {
953
+ if (start_time > end_time) {
954
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_BEGIN " on chart '%s' ('%s') on host '%s', but timings are invalid (%ld to %ld). Disabling it.",
955
+ rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost), start_time, end_time);
956
+ goto disable;
957
+ }
958
+
959
+ if (end_time - start_time != st->update_every)
960
+ rrdset_set_update_every(st, end_time - start_time);
961
+
962
+ st->last_collected_time.tv_sec = end_time;
963
+ st->last_collected_time.tv_usec = 0;
964
+
965
+ st->last_updated.tv_sec = end_time;
966
+ st->last_updated.tv_usec = 0;
967
+
968
+ ((PARSER_USER_OBJECT *) user)->replay.start_time = start_time;
969
+ ((PARSER_USER_OBJECT *) user)->replay.end_time = end_time;
970
+ ((PARSER_USER_OBJECT *) user)->replay.start_time_ut = (usec_t) start_time * USEC_PER_SEC;
971
+ ((PARSER_USER_OBJECT *) user)->replay.end_time_ut = (usec_t) end_time * USEC_PER_SEC;
972
+
973
+ st->counter++;
974
+ st->counter_done++;
975
+
976
+ // these are only needed for db mode RAM, SAVE, MAP, ALLOC
977
+ st->current_entry++;
978
+ if(st->current_entry >= st->entries)
979
+ st->current_entry -= st->entries;
980
+ }
981
+ }
982
+
983
+ return PARSER_RC_OK;
984
+
985
+disable:
986
+ ((PARSER_USER_OBJECT *)user)->enabled = 0;
987
+ return PARSER_RC_ERROR;
988
+}
989
+
990
+PARSER_RC pluginsd_replay_set(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
991
+{
992
+ char *dimension = get_word(words, num_words, 1);
993
+ char *value_str = get_word(words, num_words, 2);
994
+ char *flags_str = get_word(words, num_words, 3);
995
+
996
+ RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
997
+ RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
998
+
999
+ if (unlikely(!st)) {
1000
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_SET " on dimension '%s' on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1001
+ dimension, rrdhost_hostname(host));
1002
+ goto disable;
1003
+ }
1004
+
1005
+ if (unlikely(!dimension || !*dimension)) {
1006
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_SET " on chart '%s' of host '%s', without a dimension. Disabling it.",
1007
+ rrdset_id(st), rrdhost_hostname(host));
1008
+ goto disable;
1009
+ }
1010
+
1011
+ if (unlikely(!((PARSER_USER_OBJECT *) user)->replay.start_time || !((PARSER_USER_OBJECT *) user)->replay.end_time)) {
1012
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_SET " on dimension '%s' on host '%s', without timings from a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1013
+ dimension, rrdhost_hostname(host));
1014
+ goto disable;
1015
+ }
1016
+
1017
+ if (unlikely(!value_str || !*value_str))
1018
+ value_str = "nan";
1019
+
1020
+ if(unlikely(!flags_str))
1021
+ flags_str = "";
1022
+
1023
+ if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
1024
+ debug(D_PLUGINSD, "REPLAY: is replaying dimension '%s'/'%s' to '%s'", rrdset_id(st), dimension, value_str);
1025
+
1026
+ if (likely(value_str)) {
1027
+ RRDDIM *rd = rrddim_find(st, dimension);
1028
+ if(unlikely(!rd)) {
1029
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_SET " to dimension with id '%s' on chart '%s' ('%s') on host '%s', which does not exist. Disabling it.",
1030
+ dimension, rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost));
1031
+ goto disable;
1032
+ }
1033
+ else {
1034
+ NETDATA_DOUBLE value = strtondd(value_str, NULL);
1035
+ SN_FLAGS flags = SN_FLAG_NONE;
1036
+
1037
+ char c;
1038
+ while((c = *flags_str++)) {
1039
+ switch(c) {
1040
+ case 'R':
1041
+ flags |= SN_FLAG_RESET;
1042
+ break;
1043
+
1044
+ case 'E':
1045
+ flags |= SN_EMPTY_SLOT;
1046
+ value = NAN;
1047
+ break;
1048
+
1049
+ default:
1050
+ error("unknown flag '%c'", c);
1051
+ break;
1052
+ }
1053
+ }
1054
+
1055
+ if(!netdata_double_isnumber(value)) {
1056
+ value = NAN;
1057
+ flags = SN_EMPTY_SLOT;
1058
+ }
1059
+
1060
+ rrddim_store_metric(rd, ((PARSER_USER_OBJECT *) user)->replay.end_time_ut, value, flags);
1061
+ rd->last_collected_time.tv_sec = ((PARSER_USER_OBJECT *) user)->replay.end_time;
1062
+ rd->last_collected_time.tv_usec = 0;
1063
+ rd->collections_counter++;
1064
+ }
1065
+ }
1066
+ return PARSER_RC_OK;
1067
+
1068
+disable:
1069
+ ((PARSER_USER_OBJECT *) user)->enabled = 0;
1070
+ return PARSER_RC_ERROR;
1071
+}
1072
+
1073
+PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
1074
+{
1075
+ char *dimension = get_word(words, num_words, 1);
1076
+ char *last_collected_ut_str = get_word(words, num_words, 2);
1077
+ char *last_collected_value_str = get_word(words, num_words, 3);
1078
+ char *last_calculated_value_str = get_word(words, num_words, 4);
1079
+ char *last_stored_value_str = get_word(words, num_words, 5);
1080
+
1081
+ RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
1082
+ RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1083
+
1084
+ if (unlikely(!st)) {
1085
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " on dimension '%s' on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1086
+ dimension, rrdhost_hostname(host));
1087
+ goto disable;
1088
+ }
1089
+
1090
+ if (unlikely(!dimension || !*dimension)) {
1091
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " on chart '%s' of host '%s', without a dimension. Disabling it.",
1092
+ rrdset_id(st), rrdhost_hostname(host));
1093
+ goto disable;
1094
+ }
1095
+
1096
+ RRDDIM *rd = rrddim_find(st, dimension);
1097
+ if(unlikely(!rd)) {
1098
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " to dimension with id '%s' on chart '%s' ('%s') on host '%s', which does not exist. Disabling it.",
1099
+ dimension, rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost));
1100
+ goto disable;
1101
+ }
1102
+
1103
+ usec_t dim_last_collected_ut = (usec_t)rd->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)rd->last_collected_time.tv_usec;
1104
+ usec_t last_collected_ut = last_collected_ut_str ? str2ull(last_collected_ut_str) : 0;
1105
+ if(last_collected_ut > dim_last_collected_ut) {
1106
+ rd->last_collected_time.tv_sec = last_collected_ut / USEC_PER_SEC;
1107
+ rd->last_collected_time.tv_usec = last_collected_ut % USEC_PER_SEC;
1108
+ }
1109
+
1110
+ rd->last_collected_value = last_collected_value_str ? str2ll(last_collected_value_str, NULL) : 0;
1111
+ rd->last_calculated_value = last_calculated_value_str ? str2ndd(last_calculated_value_str, NULL) : 0;
1112
+ rd->last_stored_value = last_stored_value_str ? str2ndd(last_stored_value_str, NULL) : 0.0;
1113
+ return PARSER_RC_OK;
1114
+
1115
+disable:
1116
+ ((PARSER_USER_OBJECT *) user)->enabled = 0;
1117
+ return PARSER_RC_ERROR;
1118
+}
1119
+
1120
+PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
1121
+{
1122
+ char *last_collected_ut_str = get_word(words, num_words, 1);
1123
+ char *last_updated_ut_str = get_word(words, num_words, 2);
1124
+ char *last_collected_total_str = get_word(words, num_words, 3);
1125
+ char *collected_total_str = get_word(words, num_words, 4);
1126
+
1127
+ RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
1128
+ RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1129
+
1130
+ if (unlikely(!st)) {
1131
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE " on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1132
+ rrdhost_hostname(host));
1133
+ goto disable;
1134
+ }
1135
+
1136
+ usec_t chart_last_collected_ut = (usec_t)st->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)st->last_collected_time.tv_usec;
1137
+ usec_t last_collected_ut = last_collected_ut_str ? str2ull(last_collected_ut_str) : 0;
1138
+ if(last_collected_ut > chart_last_collected_ut) {
1139
+ st->last_collected_time.tv_sec = last_collected_ut / USEC_PER_SEC;
1140
+ st->last_collected_time.tv_usec = last_collected_ut % USEC_PER_SEC;
1141
+ }
1142
+
1143
+ usec_t chart_last_updated_ut = (usec_t)st->last_updated.tv_sec * USEC_PER_SEC + (usec_t)st->last_updated.tv_usec;
1144
+ usec_t last_updated_ut = last_updated_ut_str ? str2ull(last_updated_ut_str) : 0;
1145
+ if(last_updated_ut > chart_last_updated_ut) {
1146
+ st->last_updated.tv_sec = last_updated_ut / USEC_PER_SEC;
1147
+ st->last_updated.tv_usec = last_updated_ut % USEC_PER_SEC;
1148
+ }
1149
+
1150
+ st->last_collected_total = last_collected_total_str ? strtoll(last_collected_total_str, NULL, 0) : 0;
1151
+ st->collected_total = collected_total_str ? strtoll(collected_total_str, NULL, 0) : 0;
1152
+
1153
+ st->counter++;
1154
+ st->counter_done++;
1155
+
1156
+ return PARSER_RC_OK;
1157
+
1158
+disable:
1159
+ ((PARSER_USER_OBJECT *) user)->enabled = 0;
1160
+ return PARSER_RC_ERROR;
1161
+}
1162
+
1163
+PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user, PLUGINSD_ACTION *plugins_action __maybe_unused)
1164
+{
1165
+ if (num_words < 7) {
1166
+ error("REPLAY: malformed " PLUGINSD_KEYWORD_REPLAY_END " command");
1167
+ return PARSER_RC_ERROR;
1168
+ }
1169
+
1170
+ time_t update_every_child = str2l(get_word(words, num_words, 1));
1171
+ time_t first_entry_child = str2l(get_word(words, num_words, 2));
1172
+ time_t last_entry_child = str2l(get_word(words, num_words, 3));
1173
+
1174
+ bool start_streaming = (strcmp(get_word(words, num_words, 4), "true") == 0);
1175
+ time_t first_entry_requested = str2l(get_word(words, num_words, 5));
1176
+ time_t last_entry_requested = str2l(get_word(words, num_words, 6));
1177
+
1178
+ PARSER_USER_OBJECT *user_object = user;
1179
+
1180
+ RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
1181
+ RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1182
+
1183
+ if (unlikely(!st)) {
1184
+ error("REPLAY: requested a " PLUGINSD_KEYWORD_REPLAY_END " on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1185
+ rrdhost_hostname(host));
1186
+ return PARSER_RC_ERROR;
1187
+ }
1188
+
1189
+ ((PARSER_USER_OBJECT *) user)->st = NULL;
1190
+ ((PARSER_USER_OBJECT *) user)->count++;
1191
+
1192
+ st->counter++;
1193
+ st->counter_done++;
1194
+
1195
+ if (start_streaming) {
1196
+ if (st->update_every != update_every_child)
1197
+ rrdset_set_update_every(st, update_every_child);
1198
+
1199
+ rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
1200
+ rrdset_flag_clear(st, RRDSET_FLAG_SYNC_CLOCK);
1201
+ return PARSER_RC_OK;
1202
+ }
1203
+
1204
+ bool ok = replicate_chart_request(send_to_plugin, user_object->parser, host, st, first_entry_child, last_entry_child,
1205
+ first_entry_requested, last_entry_requested);
1206
+ return ok ? PARSER_RC_OK : PARSER_RC_ERROR;
1207
+}
1208
+
1209
static void pluginsd_process_thread_cleanup(void *ptr) {
1210
PARSER *parser = (PARSER *)ptr;
1211
rrd_collector_finished();
1246
};
1247
1248
// fp_plugin_output = our input; fp_plugin_input = our output
898
- PARSER *parser = parser_init(host, &user, fp_plugin_output, fp_plugin_input, PARSER_INPUT_SPLIT);
1249
+ PARSER *parser = parser_init(host, &user, fp_plugin_output, fp_plugin_input, PARSER_INPUT_SPLIT, NULL);
1250
1251
rrd_collector_started();
1252