pluginsd_parser.h 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217
  1. // SPDX-License-Identifier: GPL-3.0-or-later
  2. #ifndef NETDATA_PLUGINSD_PARSER_H
  3. #define NETDATA_PLUGINSD_PARSER_H
  4. #include "daemon/common.h"
  5. #define WORKER_PARSER_FIRST_JOB 3
  6. // this has to be in-sync with the same at receiver.c
  7. #define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 3)
  8. // PARSER return codes
  9. typedef enum __attribute__ ((__packed__)) parser_rc {
  10. PARSER_RC_OK, // Callback was successful, go on
  11. PARSER_RC_STOP, // Callback says STOP
  12. PARSER_RC_ERROR // Callback failed (abort rest of callbacks)
  13. } PARSER_RC;
  14. typedef enum __attribute__ ((__packed__)) parser_input_type {
  15. PARSER_INPUT_SPLIT = (1 << 1),
  16. PARSER_DEFER_UNTIL_KEYWORD = (1 << 2),
  17. } PARSER_INPUT_TYPE;
  18. typedef enum __attribute__ ((__packed__)) {
  19. PARSER_INIT_PLUGINSD = (1 << 1),
  20. PARSER_INIT_STREAMING = (1 << 2),
  21. } PARSER_REPERTOIRE;
  22. struct parser;
  23. typedef PARSER_RC (*keyword_function)(char **words, size_t num_words, struct parser *parser);
  24. typedef struct parser_keyword {
  25. char *keyword;
  26. keyword_function func;
  27. PARSER_REPERTOIRE repertoire;
  28. size_t worker_job_id;
  29. } PARSER_KEYWORD;
  30. typedef struct parser_user_object {
  31. RRDSET *st;
  32. RRDHOST *host;
  33. void *opaque;
  34. struct plugind *cd;
  35. int trust_durations;
  36. DICTIONARY *new_host_labels;
  37. DICTIONARY *chart_rrdlabels_linked_temporarily;
  38. size_t data_collections_count;
  39. int enabled;
  40. STREAM_CAPABILITIES capabilities; // receiver capabilities
  41. struct {
  42. bool parsing_host;
  43. uuid_t machine_guid;
  44. char machine_guid_str[UUID_STR_LEN];
  45. STRING *hostname;
  46. DICTIONARY *rrdlabels;
  47. } host_define;
  48. struct parser_user_object_replay {
  49. time_t start_time;
  50. time_t end_time;
  51. usec_t start_time_ut;
  52. usec_t end_time_ut;
  53. time_t wall_clock_time;
  54. bool rset_enabled;
  55. } replay;
  56. struct parser_user_object_v2 {
  57. bool locked_data_collection;
  58. RRDSET_STREAM_BUFFER stream_buffer; // sender capabilities in this
  59. time_t update_every;
  60. time_t end_time;
  61. time_t wall_clock_time;
  62. bool ml_locked;
  63. } v2;
  64. } PARSER_USER_OBJECT;
  65. typedef struct parser {
  66. uint8_t version; // Parser version
  67. PARSER_REPERTOIRE repertoire;
  68. uint32_t flags;
  69. int fd; // Socket
  70. size_t line;
  71. FILE *fp_input; // Input source e.g. stream
  72. FILE *fp_output; // Stream to send commands to plugin
  73. #ifdef ENABLE_HTTPS
  74. NETDATA_SSL *ssl_output;
  75. #endif
  76. PARSER_USER_OBJECT user; // User defined structure to hold extra state between calls
  77. struct {
  78. const char *end_keyword;
  79. BUFFER *response;
  80. void (*action)(struct parser *parser, void *action_data);
  81. void *action_data;
  82. } defer;
  83. struct {
  84. DICTIONARY *functions;
  85. usec_t smaller_timeout;
  86. } inflight;
  87. struct {
  88. SPINLOCK spinlock;
  89. } writer;
  90. } PARSER;
  91. static inline int find_first_keyword(const char *src, char *dst, int dst_size, bool *isspace_map) {
  92. const char *s = src, *keyword_start;
  93. while (unlikely(isspace_map[(uint8_t)*s])) s++;
  94. keyword_start = s;
  95. while (likely(*s && !isspace_map[(uint8_t)*s]) && dst_size > 1) {
  96. *dst++ = *s++;
  97. dst_size--;
  98. }
  99. *dst = '\0';
  100. return dst_size == 0 ? 0 : (int) (s - keyword_start);
  101. }
  102. PARSER_KEYWORD *gperf_lookup_keyword(register const char *str, register size_t len);
  103. static inline PARSER_KEYWORD *parser_find_keyword(PARSER *parser, const char *command) {
  104. PARSER_KEYWORD *t = gperf_lookup_keyword(command, strlen(command));
  105. if(t && (t->repertoire & parser->repertoire))
  106. return t;
  107. return NULL;
  108. }
  109. static inline int parser_action(PARSER *parser, char *input) {
  110. parser->line++;
  111. if(unlikely(parser->flags & PARSER_DEFER_UNTIL_KEYWORD)) {
  112. char command[PLUGINSD_LINE_MAX + 1];
  113. bool has_keyword = find_first_keyword(input, command, PLUGINSD_LINE_MAX, isspace_map_pluginsd);
  114. if(!has_keyword || strcmp(command, parser->defer.end_keyword) != 0) {
  115. if(parser->defer.response) {
  116. buffer_strcat(parser->defer.response, input);
  117. if(buffer_strlen(parser->defer.response) > 10 * 1024 * 1024) {
  118. // more than 10MB of data
  119. // a bad plugin that did not send the end_keyword
  120. internal_error(true, "PLUGINSD: deferred response is too big (%zu bytes). Stopping this plugin.", buffer_strlen(parser->defer.response));
  121. return 1;
  122. }
  123. }
  124. return 0;
  125. }
  126. else {
  127. // call the action
  128. parser->defer.action(parser, parser->defer.action_data);
  129. // empty everything
  130. parser->defer.action = NULL;
  131. parser->defer.action_data = NULL;
  132. parser->defer.end_keyword = NULL;
  133. parser->defer.response = NULL;
  134. parser->flags &= ~PARSER_DEFER_UNTIL_KEYWORD;
  135. }
  136. return 0;
  137. }
  138. char *words[PLUGINSD_MAX_WORDS];
  139. size_t num_words = quoted_strings_splitter_pluginsd(input, words, PLUGINSD_MAX_WORDS);
  140. const char *command = get_word(words, num_words, 0);
  141. if(unlikely(!command))
  142. return 0;
  143. PARSER_RC rc;
  144. PARSER_KEYWORD *t = parser_find_keyword(parser, command);
  145. if(likely(t)) {
  146. worker_is_busy(t->worker_job_id);
  147. rc = (*t->func)(words, num_words, parser);
  148. worker_is_idle();
  149. }
  150. else
  151. rc = PARSER_RC_ERROR;
  152. if(rc == PARSER_RC_ERROR) {
  153. BUFFER *wb = buffer_create(PLUGINSD_LINE_MAX, NULL);
  154. for(size_t i = 0; i < num_words ;i++) {
  155. if(i) buffer_fast_strcat(wb, " ", 1);
  156. buffer_fast_strcat(wb, "\"", 1);
  157. const char *s = get_word(words, num_words, i);
  158. buffer_strcat(wb, s?s:"");
  159. buffer_fast_strcat(wb, "\"", 1);
  160. }
  161. error("PLUGINSD: parser_action('%s') failed on line %zu: { %s } (quotes added to show parsing)",
  162. command, parser->line, buffer_tostring(wb));
  163. buffer_free(wb);
  164. }
  165. return (rc == PARSER_RC_ERROR || rc == PARSER_RC_STOP);
  166. }
  167. PARSER *parser_init(struct parser_user_object *user, FILE *fp_input, FILE *fp_output, int fd, PARSER_INPUT_TYPE flags, void *ssl);
  168. void parser_init_repertoire(PARSER *parser, PARSER_REPERTOIRE repertoire);
  169. void parser_destroy(PARSER *working_parser);
  170. void pluginsd_cleanup_v2(PARSER *parser);
  171. void inflight_functions_init(PARSER *parser);
  172. void pluginsd_keywords_init(PARSER *parser, PARSER_REPERTOIRE repertoire);
  173. #endif //NETDATA_PLUGINSD_PARSER_H