netdata/exporting/send_data.c

// SPDX-License-Identifier: GPL-3.0-or-later

#include "exporting_engine.h"

/**
 * Check if TLS is enabled in the configuration
 *
 * @param type buffer with response data.
 * @param options an instance data structure.
 * @return Returns 1 if TLS should be enabled, 0 otherwise.
 */
static int exporting_tls_is_enabled(EXPORTING_CONNECTOR_TYPE type, EXPORTING_OPTIONS options)
{
    return (type == EXPORTING_CONNECTOR_TYPE_GRAPHITE_HTTP ||
            type == EXPORTING_CONNECTOR_TYPE_JSON_HTTP ||
            type == EXPORTING_CONNECTOR_TYPE_OPENTSDB_HTTP ||
            type == EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE) &&
           options & EXPORTING_OPTION_USE_TLS;
}

/**
 * Discard response
 *
 * Discards a response received by an exporting connector instance after logging a sample of it to error.log
 *
 * @param buffer buffer with response data.
 * @param instance an instance data structure.
 * @return Always returns 0.
 */
int exporting_discard_response(BUFFER *buffer, struct instance *instance) {
#if NETDATA_INTERNAL_CHECKS
    char sample[1024];
    const char *s = buffer_tostring(buffer);
    char *d = sample, *e = &sample[sizeof(sample) - 1];

    for(; *s && d < e ;s++) {
        char c = *s;
        if(unlikely(!isprint(c))) c = ' ';
        *d++ = c;
    }
    *d = '\0';

    debug(
        D_EXPORTING,
        "EXPORTING: received %zu bytes from %s connector instance. Ignoring them. Sample: '%s'",
        buffer_strlen(buffer),
        instance->config.name,
        sample);
#else
    UNUSED(instance);
#endif /* NETDATA_INTERNAL_CHECKS */

    buffer_flush(buffer);
    return 0;
}

/**
 * Receive response
 *
 * @param sock communication socket.
 * @param instance an instance data structure.
 */
void simple_connector_receive_response(int *sock, struct instance *instance)
{
    static BUFFER *response = NULL;
    if (!response)
        response = buffer_create(4096);

    struct stats *stats = &instance->stats;
#ifdef ENABLE_HTTPS
    uint32_t options = (uint32_t)instance->config.options;
    struct simple_connector_data *connector_specific_data = instance->connector_specific_data;

    if (options & EXPORTING_OPTION_USE_TLS)
        ERR_clear_error();
#endif

    errno = 0;

    // loop through to collect all data
    while (*sock != -1 && errno != EWOULDBLOCK) {
        ssize_t r;
#ifdef ENABLE_HTTPS
        if (exporting_tls_is_enabled(instance->config.type, options) &&
            connector_specific_data->conn &&
            connector_specific_data->flags == NETDATA_SSL_HANDSHAKE_COMPLETE) {
            r = (ssize_t)SSL_read(connector_specific_data->conn,
                                  &response->buffer[response->len],
                                  (int) (response->size - response->len));

            if (likely(r > 0)) {
                // we received some data
                response->len += r;
                stats->received_bytes += r;
                stats->receptions++;
                continue;
            } else {
                int sslerrno = SSL_get_error(connector_specific_data->conn, (int) r);
                u_long sslerr = ERR_get_error();
                char buf[256];
                switch (sslerrno) {
                    case SSL_ERROR_WANT_READ:
                    case SSL_ERROR_WANT_WRITE:
                        goto endloop;
                    default:
                        ERR_error_string_n(sslerr, buf, sizeof(buf));
                        error("SSL error (%s)",
                              ERR_error_string((long)SSL_get_error(connector_specific_data->conn, (int)r), NULL));
                        goto endloop;
                }
            }
        } else {
            r = recv(*sock, &response->buffer[response->len], response->size - response->len, MSG_DONTWAIT);
        }
#else
        r = recv(*sock, &response->buffer[response->len], response->size - response->len, MSG_DONTWAIT);
#endif
        if (likely(r > 0)) {
            // we received some data
            response->len += r;
            stats->received_bytes += r;
            stats->receptions++;
        } else if (r == 0) {
            error("EXPORTING: '%s' closed the socket", instance->config.destination);
            close(*sock);
            *sock = -1;
        } else {
            // failed to receive data
            if (errno != EAGAIN && errno != EWOULDBLOCK) {
                error("EXPORTING: cannot receive data from '%s'.", instance->config.destination);
            }
        }

#ifdef UNIT_TESTING
        break;
#endif
    }
#ifdef ENABLE_HTTPS
endloop:
#endif

    // if we received data, process them
    if (buffer_strlen(response))
        instance->check_response(response, instance);
}

/**
 * Send buffer to a server
 *
 * @param sock communication socket.
 * @param failures the number of communication failures.
 * @param instance an instance data structure.
 */
void simple_connector_send_buffer(
    int *sock, int *failures, struct instance *instance, BUFFER *header, BUFFER *buffer, size_t buffered_metrics)
{
    int flags = 0;
#ifdef MSG_NOSIGNAL
    flags += MSG_NOSIGNAL;
#endif

#ifdef ENABLE_HTTPS
    uint32_t options = (uint32_t)instance->config.options;
    struct simple_connector_data *connector_specific_data = instance->connector_specific_data;

    if (options & EXPORTING_OPTION_USE_TLS)
        ERR_clear_error();
#endif

    struct stats *stats = &instance->stats;
    ssize_t header_sent_bytes = 0;
    ssize_t buffer_sent_bytes = 0;
    size_t header_len = buffer_strlen(header);
    size_t buffer_len = buffer_strlen(buffer);

#ifdef ENABLE_HTTPS
    if (exporting_tls_is_enabled(instance->config.type, options) &&
        connector_specific_data->conn &&
        connector_specific_data->flags == NETDATA_SSL_HANDSHAKE_COMPLETE) {
        if (header_len)
            header_sent_bytes = (ssize_t)SSL_write(connector_specific_data->conn, buffer_tostring(header), header_len);
        if ((size_t)header_sent_bytes == header_len)
            buffer_sent_bytes = (ssize_t)SSL_write(connector_specific_data->conn, buffer_tostring(buffer), buffer_len);
    } else {
        if (header_len)
            header_sent_bytes = send(*sock, buffer_tostring(header), header_len, flags);
        if ((size_t)header_sent_bytes == header_len)
            buffer_sent_bytes = send(*sock, buffer_tostring(buffer), buffer_len, flags);
    }
#else
    if (header_len)
        header_sent_bytes = send(*sock, buffer_tostring(header), header_len, flags);
    if ((size_t)header_sent_bytes == header_len)
        buffer_sent_bytes = send(*sock, buffer_tostring(buffer), buffer_len, flags);
#endif

    if ((size_t)buffer_sent_bytes == buffer_len) {
        // we sent the data successfully
        stats->transmission_successes++;
        stats->sent_metrics += buffered_metrics;
        stats->sent_bytes += buffer_sent_bytes;

        // reset the failures count
        *failures = 0;

        // empty the buffer
        buffer_flush(buffer);
    } else {
        // oops! we couldn't send (all or some of the) data
        error(
            "EXPORTING: failed to write data to '%s'. Willing to write %zu bytes, wrote %zd bytes. Will re-connect.",
            instance->config.destination,
            buffer_len,
            buffer_sent_bytes);
        stats->transmission_failures++;

        if(buffer_sent_bytes != -1)
            stats->sent_bytes += buffer_sent_bytes;

        // increment the counter we check for data loss
        (*failures)++;

        // close the socket - we will re-open it next time
        close(*sock);
        *sock = -1;
    }
}

/**
 * Simple connector worker
 *
 * Runs in a separate thread for every instance.
 *
 * @param instance_p an instance data structure.
 */
void simple_connector_worker(void *instance_p)
{
    struct instance *instance = (struct instance*)instance_p;
    struct simple_connector_data *connector_specific_data = instance->connector_specific_data;

#ifdef ENABLE_HTTPS
    uint32_t options = (uint32_t)instance->config.options;

    if (options & EXPORTING_OPTION_USE_TLS)
        ERR_clear_error();
#endif
    struct simple_connector_config *connector_specific_config = instance->config.connector_specific_config;

    int sock = -1;
    struct timeval timeout = { .tv_sec = (instance->config.timeoutms * 1000) / 1000000,
                               .tv_usec = (instance->config.timeoutms * 1000) % 1000000 };
    int failures = 0;

    while (!instance->engine->exit) {
        struct stats *stats = &instance->stats;
        int send_stats = 0;

        if (instance->data_is_ready)
            send_stats = 1;

        uv_mutex_lock(&instance->mutex);
        if (!connector_specific_data->first_buffer->used || failures) {
            while (!instance->data_is_ready)
                uv_cond_wait(&instance->cond_var, &instance->mutex);
            instance->data_is_ready = 0;
            send_stats = 1;
        }

        if (unlikely(instance->engine->exit)) {
            uv_mutex_unlock(&instance->mutex);
            break;
        }

        // ------------------------------------------------------------------------
        // detach buffer

        size_t buffered_metrics;

        if (!connector_specific_data->previous_buffer ||
            (connector_specific_data->previous_buffer == connector_specific_data->first_buffer &&
             connector_specific_data->first_buffer->used == 1)) {
            BUFFER *header, *buffer;

            header = connector_specific_data->first_buffer->header;
            buffer = connector_specific_data->first_buffer->buffer;
            connector_specific_data->buffered_metrics = connector_specific_data->first_buffer->buffered_metrics;
            connector_specific_data->buffered_bytes = connector_specific_data->first_buffer->buffered_bytes;

            buffered_metrics = connector_specific_data->buffered_metrics;

            buffer_flush(connector_specific_data->header);
            connector_specific_data->first_buffer->header = connector_specific_data->header;
            connector_specific_data->header = header;

            buffer_flush(connector_specific_data->buffer);
            connector_specific_data->first_buffer->buffer = connector_specific_data->buffer;
            connector_specific_data->buffer = buffer;
        } else {
            buffered_metrics = connector_specific_data->buffered_metrics;
        }

        uv_mutex_unlock(&instance->mutex);

        // ------------------------------------------------------------------------
        // if we are connected, receive a response, without blocking

        if (likely(sock != -1))
            simple_connector_receive_response(&sock, instance);

        // ------------------------------------------------------------------------
        // if we are not connected, connect to a data collecting server

        if (unlikely(sock == -1)) {
            size_t reconnects = 0;

            sock = connect_to_one_of(
                instance->config.destination,
                connector_specific_config->default_port,
                &timeout,
                &reconnects,
                connector_specific_data->connected_to,
                CONNECTED_TO_MAX);
#ifdef ENABLE_HTTPS
            if (exporting_tls_is_enabled(instance->config.type, options) && sock != -1) {
                if (netdata_exporting_ctx) {
                    if (sock_delnonblock(sock) < 0)
                        error("Exporting cannot remove the non-blocking flag from socket %d", sock);

                    if (connector_specific_data->conn == NULL) {
                        connector_specific_data->conn = SSL_new(netdata_exporting_ctx);
                        if (connector_specific_data->conn == NULL) {
                            error("Failed to allocate SSL structure to socket %d.", sock);
                            connector_specific_data->flags = NETDATA_SSL_NO_HANDSHAKE;
                        }
                    } else {
                        SSL_clear(connector_specific_data->conn);
                    }

                    if (connector_specific_data->conn) {
                        if (SSL_set_fd(connector_specific_data->conn, sock) != 1) {
                            error("Failed to set the socket to the SSL on socket fd %d.", sock);
                            connector_specific_data->flags = NETDATA_SSL_NO_HANDSHAKE;
                        } else {
                            connector_specific_data->flags = NETDATA_SSL_HANDSHAKE_COMPLETE;
                            SSL_set_connect_state(connector_specific_data->conn);
                            int err = SSL_connect(connector_specific_data->conn);
                            if (err != 1) {
                                err = SSL_get_error(connector_specific_data->conn, err);
                                error(
                                    "SSL cannot connect with the server:  %s ",
                                    ERR_error_string((long)SSL_get_error(connector_specific_data->conn, err), NULL));
                                connector_specific_data->flags = NETDATA_SSL_NO_HANDSHAKE;
                            } else {
                                info("Exporting established a SSL connection.");

                                struct timeval tv;
                                tv.tv_sec = timeout.tv_sec / 4;
                                tv.tv_usec = 0;

                                if (!tv.tv_sec)
                                    tv.tv_sec = 2;

                                if (setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, (const char *)&tv, sizeof(tv)))
                                    error("Cannot set timeout to socket %d, this can block communication", sock);
                            }
                        }
                    }
                }
            }
#endif

            stats->reconnects += reconnects;
        }

        if (unlikely(instance->engine->exit))
            break;

        // ------------------------------------------------------------------------
        // if we are connected, send our buffer to the data collecting server

        failures = 0;

        if (likely(sock != -1)) {
            simple_connector_send_buffer(
                &sock,
                &failures,
                instance,
                connector_specific_data->header,
                connector_specific_data->buffer,
                buffered_metrics);
        } else {
            error("EXPORTING: failed to update '%s'", instance->config.destination);
            stats->transmission_failures++;

            // increment the counter we check for data loss
            failures++;
        }

        if (!failures) {
            connector_specific_data->first_buffer->buffered_metrics =
                connector_specific_data->first_buffer->buffered_bytes = connector_specific_data->first_buffer->used = 0;
            connector_specific_data->first_buffer = connector_specific_data->first_buffer->next;
        }

        if (unlikely(instance->engine->exit))
            break;

        if (send_stats) {
            uv_mutex_lock(&instance->mutex);

            stats->buffered_metrics = connector_specific_data->total_buffered_metrics;

            send_internal_metrics(instance);

            stats->buffered_metrics = 0;

            // reset the internal monitoring chart counters
            connector_specific_data->total_buffered_metrics =
            stats->buffered_bytes =
            stats->receptions =
            stats->received_bytes =
            stats->sent_metrics =
            stats->sent_bytes =
            stats->transmission_successes =
            stats->transmission_failures =
            stats->reconnects =
            stats->data_lost_events =
            stats->lost_metrics =
            stats->lost_bytes = 0;

            uv_mutex_unlock(&instance->mutex);
        }

#ifdef UNIT_TESTING
        return;
#endif
    }

#if ENABLE_PROMETHEUS_REMOTE_WRITE
    if (instance->config.type == EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE)
        clean_prometheus_remote_write(instance);
#endif

    simple_connector_cleanup(instance);
}