| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #include "exporting_engine.h" |
| 4 | |
| 5 | /** |
| 6 | * Check if TLS is enabled in the configuration |
| 7 | * |
| 8 | * @param type buffer with response data. |
| 9 | * @param options an instance data structure. |
| 10 | * @return Returns 1 if TLS should be enabled, 0 otherwise. |
| 11 | */ |
| 12 | static int exporting_tls_is_enabled(EXPORTING_CONNECTOR_TYPE type __maybe_unused, EXPORTING_OPTIONS options __maybe_unused) |
| 13 | { |
| 14 | |
| 15 | return (type == EXPORTING_CONNECTOR_TYPE_GRAPHITE_HTTP || |
| 16 | type == EXPORTING_CONNECTOR_TYPE_JSON_HTTP || |
| 17 | type == EXPORTING_CONNECTOR_TYPE_OPENTSDB_HTTP || |
| 18 | type == EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE) && |
| 19 | options & EXPORTING_OPTION_USE_TLS; |
| 20 | } |
| 21 | |
| 22 | /** |
| 23 | * Discard response |
| 24 | * |
| 25 | * Discards a response received by an exporting connector instance after logging a sample of it to error.log |
| 26 | * |
| 27 | * @param buffer buffer with response data. |
| 28 | * @param instance an instance data structure. |
| 29 | * @return Always returns 0. |
| 30 | */ |
| 31 | int exporting_discard_response(BUFFER *buffer, struct instance *instance) { |
| 32 | #ifdef NETDATA_INTERNAL_CHECKS |
| 33 | char sample[1024]; |
| 34 | const char *s = buffer_tostring(buffer); |
| 35 | char *d = sample, *e = &sample[sizeof(sample) - 1]; |
| 36 | |
| 37 | for(; *s && d < e ;s++) { |
| 38 | char c = *s; |
| 39 | if(unlikely(!isprint(c))) c = ' '; |
| 40 | *d++ = c; |
| 41 | } |
| 42 | *d = '\0'; |
| 43 | |
| 44 | netdata_log_debug(D_EXPORTING, |
| 45 | "EXPORTING: received %zu bytes from %s connector instance. Ignoring them. Sample: '%s'", |
| 46 | buffer_strlen(buffer), |
| 47 | instance->config.name, |
| 48 | sample); |
| 49 | #else |
| 50 | UNUSED(instance); |
| 51 | #endif /* NETDATA_INTERNAL_CHECKS */ |
| 52 | |
| 53 | buffer_flush(buffer); |
| 54 | return 0; |
| 55 | } |
| 56 | |
| 57 | /** |
| 58 | * Receive response |
| 59 | * |
| 60 | * @param sock communication socket. |
| 61 | * @param instance an instance data structure. |
| 62 | */ |
| 63 | void simple_connector_receive_response(int *sock, struct instance *instance) |
| 64 | { |
| 65 | static BUFFER *response = NULL; |
| 66 | if (!response) |
| 67 | response = buffer_create(4096, &netdata_buffers_statistics.buffers_exporters); |
| 68 | |
| 69 | struct stats *stats = &instance->stats; |
| 70 | uint32_t options = (uint32_t)instance->config.options; |
| 71 | struct simple_connector_data *connector_specific_data = instance->connector_specific_data; |
| 72 | |
| 73 | if (options & EXPORTING_OPTION_USE_TLS) |
| 74 | ERR_clear_error(); |
| 75 | |
| 76 | errno_clear(); |
| 77 | |
| 78 | // loop through to collect all data |
| 79 | while (*sock != -1 && errno != EWOULDBLOCK) { |
| 80 | ssize_t r; |
| 81 | if (SSL_connection(&connector_specific_data->ssl)) |
| 82 | r = netdata_ssl_read(&connector_specific_data->ssl, &response->buffer[response->len], |
| 83 | (int) (response->size - response->len)); |
| 84 | else |
| 85 | r = recv(*sock, &response->buffer[response->len], response->size - response->len, MSG_DONTWAIT); |
| 86 | |
| 87 | if (likely(r > 0)) { |
| 88 | // we received some data |
| 89 | response->len += r; |
| 90 | stats->received_bytes += r; |
| 91 | stats->receptions++; |
| 92 | } |
| 93 | else if (r == 0) { |
| 94 | netdata_log_error("EXPORTING: '%s' closed the socket", instance->config.destination); |
| 95 | close(*sock); |
| 96 | *sock = -1; |
| 97 | } |
| 98 | else { |
| 99 | // failed to receive data |
| 100 | if (errno != EAGAIN && errno != EWOULDBLOCK) { |
| 101 | netdata_log_error("EXPORTING: cannot receive data from '%s'.", instance->config.destination); |
| 102 | close(*sock); |
| 103 | *sock = -1; |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | #ifdef UNIT_TESTING |
| 108 | break; |
| 109 | #endif |
| 110 | } |
| 111 | |
| 112 | // if we received data, process them |
| 113 | if (buffer_strlen(response)) |
| 114 | instance->check_response(response, instance); |
| 115 | } |
| 116 | |
| 117 | /** |
| 118 | * Send buffer to a server |
| 119 | * |
| 120 | * @param sock communication socket. |
| 121 | * @param failures the number of communication failures. |
| 122 | * @param instance an instance data structure. |
| 123 | */ |
| 124 | void simple_connector_send_buffer( |
| 125 | int *sock, int *failures, struct instance *instance, BUFFER *header, BUFFER *buffer, size_t buffered_metrics) |
| 126 | { |
| 127 | int flags = 0; |
| 128 | #ifdef MSG_NOSIGNAL |
| 129 | flags += MSG_NOSIGNAL; |
| 130 | #endif |
| 131 | |
| 132 | // Safety check to prevent NULL pointer crashes, but don't allocate new memory |
| 133 | if (unlikely(!buffer || !header)) { |
| 134 | netdata_log_error("EXPORTING: NULL %s passed to simple_connector_send_buffer for instance %s", |
| 135 | (!buffer && !header) ? "buffer and header" : (!buffer ? "buffer" : "header"), |
| 136 | instance->config.name ? instance->config.name : "unknown"); |
| 137 | (*failures)++; |
| 138 | return; |
| 139 | } |
| 140 | |
| 141 | uint32_t options = (uint32_t)instance->config.options; |
| 142 | struct simple_connector_data *connector_specific_data = instance->connector_specific_data; |
| 143 | |
| 144 | if (options & EXPORTING_OPTION_USE_TLS) |
| 145 | ERR_clear_error(); |
| 146 | |
| 147 | struct stats *stats = &instance->stats; |
| 148 | ssize_t header_sent_bytes = 0; |
| 149 | ssize_t buffer_sent_bytes = 0; |
| 150 | size_t header_len = buffer_strlen(header); |
| 151 | size_t buffer_len = buffer_strlen(buffer); |
| 152 | |
| 153 | if (SSL_connection(&connector_specific_data->ssl)) { |
| 154 | |
| 155 | if (header_len) |
| 156 | header_sent_bytes = netdata_ssl_write(&connector_specific_data->ssl, buffer_tostring(header), header_len); |
| 157 | |
| 158 | if ((size_t)header_sent_bytes == header_len) |
| 159 | buffer_sent_bytes = netdata_ssl_write(&connector_specific_data->ssl, buffer_tostring(buffer), buffer_len); |
| 160 | |
| 161 | } |
| 162 | else { |
| 163 | if (header_len) |
| 164 | header_sent_bytes = send(*sock, buffer_tostring(header), header_len, flags); |
| 165 | if ((size_t)header_sent_bytes == header_len) |
| 166 | buffer_sent_bytes = send(*sock, buffer_tostring(buffer), buffer_len, flags); |
| 167 | } |
| 168 | |
| 169 | if ((size_t)buffer_sent_bytes == buffer_len) { |
| 170 | // we sent the data successfully |
| 171 | stats->transmission_successes++; |
| 172 | stats->sent_metrics += buffered_metrics; |
| 173 | stats->sent_bytes += buffer_sent_bytes; |
| 174 | |
| 175 | // reset the failures count |
| 176 | *failures = 0; |
| 177 | |
| 178 | // empty the buffer |
| 179 | buffer_flush(buffer); |
| 180 | } else { |
| 181 | // oops! we couldn't send (all or some of the) data |
| 182 | netdata_log_error( |
| 183 | "EXPORTING: failed to write data to '%s'. Willing to write %zu bytes, wrote %zd bytes. Will re-connect.", |
| 184 | instance->config.destination, |
| 185 | buffer_len, |
| 186 | buffer_sent_bytes); |
| 187 | stats->transmission_failures++; |
| 188 | |
| 189 | if(buffer_sent_bytes != -1) |
| 190 | stats->sent_bytes += buffer_sent_bytes; |
| 191 | |
| 192 | // increment the counter we check for data loss |
| 193 | (*failures)++; |
| 194 | |
| 195 | // close the socket - we will re-open it next time |
| 196 | close(*sock); |
| 197 | *sock = -1; |
| 198 | } |
| 199 | } |
| 200 | |
| 201 | /** |
| 202 | * Simple connector worker |
| 203 | * |
| 204 | * Runs in a separate thread for every instance. |
| 205 | * |
| 206 | * @param instance_p an instance data structure. |
| 207 | */ |
| 208 | void simple_connector_worker(void *instance_p) |
| 209 | { |
| 210 | struct instance *instance = (struct instance*)instance_p; |
| 211 | struct simple_connector_data *connector_specific_data = instance->connector_specific_data; |
| 212 | |
| 213 | // Thread name is set during creation |
| 214 | |
| 215 | uint32_t options = (uint32_t)instance->config.options; |
| 216 | |
| 217 | if (options & EXPORTING_OPTION_USE_TLS) |
| 218 | ERR_clear_error(); |
| 219 | |
| 220 | struct simple_connector_config *connector_specific_config = instance->config.connector_specific_config; |
| 221 | |
| 222 | int sock = -1; |
| 223 | struct timeval timeout = { .tv_sec = (instance->config.timeoutms * 1000) / 1000000, |
| 224 | .tv_usec = (instance->config.timeoutms * 1000) % 1000000 }; |
| 225 | int failures = 0; |
| 226 | |
| 227 | while (!instance->engine->exit) { |
| 228 | struct stats *stats = &instance->stats; |
| 229 | int send_stats = 0; |
| 230 | |
| 231 | if (instance->data_is_ready) |
| 232 | send_stats = 1; |
| 233 | |
| 234 | netdata_mutex_lock(&instance->mutex); |
| 235 | if (!connector_specific_data->first_buffer->used || failures) { |
| 236 | while (!instance->data_is_ready) |
| 237 | netdata_cond_wait(&instance->cond_var, &instance->mutex); |
| 238 | instance->data_is_ready = 0; |
| 239 | send_stats = 1; |
| 240 | } |
| 241 | |
| 242 | if (unlikely(instance->engine->exit)) { |
| 243 | netdata_mutex_unlock(&instance->mutex); |
| 244 | break; |
| 245 | } |
| 246 | |
| 247 | // ------------------------------------------------------------------------ |
| 248 | // detach buffer |
| 249 | |
| 250 | size_t buffered_metrics; |
| 251 | |
| 252 | if (!connector_specific_data->previous_buffer || |
| 253 | (connector_specific_data->previous_buffer == connector_specific_data->first_buffer && |
| 254 | connector_specific_data->first_buffer->used == 1)) { |
| 255 | BUFFER *header, *buffer; |
| 256 | |
| 257 | header = connector_specific_data->first_buffer->header; |
| 258 | buffer = connector_specific_data->first_buffer->buffer; |
| 259 | connector_specific_data->buffered_metrics = connector_specific_data->first_buffer->buffered_metrics; |
| 260 | connector_specific_data->buffered_bytes = connector_specific_data->first_buffer->buffered_bytes; |
| 261 | |
| 262 | buffered_metrics = connector_specific_data->buffered_metrics; |
| 263 | |
| 264 | buffer_flush(connector_specific_data->header); |
| 265 | connector_specific_data->first_buffer->header = connector_specific_data->header; |
| 266 | connector_specific_data->header = header; |
| 267 | |
| 268 | buffer_flush(connector_specific_data->buffer); |
| 269 | connector_specific_data->first_buffer->buffer = connector_specific_data->buffer; |
| 270 | connector_specific_data->buffer = buffer; |
| 271 | } else { |
| 272 | buffered_metrics = connector_specific_data->buffered_metrics; |
| 273 | } |
| 274 | |
| 275 | netdata_mutex_unlock(&instance->mutex); |
| 276 | |
| 277 | // ------------------------------------------------------------------------ |
| 278 | // if we are connected, receive a response, without blocking |
| 279 | |
| 280 | if (likely(sock != -1)) |
| 281 | simple_connector_receive_response(&sock, instance); |
| 282 | |
| 283 | // ------------------------------------------------------------------------ |
| 284 | // if we are not connected, connect to a data collecting server |
| 285 | |
| 286 | if (unlikely(sock == -1)) { |
| 287 | size_t reconnects = 0; |
| 288 | |
| 289 | sock = connect_to_one_of_urls( |
| 290 | instance->config.destination, |
| 291 | connector_specific_config->default_port, |
| 292 | &timeout, |
| 293 | &reconnects, |
| 294 | connector_specific_data->connected_to, |
| 295 | CONNECTED_TO_MAX); |
| 296 | |
| 297 | if (exporting_tls_is_enabled(instance->config.type, options) && sock != -1) { |
| 298 | if (netdata_ssl_exporting_ctx) { |
| 299 | if (sock_setnonblock(sock, false) != 0) |
| 300 | netdata_log_error("Exporting cannot remove the non-blocking flag from socket %d", sock); |
| 301 | |
| 302 | if(netdata_ssl_open(&connector_specific_data->ssl, netdata_ssl_exporting_ctx, sock)) { |
| 303 | if(netdata_ssl_connect(&connector_specific_data->ssl)) { |
| 304 | netdata_log_info("Exporting established a SSL connection."); |
| 305 | |
| 306 | struct timeval tv; |
| 307 | tv.tv_sec = timeout.tv_sec / 4; |
| 308 | tv.tv_usec = 0; |
| 309 | |
| 310 | if (!tv.tv_sec) |
| 311 | tv.tv_sec = 2; |
| 312 | |
| 313 | if (setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char *)&tv, sizeof(tv))) |
| 314 | netdata_log_error("Cannot set timeout to socket %d, this can block communication", sock); |
| 315 | } |
| 316 | } |
| 317 | } |
| 318 | } |
| 319 | |
| 320 | stats->reconnects += reconnects; |
| 321 | } |
| 322 | |
| 323 | if (unlikely(instance->engine->exit)) |
| 324 | break; |
| 325 | |
| 326 | // ------------------------------------------------------------------------ |
| 327 | // if we are connected, send our buffer to the data collecting server |
| 328 | |
| 329 | failures = 0; |
| 330 | |
| 331 | if (likely(sock != -1)) { |
| 332 | simple_connector_send_buffer( |
| 333 | &sock, |
| 334 | &failures, |
| 335 | instance, |
| 336 | connector_specific_data->header, |
| 337 | connector_specific_data->buffer, |
| 338 | buffered_metrics); |
| 339 | } else { |
| 340 | netdata_log_error("EXPORTING: failed to update '%s'", instance->config.destination); |
| 341 | stats->transmission_failures++; |
| 342 | |
| 343 | // increment the counter we check for data loss |
| 344 | failures++; |
| 345 | } |
| 346 | |
| 347 | if (!failures) { |
| 348 | connector_specific_data->first_buffer->buffered_metrics = |
| 349 | connector_specific_data->first_buffer->buffered_bytes = connector_specific_data->first_buffer->used = 0; |
| 350 | connector_specific_data->first_buffer = connector_specific_data->first_buffer->next; |
| 351 | } |
| 352 | |
| 353 | if (unlikely(instance->engine->exit)) |
| 354 | break; |
| 355 | |
| 356 | if (send_stats) { |
| 357 | netdata_mutex_lock(&instance->mutex); |
| 358 | |
| 359 | stats->buffered_metrics = connector_specific_data->total_buffered_metrics; |
| 360 | |
| 361 | send_internal_metrics(instance); |
| 362 | |
| 363 | stats->buffered_metrics = 0; |
| 364 | |
| 365 | // reset the internal monitoring chart counters |
| 366 | connector_specific_data->total_buffered_metrics = |
| 367 | stats->buffered_bytes = |
| 368 | stats->receptions = |
| 369 | stats->received_bytes = |
| 370 | stats->sent_metrics = |
| 371 | stats->sent_bytes = |
| 372 | stats->transmission_successes = |
| 373 | stats->transmission_failures = |
| 374 | stats->reconnects = |
| 375 | stats->data_lost_events = |
| 376 | stats->lost_metrics = |
| 377 | stats->lost_bytes = 0; |
| 378 | |
| 379 | netdata_mutex_unlock(&instance->mutex); |
| 380 | } |
| 381 | |
| 382 | #ifdef UNIT_TESTING |
| 383 | return; |
| 384 | #endif |
| 385 | } |
| 386 | |
| 387 | #ifdef ENABLE_PROMETHEUS_REMOTE_WRITE |
| 388 | if (instance->config.type == EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE) |
| 389 | clean_prometheus_remote_write(instance); |
| 390 | #endif |
| 391 | |
| 392 | simple_connector_cleanup(instance); |
| 393 | } |