| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "database/rrd.h" |
| 4 | #include "stream-receiver-internals.h" |
| 5 | #include "stream-sender-internals.h" |
| 6 | #include "stream-replication-sender.h" |
| 7 | |
| 8 | static struct config stream_config = APPCONFIG_INITIALIZER; |
| 9 | |
| 10 | /** |
| 11 | * Free stream configuration |
| 12 | * |
| 13 | * Free all memory associated with the stream configuration. |
| 14 | * Called during shutdown to prevent memory leaks. |
| 15 | */ |
| 16 | void stream_config_free(void) { |
| 17 | // Free the configuration |
| 18 | inicfg_free(&stream_config); |
| 19 | |
| 20 | // Free the allocated strings in stream_send structure |
| 21 | string_freez(stream_send.api_key); |
| 22 | string_freez(stream_send.send_charts_matching); |
| 23 | string_freez(stream_send.parents.destination); |
| 24 | string_freez(stream_send.parents.ssl_ca_path); |
| 25 | string_freez(stream_send.parents.ssl_ca_file); |
| 26 | |
| 27 | // Reset the pointers to NULL |
| 28 | stream_send.api_key = NULL; |
| 29 | stream_send.send_charts_matching = NULL; |
| 30 | stream_send.parents.destination = NULL; |
| 31 | stream_send.parents.ssl_ca_path = NULL; |
| 32 | stream_send.parents.ssl_ca_file = NULL; |
| 33 | } |
| 34 | |
| 35 | struct _stream_send stream_send = { |
| 36 | .enabled = false, |
| 37 | .api_key = NULL, |
| 38 | .send_charts_matching = NULL, |
| 39 | .initial_clock_resync_iterations = 60, |
| 40 | |
| 41 | .buffer_max_size = CBUFFER_INITIAL_MAX_SIZE, |
| 42 | |
| 43 | .replication = { |
| 44 | .prefetch = 0, |
| 45 | .threads = 0, |
| 46 | }, |
| 47 | |
| 48 | .parents = { |
| 49 | .destination = NULL, |
| 50 | .default_port = 19999, |
| 51 | .h2o = false, |
| 52 | .timeout_s = 300, |
| 53 | .reconnect_delay_s = 15, |
| 54 | .ssl_ca_path = NULL, |
| 55 | .ssl_ca_file = NULL, |
| 56 | }, |
| 57 | |
| 58 | .compression = { |
| 59 | .enabled = true, |
| 60 | .levels = { |
| 61 | [COMPRESSION_ALGORITHM_NONE] = 0, |
| 62 | [COMPRESSION_ALGORITHM_ZSTD] = 3, // 1 (faster) - 22 (smaller) |
| 63 | [COMPRESSION_ALGORITHM_LZ4] = 1, // 1 (smaller) - 9 (faster) |
| 64 | [COMPRESSION_ALGORITHM_BROTLI] = 3, // 0 (faster) - 11 (smaller) |
| 65 | [COMPRESSION_ALGORITHM_GZIP] = 3, // 1 (faster) - 9 (smaller) |
| 66 | } |
| 67 | }, |
| 68 | }; |
| 69 | |
| 70 | struct _stream_receive stream_receive = { |
| 71 | .replication = { |
| 72 | .enabled = true, |
| 73 | .period = 86400, |
| 74 | .step = 3600, |
| 75 | } |
| 76 | }; |
| 77 | |
| 78 | void stream_conf_set_sender_compression_levels(ND_COMPRESSION_PROFILE profile) { |
| 79 | switch(profile) { |
| 80 | default: |
| 81 | case ND_COMPRESSION_DEFAULT: |
| 82 | stream_send.compression.levels[COMPRESSION_ALGORITHM_ZSTD] = 3; |
| 83 | stream_send.compression.levels[COMPRESSION_ALGORITHM_LZ4] = 1; |
| 84 | stream_send.compression.levels[COMPRESSION_ALGORITHM_BROTLI] = 3; |
| 85 | stream_send.compression.levels[COMPRESSION_ALGORITHM_GZIP] = 3; |
| 86 | break; |
| 87 | |
| 88 | case ND_COMPRESSION_FASTEST: |
| 89 | stream_send.compression.levels[COMPRESSION_ALGORITHM_ZSTD] = 1; |
| 90 | stream_send.compression.levels[COMPRESSION_ALGORITHM_LZ4] = 9; |
| 91 | stream_send.compression.levels[COMPRESSION_ALGORITHM_BROTLI] = 1; |
| 92 | stream_send.compression.levels[COMPRESSION_ALGORITHM_GZIP] = 1; |
| 93 | break; |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | static void stream_conf_load_internal() { |
| 98 | errno_clear(); |
| 99 | char *filename = filename_from_path_entry_strdupz(netdata_configured_user_config_dir, "stream.conf"); |
| 100 | if(!inicfg_load(&stream_config, filename, 0, NULL)) { |
| 101 | nd_log_daemon(NDLP_NOTICE, "CONFIG: cannot load user config '%s'. Will try stock config.", filename); |
| 102 | freez(filename); |
| 103 | |
| 104 | filename = filename_from_path_entry_strdupz(netdata_configured_stock_config_dir, "stream.conf"); |
| 105 | if(!inicfg_load(&stream_config, filename, 0, NULL)) |
| 106 | nd_log_daemon(NDLP_NOTICE, "CONFIG: cannot load stock config '%s'. Running with internal defaults.", filename); |
| 107 | } |
| 108 | |
| 109 | freez(filename); |
| 110 | |
| 111 | inicfg_move(&stream_config, |
| 112 | CONFIG_SECTION_STREAM, "timeout seconds", |
| 113 | CONFIG_SECTION_STREAM, "timeout"); |
| 114 | |
| 115 | inicfg_move(&stream_config, |
| 116 | CONFIG_SECTION_STREAM, "reconnect delay seconds", |
| 117 | CONFIG_SECTION_STREAM, "reconnect delay"); |
| 118 | |
| 119 | inicfg_move_everywhere(&stream_config, "default memory mode", "db"); |
| 120 | inicfg_move_everywhere(&stream_config, "memory mode", "db"); |
| 121 | inicfg_move_everywhere(&stream_config, "db mode", "db"); |
| 122 | inicfg_move_everywhere(&stream_config, "default history", "retention"); |
| 123 | inicfg_move_everywhere(&stream_config, "history", "retention"); |
| 124 | inicfg_move_everywhere(&stream_config, "default proxy enabled", "proxy enabled"); |
| 125 | inicfg_move_everywhere(&stream_config, "default proxy destination", "proxy destination"); |
| 126 | inicfg_move_everywhere(&stream_config, "default proxy api key", "proxy api key"); |
| 127 | inicfg_move_everywhere(&stream_config, "default proxy send charts matching", "proxy send charts matching"); |
| 128 | inicfg_move_everywhere(&stream_config, "default health log history", "health log retention"); |
| 129 | inicfg_move_everywhere(&stream_config, "health log history", "health log retention"); |
| 130 | inicfg_move_everywhere(&stream_config, "seconds to replicate", "replication period"); |
| 131 | inicfg_move_everywhere(&stream_config, "seconds per replication step", "replication step"); |
| 132 | inicfg_move_everywhere(&stream_config, "default postpone alarms on connect seconds", "postpone alerts on connect"); |
| 133 | inicfg_move_everywhere(&stream_config, "postpone alarms on connect seconds", "postpone alerts on connect"); |
| 134 | inicfg_move_everywhere(&stream_config, "health enabled by default", "health enabled"); |
| 135 | inicfg_move_everywhere(&stream_config, "buffer size bytes", "buffer size"); |
| 136 | } |
| 137 | |
| 138 | bool stream_conf_receiver_needs_dbengine(void) { |
| 139 | return stream_conf_needs_dbengine(&stream_config); |
| 140 | } |
| 141 | |
| 142 | void stream_conf_load() { |
| 143 | FUNCTION_RUN_ONCE(); |
| 144 | |
| 145 | stream_conf_load_internal(); |
| 146 | check_local_streaming_capabilities(); |
| 147 | |
| 148 | stream_send.enabled = |
| 149 | inicfg_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "enabled", stream_send.enabled); |
| 150 | |
| 151 | stream_send.parents.destination = |
| 152 | string_strdupz(inicfg_get(&stream_config, CONFIG_SECTION_STREAM, "destination", "")); |
| 153 | |
| 154 | stream_send.api_key = |
| 155 | string_strdupz(inicfg_get(&stream_config, CONFIG_SECTION_STREAM, "api key", "")); |
| 156 | |
| 157 | stream_send.send_charts_matching = |
| 158 | string_strdupz(inicfg_get(&stream_config, CONFIG_SECTION_STREAM, "send charts matching", "*")); |
| 159 | |
| 160 | stream_receive.replication.enabled = |
| 161 | inicfg_get_boolean(&netdata_config, CONFIG_SECTION_DB, "enable replication", |
| 162 | stream_receive.replication.enabled); |
| 163 | |
| 164 | stream_receive.replication.period = |
| 165 | inicfg_get_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "replication period", |
| 166 | stream_receive.replication.period); |
| 167 | |
| 168 | stream_receive.replication.step = |
| 169 | inicfg_get_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "replication step", |
| 170 | stream_receive.replication.step); |
| 171 | |
| 172 | stream_send.replication.threads = inicfg_get_number_range( |
| 173 | &netdata_config, CONFIG_SECTION_DB, "replication threads", |
| 174 | replication_threads_default(), 1, MAX_REPLICATION_THREADS); |
| 175 | |
| 176 | stream_send.replication.prefetch = inicfg_get_number_range( |
| 177 | &netdata_config, CONFIG_SECTION_DB, "replication prefetch", |
| 178 | replication_prefetch_default(), 1, MAX_REPLICATION_PREFETCH); |
| 179 | |
| 180 | stream_send.buffer_max_size = (size_t)inicfg_get_size_bytes( |
| 181 | &stream_config, CONFIG_SECTION_STREAM, "buffer size", |
| 182 | stream_send.buffer_max_size); |
| 183 | |
| 184 | stream_send.parents.reconnect_delay_s = (unsigned int)inicfg_get_duration_seconds( |
| 185 | &stream_config, CONFIG_SECTION_STREAM, "reconnect delay", |
| 186 | stream_send.parents.reconnect_delay_s); |
| 187 | if(stream_send.parents.reconnect_delay_s < SENDER_MIN_RECONNECT_DELAY) |
| 188 | stream_send.parents.reconnect_delay_s = SENDER_MIN_RECONNECT_DELAY; |
| 189 | |
| 190 | stream_send.compression.enabled = |
| 191 | inicfg_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "enable compression", |
| 192 | stream_send.compression.enabled); |
| 193 | |
| 194 | stream_send.compression.levels[COMPRESSION_ALGORITHM_BROTLI] = (int)inicfg_get_number( |
| 195 | &stream_config, CONFIG_SECTION_STREAM, "brotli compression level", |
| 196 | stream_send.compression.levels[COMPRESSION_ALGORITHM_BROTLI]); |
| 197 | |
| 198 | stream_send.compression.levels[COMPRESSION_ALGORITHM_ZSTD] = (int)inicfg_get_number( |
| 199 | &stream_config, CONFIG_SECTION_STREAM, "zstd compression level", |
| 200 | stream_send.compression.levels[COMPRESSION_ALGORITHM_ZSTD]); |
| 201 | |
| 202 | stream_send.compression.levels[COMPRESSION_ALGORITHM_LZ4] = (int)inicfg_get_number( |
| 203 | &stream_config, CONFIG_SECTION_STREAM, "lz4 compression acceleration", |
| 204 | stream_send.compression.levels[COMPRESSION_ALGORITHM_LZ4]); |
| 205 | |
| 206 | stream_send.compression.levels[COMPRESSION_ALGORITHM_GZIP] = (int)inicfg_get_number( |
| 207 | &stream_config, CONFIG_SECTION_STREAM, "gzip compression level", |
| 208 | stream_send.compression.levels[COMPRESSION_ALGORITHM_GZIP]); |
| 209 | |
| 210 | stream_send.parents.h2o = inicfg_get_boolean( |
| 211 | &stream_config, CONFIG_SECTION_STREAM, "parent using h2o", |
| 212 | stream_send.parents.h2o); |
| 213 | |
| 214 | stream_send.parents.timeout_s = (int)inicfg_get_duration_seconds( |
| 215 | &stream_config, CONFIG_SECTION_STREAM, "timeout", |
| 216 | stream_send.parents.timeout_s); |
| 217 | |
| 218 | stream_send.buffer_max_size = (size_t)inicfg_get_number( |
| 219 | &stream_config, CONFIG_SECTION_STREAM, "buffer size bytes", |
| 220 | stream_send.buffer_max_size); |
| 221 | |
| 222 | stream_send.parents.default_port = (int)inicfg_get_number( |
| 223 | &stream_config, CONFIG_SECTION_STREAM, "default port", |
| 224 | stream_send.parents.default_port); |
| 225 | |
| 226 | stream_send.initial_clock_resync_iterations = (unsigned int)inicfg_get_number( |
| 227 | &stream_config, CONFIG_SECTION_STREAM, "initial clock resync iterations", |
| 228 | stream_send.initial_clock_resync_iterations); // TODO: REMOVE FOR SLEW / GAPFILLING |
| 229 | |
| 230 | netdata_ssl_validate_certificate_sender = !inicfg_get_boolean( |
| 231 | &stream_config, CONFIG_SECTION_STREAM, "ssl skip certificate verification", |
| 232 | !netdata_ssl_validate_certificate); |
| 233 | |
| 234 | if(!netdata_ssl_validate_certificate_sender) |
| 235 | nd_log_daemon(NDLP_NOTICE, "SSL: streaming senders will skip SSL certificates verification."); |
| 236 | |
| 237 | stream_send.parents.ssl_ca_path = string_strdupz(inicfg_get_path(&stream_config, CONFIG_SECTION_STREAM, "CApath", NULL)); |
| 238 | stream_send.parents.ssl_ca_file = string_strdupz(inicfg_get_filename(&stream_config, CONFIG_SECTION_STREAM, "CAfile", NULL)); |
| 239 | |
| 240 | if(stream_send.enabled && (!stream_send.parents.destination || !stream_send.api_key)) { |
| 241 | nd_log_daemon( |
| 242 | NDLP_ERR, |
| 243 | "STREAM [send]: cannot enable sending thread - missing required fields (destination: %s, api key: %s)", |
| 244 | stream_send.parents.destination ? "present" : "missing", |
| 245 | stream_send.api_key ? "present" : "missing"); |
| 246 | stream_send.enabled = false; |
| 247 | } |
| 248 | |
| 249 | stream_conf_is_parent(true); |
| 250 | } |
| 251 | |
| 252 | bool stream_conf_is_parent(bool recheck) { |
| 253 | static bool rc = false, queried = false; |
| 254 | if(!recheck && queried) |
| 255 | return rc; |
| 256 | |
| 257 | rc = stream_conf_has_api_enabled(&stream_config); |
| 258 | queried = true; |
| 259 | |
| 260 | return rc; |
| 261 | } |
| 262 | |
| 263 | bool stream_conf_is_child(void) { |
| 264 | return stream_send.enabled; |
| 265 | } |
| 266 | |
| 267 | void stream_conf_receiver_config(struct receiver_state *rpt, struct stream_receiver_config *config, const char *api_key, const char *machine_guid) { |
| 268 | config->mode = rrd_memory_mode_id( |
| 269 | inicfg_get(&stream_config, machine_guid, "db", |
| 270 | inicfg_get(&stream_config, api_key, "db", |
| 271 | rrd_memory_mode_name(default_rrd_memory_mode)))); |
| 272 | |
| 273 | if (unlikely(config->mode == RRD_DB_MODE_DBENGINE && !dbengine_enabled)) { |
| 274 | netdata_log_error("STREAM RCV '%s' [from [%s]:%s]: " |
| 275 | "dbengine is not enabled, falling back to default." |
| 276 | , rpt->hostname |
| 277 | , rpt->remote_ip, rpt->remote_port); |
| 278 | config->mode = default_rrd_memory_mode; |
| 279 | } |
| 280 | |
| 281 | config->history = (int) |
| 282 | inicfg_get_number(&stream_config, machine_guid, "retention", |
| 283 | inicfg_get_number(&stream_config, api_key, "retention", |
| 284 | default_rrd_history_entries)); |
| 285 | if(config->history < 5) config->history = 5; |
| 286 | |
| 287 | config->health.enabled = |
| 288 | inicfg_get_boolean_ondemand(&stream_config, machine_guid, "health enabled", |
| 289 | inicfg_get_boolean_ondemand(&stream_config, api_key, "health enabled", |
| 290 | health_plugin_enabled())); |
| 291 | |
| 292 | config->health.delay = |
| 293 | inicfg_get_duration_seconds(&stream_config, machine_guid, "postpone alerts on connect", |
| 294 | inicfg_get_duration_seconds(&stream_config, api_key, "postpone alerts on connect", |
| 295 | 60)); |
| 296 | |
| 297 | config->update_every = (int)inicfg_get_duration_seconds(&stream_config, machine_guid, "update every", config->update_every); |
| 298 | if(config->update_every < 0) config->update_every = 1; |
| 299 | |
| 300 | config->health.history = |
| 301 | inicfg_get_duration_seconds(&stream_config, machine_guid, "health log retention", |
| 302 | inicfg_get_duration_seconds(&stream_config, api_key, "health log retention", |
| 303 | HEALTH_LOG_RETENTION_DEFAULT)); |
| 304 | |
| 305 | config->send.enabled = |
| 306 | inicfg_get_boolean(&stream_config, machine_guid, "proxy enabled", |
| 307 | inicfg_get_boolean(&stream_config, api_key, "proxy enabled", |
| 308 | stream_send.enabled)); |
| 309 | |
| 310 | config->send.parents = string_strdupz( |
| 311 | inicfg_get(&stream_config, machine_guid, "proxy destination", |
| 312 | inicfg_get(&stream_config, api_key, "proxy destination", |
| 313 | string2str(stream_send.parents.destination)))); |
| 314 | |
| 315 | config->send.api_key = string_strdupz( |
| 316 | inicfg_get(&stream_config, machine_guid, "proxy api key", |
| 317 | inicfg_get(&stream_config, api_key, "proxy api key", |
| 318 | string2str(stream_send.api_key)))); |
| 319 | |
| 320 | config->send.charts_matching = string_strdupz( |
| 321 | inicfg_get(&stream_config, machine_guid, "proxy send charts matching", |
| 322 | inicfg_get(&stream_config, api_key, "proxy send charts matching", |
| 323 | string2str(stream_send.send_charts_matching)))); |
| 324 | |
| 325 | config->replication.enabled = |
| 326 | inicfg_get_boolean(&stream_config, machine_guid, "enable replication", |
| 327 | inicfg_get_boolean(&stream_config, api_key, "enable replication", |
| 328 | stream_receive.replication.enabled)); |
| 329 | |
| 330 | config->replication.period = |
| 331 | inicfg_get_duration_seconds(&stream_config, machine_guid, "replication period", |
| 332 | inicfg_get_duration_seconds(&stream_config, api_key, "replication period", |
| 333 | stream_receive.replication.period)); |
| 334 | |
| 335 | config->replication.step = |
| 336 | inicfg_get_duration_seconds(&stream_config, machine_guid, "replication step", |
| 337 | inicfg_get_duration_seconds(&stream_config, api_key, "replication step", |
| 338 | stream_receive.replication.step)); |
| 339 | |
| 340 | config->compression.enabled = |
| 341 | inicfg_get_boolean(&stream_config, machine_guid, "enable compression", |
| 342 | inicfg_get_boolean(&stream_config, api_key, "enable compression", |
| 343 | stream_send.compression.enabled)); |
| 344 | |
| 345 | if(config->compression.enabled) { |
| 346 | stream_parse_compression_order( |
| 347 | config, |
| 348 | inicfg_get( |
| 349 | &stream_config, machine_guid, "compression algorithms order", |
| 350 | inicfg_get(&stream_config, api_key, "compression algorithms order", STREAM_COMPRESSION_ALGORITHMS_ORDER))); |
| 351 | } |
| 352 | } |
| 353 | |
| 354 | bool stream_conf_is_key_type(const char *api_key, const char *type) { |
| 355 | const char *api_key_type = inicfg_get(&stream_config, api_key, "type", type); |
| 356 | if(!api_key_type || !*api_key_type) api_key_type = "unknown"; |
| 357 | return strcmp(api_key_type, type) == 0; |
| 358 | } |
| 359 | |
| 360 | bool stream_conf_api_key_is_enabled(const char *api_key, bool enabled) { |
| 361 | return inicfg_get_boolean(&stream_config, api_key, "enabled", enabled); |
| 362 | } |
| 363 | |
| 364 | bool stream_conf_api_key_allows_client(const char *api_key, const char *client_ip) { |
| 365 | SIMPLE_PATTERN *key_allow_from = simple_pattern_create( |
| 366 | inicfg_get(&stream_config, api_key, "allow from", "*"), |
| 367 | NULL, SIMPLE_PATTERN_EXACT, true); |
| 368 | |
| 369 | bool rc = true; |
| 370 | |
| 371 | if(key_allow_from) { |
| 372 | rc = simple_pattern_matches(key_allow_from, client_ip); |
| 373 | simple_pattern_free(key_allow_from); |
| 374 | } |
| 375 | |
| 376 | return rc; |
| 377 | } |