diff options
Diffstat (limited to 'streaming')
-rw-r--r-- | streaming/compression.c | 236 | ||||
-rw-r--r-- | streaming/receiver.c | 428 | ||||
-rw-r--r-- | streaming/replication.c | 501 | ||||
-rw-r--r-- | streaming/replication.h | 5 | ||||
-rw-r--r-- | streaming/rrdpush.c | 3 | ||||
-rw-r--r-- | streaming/rrdpush.h | 33 | ||||
-rw-r--r-- | streaming/sender.c | 257 |
7 files changed, 912 insertions, 551 deletions
diff --git a/streaming/compression.c b/streaming/compression.c index 1fddc02b91..7ba9dbf198 100644 --- a/streaming/compression.c +++ b/streaming/compression.c @@ -5,6 +5,7 @@ #define STREAM_COMPRESSION_MSG "STREAM_COMPRESSION" +// signature MUST end with a newline #define SIGNATURE ((uint32_t)('z' | 0x80) | (0x80 << 8) | (0x80 << 16) | ('\n' << 24)) #define SIGNATURE_MASK ((uint32_t)0xff | (0x80 << 8) | (0x80 << 16) | (0xff << 24)) #define SIGNATURE_SIZE 4 @@ -29,7 +30,7 @@ static void lz4_compressor_reset(struct compressor_state *state) if (state->data) { if (state->data->stream) { LZ4_resetStream_fast(state->data->stream); - info("%s: Compressor Reset", STREAM_COMPRESSION_MSG); + internal_error(true, "%s: compressor reset", STREAM_COMPRESSION_MSG); } state->data->input_ring_buffer_pos = 0; } @@ -139,11 +140,12 @@ struct compressor_state *create_compressor() /* * LZ4 streaming API decompressor specific data */ -struct decompressor_data { - LZ4_streamDecode_t *stream; - char *stream_buffer; - size_t stream_buffer_size; - size_t stream_buffer_pos; +struct decompressor_stream { + LZ4_streamDecode_t *lz4_stream; + char *buffer; + size_t size; + size_t write_at; + size_t read_at; }; /* @@ -151,12 +153,12 @@ struct decompressor_data { */ static void lz4_decompressor_reset(struct decompressor_state *state) { - if (state->data) { - if (state->data->stream) - LZ4_setStreamDecode(state->data->stream, NULL, 0); - state->data->stream_buffer_pos = 0; - state->buffer_len = 0; - state->out_buffer_len = 0; + if (state->stream) { + if (state->stream->lz4_stream) + LZ4_setStreamDecode(state->stream->lz4_stream, NULL, 0); + + state->stream->write_at = 0; + state->stream->read_at = 0; } } @@ -167,177 +169,129 @@ static void lz4_decompressor_destroy(struct decompressor_state **state) { if (state && *state) { struct decompressor_state *s = *state; - if (s->data) { + if (s->stream) { debug(D_STREAM, "%s: Destroying decompressor.", STREAM_COMPRESSION_MSG); - if (s->data->stream) - LZ4_freeStreamDecode(s->data->stream); - freez(s->data->stream_buffer); - freez(s->data); + if (s->stream->lz4_stream) + LZ4_freeStreamDecode(s->stream->lz4_stream); + freez(s->stream->buffer); + freez(s->stream); } - freez(s->buffer); freez(s); *state = NULL; } } -static size_t decode_compress_header(const char *data, size_t data_size) -{ - if (!data || !data_size) +static size_t decode_compress_header(const char *data, size_t data_size) { + if (unlikely(!data || !data_size)) return 0; - if (data_size < SIGNATURE_SIZE) + + if (unlikely(data_size != SIGNATURE_SIZE)) return 0; + uint32_t sign = *(uint32_t *)data; - if ((sign & SIGNATURE_MASK) != SIGNATURE) + if (unlikely((sign & SIGNATURE_MASK) != SIGNATURE)) return 0; + size_t length = ((sign >> 8) & 0x7f) | ((sign >> 9) & (0x7f << 7)); return length; } /* - * Check input data for the compression header - * Return the size of compressed data or 0 for uncompressed data - */ -size_t is_compressed_data(const char *data, size_t data_size) -{ - return decode_compress_header(data, data_size); -} - -/* * Start the collection of compressed data in an internal buffer * Return the size of compressed data or 0 for uncompressed data */ -static size_t lz4_decompressor_start(struct decompressor_state *state, const char *header, size_t header_size) -{ - size_t length = decode_compress_header(header, header_size); - if (!length) - return 0; - - if (!state->buffer) { - state->buffer = mallocz(length); - state->buffer_size = length; - } else if (state->buffer_size < length) { - state->buffer = reallocz(state->buffer, length); - state->buffer_size = length; - } - state->buffer_len = length; - state->buffer_pos = 0; - state->out_buffer_pos = 0; - state->out_buffer_len = 0; - return length; -} - -/* - * Add a chunk of compressed data to the internal buffer - * Return the current size of compressed data or 0 for error - */ -static size_t lz4_decompressor_put(struct decompressor_state *state, const char *data, size_t size) -{ - if (!state || !size || !data) - return 0; - if (!state->buffer) - fatal("STREAM: No decompressor buffer allocated"); - - if (state->buffer_pos + size > state->buffer_len) { - error("STREAM: Decompressor buffer overflow %lu + %lu > %lu", - (long unsigned int) state->buffer_pos, (long unsigned int) size, - (long unsigned int) state->buffer_len); - size = state->buffer_len - state->buffer_pos; - } - memcpy(state->buffer + state->buffer_pos, data, size); - state->buffer_pos += size; - return state->buffer_pos; -} +static size_t lz4_decompressor_start(struct decompressor_state *state __maybe_unused, const char *header, size_t header_size) { + if(unlikely(state->stream->read_at != state->stream->write_at)) + fatal("%s: asked to decompress new data, while there are unread data in the decompression buffer!" + , STREAM_COMPRESSION_MSG); -static size_t saving_percent(size_t comp_len, size_t src_len) -{ - if (comp_len > src_len) - comp_len = src_len; - if (!src_len) - return 0; - return 100 - comp_len * 100 / src_len; + return decode_compress_header(header, header_size); } /* * Decompress the compressed data in the internal buffer * Return the size of uncompressed data or 0 for error */ -static size_t lz4_decompressor_decompress(struct decompressor_state *state) -{ - if (!state) - return 0; - if (!state->buffer) { - error("%s: No decompressor buffer allocated", STREAM_COMPRESSION_MSG); +static size_t lz4_decompressor_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) { + if (unlikely(!state || !compressed_data || !compressed_size)) return 0; + + if(unlikely(state->stream->read_at != state->stream->write_at)) + fatal("%s: asked to decompress new data, while there are unread data in the decompression buffer!" + , STREAM_COMPRESSION_MSG); + + if (unlikely(state->stream->write_at >= state->stream->size / 2)) { + state->stream->write_at = 0; + state->stream->read_at = 0; } - - long int decompressed_size = LZ4_decompress_safe_continue(state->data->stream, state->buffer, - state->data->stream_buffer + state->data->stream_buffer_pos, - state->buffer_len, state->data->stream_buffer_size - state->data->stream_buffer_pos); - if (decompressed_size < 0) { - error("%s: Decompressor error %ld", STREAM_COMPRESSION_MSG, decompressed_size); + + long int decompressed_size = LZ4_decompress_safe_continue( + state->stream->lz4_stream + , compressed_data + , state->stream->buffer + state->stream->write_at + , (int)compressed_size + , (int)(state->stream->size - state->stream->write_at) + ); + + if (unlikely(decompressed_size < 0)) { + error("%s: decompressor returned negative decompressed bytes: %ld", STREAM_COMPRESSION_MSG, decompressed_size); return 0; } - state->out_buffer = state->data->stream_buffer + state->data->stream_buffer_pos; - state->data->stream_buffer_pos += decompressed_size; - if (state->data->stream_buffer_pos >= state->data->stream_buffer_size - COMPRESSION_MAX_MSG_SIZE) - state->data->stream_buffer_pos = 0; - state->out_buffer_len = decompressed_size; - state->out_buffer_pos = 0; + if(unlikely(decompressed_size + state->stream->write_at > state->stream->size)) + fatal("%s: decompressor overflown the stream_buffer. size: %zu, pos: %zu, added: %ld, exceeding the buffer by %zu" + , STREAM_COMPRESSION_MSG + , state->stream->size + , state->stream->write_at + , decompressed_size + , state->stream->write_at + decompressed_size - state->stream->size + ); - // Some compression statistics - size_t old_avg_saving = saving_percent(state->total_compressed, state->total_uncompressed); - size_t old_avg_size = state->packet_count ? state->total_uncompressed / state->packet_count : 0; + state->stream->write_at += decompressed_size; - state->total_compressed += state->buffer_len + SIGNATURE_SIZE; + // statistics + state->total_compressed += compressed_size + SIGNATURE_SIZE; state->total_uncompressed += decompressed_size; state->packet_count++; - size_t saving = saving_percent(state->buffer_len, decompressed_size); - size_t avg_saving = saving_percent(state->total_compressed, state->total_uncompressed); - size_t avg_size = state->total_uncompressed / state->packet_count; - - (void)saving; - - if (old_avg_saving != avg_saving || old_avg_size != avg_size){ - debug(D_STREAM, "%s: Saving: %lu%% (avg. %lu%%), avg.size: %lu", STREAM_COMPRESSION_MSG, - (long unsigned int) saving, (long unsigned int) avg_saving, (long unsigned int) avg_size); - } return decompressed_size; } /* * Return the size of uncompressed data left in the internal buffer or 0 for error */ -static size_t lz4_decompressor_decompressed_bytes_in_buffer(struct decompressor_state *state) -{ - return state->out_buffer_len ? - state->out_buffer_len - state->out_buffer_pos : 0; +static size_t lz4_decompressor_decompressed_bytes_in_buffer(struct decompressor_state *state) { + if(unlikely(state->stream->read_at > state->stream->write_at)) + fatal("%s: invalid read/write stream positions" + , STREAM_COMPRESSION_MSG); + + return state->stream->write_at - state->stream->read_at; } /* * Fill the buffer provided with uncompressed data from the internal buffer * Return the size of uncompressed data copied or 0 for error */ -static size_t lz4_decompressor_get(struct decompressor_state *state, char *data, size_t size) -{ - if (!state || !size || !data) +static size_t lz4_decompressor_get(struct decompressor_state *state, char *dst, size_t size) { + if (unlikely(!state || !size || !dst)) return 0; - if (!state->out_buffer) - fatal("%s: No decompressor output buffer allocated", STREAM_COMPRESSION_MSG); - if (state->out_buffer_pos + size > state->out_buffer_len) - size = state->out_buffer_len - state->out_buffer_pos; - - char *p = state->out_buffer + state->out_buffer_pos, *endp = p + size, *last_lf = NULL; - for (; p < endp; ++p) - if (*p == '\n' || *p == 0) - last_lf = p; - if (last_lf) - size = last_lf + 1 - (state->out_buffer + state->out_buffer_pos); - - memcpy(data, state->out_buffer + state->out_buffer_pos, size); - state->out_buffer_pos += size; - return size; + + size_t remaining = lz4_decompressor_decompressed_bytes_in_buffer(state); + if(unlikely(!remaining)) + return 0; + + size_t bytes_to_return = size; + if(bytes_to_return > remaining) + bytes_to_return = remaining; + + memcpy(dst, state->stream->buffer + state->stream->read_at, bytes_to_return); + state->stream->read_at += bytes_to_return; + + if(unlikely(state->stream->read_at > state->stream->write_at)) + fatal("%s: invalid read/write stream positions" + , STREAM_COMPRESSION_MSG); + + return bytes_to_return; } /* @@ -347,20 +301,20 @@ static size_t lz4_decompressor_get(struct decompressor_state *state, char *data, struct decompressor_state *create_decompressor() { struct decompressor_state *state = callocz(1, sizeof(struct decompressor_state)); + state->signature_size = SIGNATURE_SIZE; state->reset = lz4_decompressor_reset; state->start = lz4_decompressor_start; - state->put = lz4_decompressor_put; state->decompress = lz4_decompressor_decompress; state->get = lz4_decompressor_get; state->decompressed_bytes_in_buffer = lz4_decompressor_decompressed_bytes_in_buffer; state->destroy = lz4_decompressor_destroy; - state->data = callocz(1, sizeof(struct decompressor_data)); - fatal_assert(state->data); - state->data->stream = LZ4_createStreamDecode(); - state->data->stream_buffer_size = LZ4_decoderRingBufferSize(COMPRESSION_MAX_MSG_SIZE); - state->data->stream_buffer = mallocz(state->data->stream_buffer_size); - fatal_assert(state->data->stream_buffer); + state->stream = callocz(1, sizeof(struct decompressor_stream)); + fatal_assert(state->stream); + state->stream->lz4_stream = LZ4_createStreamDecode(); + state->stream->size = LZ4_decoderRingBufferSize(COMPRESSION_MAX_MSG_SIZE) * 2; + state->stream->buffer = mallocz(state->stream->size); + fatal_assert(state->stream->buffer); state->reset(state); debug(D_STREAM, "%s: Initialize streaming decompression!", STREAM_COMPRESSION_MSG); return state; diff --git a/streaming/receiver.c b/streaming/receiver.c index 40673f05b4..32b510e7ce 100644 --- a/streaming/receiver.c +++ b/streaming/receiver.c @@ -3,7 +3,12 @@ #include "rrdpush.h" #include "parser/parser.h" +// IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly #define WORKER_RECEIVER_JOB_BYTES_READ (WORKER_PARSER_FIRST_JOB - 1) +#define WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED (WORKER_PARSER_FIRST_JOB - 2) + +// this has to be the same at parser.h +#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 3) #if WORKER_PARSER_FIRST_JOB < 1 #error The define WORKER_PARSER_FIRST_JOB needs to be at least 1 @@ -110,185 +115,182 @@ PARSER_RC streaming_claimed_id(char **words, size_t num_words, void *user) return PARSER_RC_OK; } +static int read_stream(struct receiver_state *r, char* buffer, size_t size) { + if(unlikely(!size)) { + internal_error(true, "%s() asked to read zero bytes", __FUNCTION__); + return 0; + } -#ifndef ENABLE_COMPRESSION -/* The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing. - */ -static int receiver_read(struct receiver_state *r, FILE *fp) { #ifdef ENABLE_HTTPS - if (r->ssl.conn && !r->ssl.flags) { - ERR_clear_error(); - int desired = sizeof(r->read_buffer) - r->read_len - 1; - int ret = SSL_read(r->ssl.conn, r->read_buffer + r->read_len, desired); - if (ret > 0 ) { - r->read_len += ret; - worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret); - return 0; - } - // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket - u_long err; - char buf[256]; - while ((err = ERR_get_error()) != 0) { - ERR_error_string_n(err, buf, sizeof(buf)); - error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf); - } - return 1; - } + if (r->ssl.conn && r->ssl.flags == NETDATA_SSL_HANDSHAKE_COMPLETE) + return (int)netdata_ssl_read(r->ssl.conn, buffer, size); #endif - if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp)) - return 1; - r->read_len = strlen(r->read_buffer); - worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, r->read_len); - return 0; -} -#else -/* - * The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing. - * if SSL encryption is on, then use SSL API for reading stream data. - * Use line oriented fgets() in buffer from receiver_state is provided. - * In other cases use fread to read binary data from socket. - * Return zero on success and the number of bytes were read using pointer in the last argument. - */ -static int read_stream(struct receiver_state *r, FILE *fp, char* buffer, size_t size, int* ret) { - if (!ret) - return 1; - *ret = 0; -#ifdef ENABLE_HTTPS - if (r->ssl.conn && !r->ssl.flags) { - ERR_clear_error(); - if (buffer != r->read_buffer + r->read_len) { - *ret = SSL_read(r->ssl.conn, buffer, size); - if (*ret > 0 ) - return 0; - } else { - // we need to receive data with LF to parse compression header - size_t ofs = 0; - int res = 0; - errno = 0; - while (ofs < size) { - do { - res = SSL_read(r->ssl.conn, buffer + ofs, 1); - // When either SSL_ERROR_SYSCALL (OpenSSL < 3.0) or SSL_ERROR_SSL(OpenSSL > 3.0) happens, - // the connection was lost https://www.openssl.org/docs/man3.0/man3/SSL_get_error.html, - // without the test we will have an infinite loop https://github.com/netdata/netdata/issues/13092 - int local_ssl_err = SSL_get_error(r->ssl.conn, res); - if (local_ssl_err == SSL_ERROR_SYSCALL || local_ssl_err == SSL_ERROR_SSL) { - error("The SSL connection has error SSL_ERROR_SYSCALL(%d) and system is registering errno = %d", - local_ssl_err, errno); - return 1; - } - } while (res == 0); - - if (res < 0) - break; - if (buffer[ofs] == '\n') - break; - ofs += res; - } - if (res > 0) { - ofs += res; - *ret = ofs; - buffer[ofs] = 0; - return 0; - } - } - // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket - u_long err; - char buf[256]; - while ((err = ERR_get_error()) != 0) { - ERR_error_string_n(err, buf, sizeof(buf)); - error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf); - } - return 1; + + ssize_t bytes_read = read(r->fd, buffer, size); + if(bytes_read == 0 && (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINPROGRESS)) { + error("STREAM: %s(): timeout while waiting for data on socket!", __FUNCTION__); + bytes_read = -3; } -#endif - if (buffer != r->read_buffer + r->read_len) { - // read to external buffer - *ret = fread(buffer, 1, size, fp); - if (!*ret) - return 1; - } else { - if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp)) - return 1; - *ret = strlen(r->read_buffer); + else if (bytes_read == 0) { + error("STREAM: %s(): EOF while reading data from socket!", __FUNCTION__); + bytes_read = -1; } - return 0; + else if (bytes_read < 0) { + error("STREAM: %s() failed to read from socket!", __FUNCTION__); + bytes_read = -2; + } + +// do { +// bytes_read = (int) fread(buffer, 1, size, fp); +// if (unlikely(bytes_read <= 0)) { +// if(feof(fp)) { +// internal_error(true, "%s(): fread() failed with EOF", __FUNCTION__); +// bytes_read = -2; +// } +// else if(ferror(fp)) { +// internal_error(true, "%s(): fread() failed with ERROR", __FUNCTION__); +// bytes_read = -3; +// } +// else bytes_read = 0; +// } +// else +// worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, bytes_read); +// } while(bytes_read == 0); + + return (int)bytes_read; } -/* - * Get the next line of data for parsing. - * Return data from the decompressor buffer if available. - * Otherwise read next line from the socket and check for compression header. - * Return the line was read If no compression header was found. - * Otherwise read the entire block of compressed data, decompress it - * and return it in receiver_state buffer. - * Return zero on success. - */ -static int receiver_read(struct receiver_state *r, FILE *fp) { - // check any decompressed data present - if (r->decompressor && r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) { - size_t available = sizeof(r->read_buffer) - r->read_len; +static bool receiver_read_uncompressed(struct receiver_state *r) { +#ifdef NETDATA_INTERNAL_CHECKS + if(r->read_buffer[r->read_len] != '\0') + fatal("%s(): read_buffer does not start with zero", __FUNCTION__ ); +#endif + + int bytes_read = read_stream(r, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1); + if(unlikely(bytes_read <= 0)) + return false; + + worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes_read); + worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_read); + + r->read_len += bytes_read; + r->read_buffer[r->read_len] = '\0'; + + return true; +} + +#ifdef ENABLE_COMPRESSION +static bool receiver_read_compressed(struct receiver_state *r) { + +#ifdef NETDATA_INTERNAL_CHECKS + if(r->read_buffer[r->read_len] != '\0') + fatal("%s: read_buffer does not start with zero #2", __FUNCTION__ ); +#endif + + // first use any available uncompressed data + if (r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) { + size_t available = sizeof(r->read_buffer) - r->read_len - 1; if (available) { size_t len = r->decompressor->get(r->decompressor, r->read_buffer + r->read_len, available); if (!len) { - internal_error(true, "decompressor returned zero length"); - return 1; + internal_error(true, "decompressor returned zero length #1"); + return false; } - r->read_len += len; + r->read_len += (int)len; + r->read_buffer[r->read_len] = '\0'; } - return 0; + else + internal_error(true, "The line to read is too big! Already have %d bytes in read_buffer.", r->read_len); + + return true; } - int ret = 0; - if (read_stream(r, fp, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1, &ret)) { - internal_error(true, "read_stream() failed (1)."); - return 1; + // no decompressed data available + // read the compression signature of the next block + + if(unlikely(r->read_len + r->decompressor->signature_size > sizeof(r->read_buffer) - 1)) { + internal_error(true, "The last incomplete line does not leave enough room for the next compression header! Already have %d bytes in read_buffer.", r->read_len); + return false; } - worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret); + // read the compression signature from the stream + // we have to do a loop here, because read_stream() may return less than the data we need + int bytes_read = 0; + do { + int ret = read_stream(r, r->read_buffer + r->read_len + bytes_read, r->decompressor->signature_size - bytes_read); + if (unlikely(ret <= 0)) + return false; - if (!is_compressed_data(r->read_buffer, ret)) { - r->read_len += ret; - return 0; + bytes_read += ret; + } while(unlikely(bytes_read < (int)r->decompressor->signature_size)); + + worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes_read); + + if(unlikely(bytes_read != (int)r->decompressor->signature_size)) + fatal("read %d bytes, but expected compression signature of size %zu", bytes_read, r->decompressor->signature_size); + + size_t compressed_message_size = r->decompressor->start(r->decompressor, r->read_buffer + r->read_len, bytes_read); + if (unlikely(!compressed_message_size)) { + internal_error(true, "multiplexed uncompressed data in compressed stream!"); + r->read_len += bytes_read; + r->read_buffer[r->read_len] = '\0'; + return true; + } + + if(unlikely(compressed_message_size > COMPRESSION_MAX_MSG_SIZE)) { + error("received a compressed message of %zu bytes, which is bigger than the max compressed message size supported of %zu. Ignoring message.", + compressed_message_size, (size_t)COMPRESSION_MAX_MSG_SIZE); + return false; } - if (unlikely(!r->decompressor)) - r->decompressor = create_decompressor(); - - size_t bytes_to_read = r->decompressor->start(r->decompressor, r->read_buffer, ret); + // delete compression header from our read buffer + r->read_buffer[r->read_len] = '\0'; - // Read the entire block of compressed data because - // we're unable to decompress incomplete block - char compressed[bytes_to_read]; + // Read the entire compressed block of compressed data + char compressed[compressed_message_size]; + size_t compressed_bytes_read = 0; do { - if (read_stream(r, fp, compressed, bytes_to_read, &ret)) { - internal_error(true, "read_stream() failed (2)."); - return 1; + size_t start = compressed_bytes_read; + size_t remaining = compressed_message_size - start; + + int last_read_bytes = read_stream(r, &compressed[start], remaining); + if (unlikely(last_read_bytes <= 0)) { + internal_error(true, "read_stream() failed #2, with code %d", last_read_bytes); + return false; } - worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, ret); + compressed_bytes_read += last_read_bytes; - // Send input data to decompressor - if (ret) - r->decompressor->put(r->decompressor, compressed, ret); + } while(unlikely(compressed_message_size > compressed_bytes_read)); - bytes_to_read -= ret; - } while (bytes_to_read > 0); + worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)compressed_bytes_read); - // Decompress - size_t bytes_to_parse = r->decompressor->decompress(r->decompressor); + // decompress the compressed block + size_t bytes_to_parse = r->decompressor->decompress(r->decompressor, compressed, compressed_bytes_read); if (!bytes_to_parse) { internal_error(true, "no bytes to parse."); - return 1; + return false; } - // Fill read buffer with decompressed data - r->read_len = r->decompressor->get(r->decompressor, r->read_buffer, sizeof(r->read_buffer)); - return 0; -} + worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes_to_parse); -#endif + // fill read buffer with decompressed data + size_t len = (int)r->decompressor->get(r->decompressor, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1); + if (!len) { + internal_error(true, "decompressor returned zero length #2"); + return false; + } + r->read_len += (int)len; + r->read_buffer[r->read_len] = '\0'; + + return true; +} +#else // !ENABLE_COMPRESSION +static bool receiver_read_compressed(struct receiver_state *r) { + return receiver_read_uncompressed(r); +} +#endif // ENABLE_COMPRESSION /* Produce a full line if one exists, statefully return where we start next time. * When we hit the end of the buffer with a partial line move it to the beginning for the next fill. @@ -302,7 +304,10 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b char *de = &buffer[buffer_length - 2]; if(ss >= se) { + *ds = '\0'; + *pos = 0; r->read_len = 0; + r->read_buffer[r->read_len] = '\0'; return NULL; } @@ -333,6 +338,9 @@ static char *receiver_next_line(struct receiver_state *r, char *buffer, size_t b // move everything to the beginning memmove(r->read_buffer, &r->read_buffer[start], r->read_len - start); r->read_len -= (int)start; + r->read_buffer[r->read_len] = '\0'; + *ds = '\0'; + *pos = 0; return NULL; } @@ -342,7 +350,7 @@ static void streaming_parser_thread_cleanup(void *ptr) { parser_destroy(parser); } -static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp_in, FILE *fp_out, void *ssl) { +static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, int fd, void *ssl) { size_t result; PARSER_USER_OBJECT user = { @@ -353,7 +361,7 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, F .trust_durations = 1 }; - PARSER *parser = parser_init(rpt->host, &user, fp_in, fp_out, PARSER_INPUT_SPLIT, ssl); + PARSER *parser = parser_init(rpt->host, &user, NULL, NULL, fd, PARSER_INPUT_SPLIT, ssl); rrd_collector_started(); @@ -365,36 +373,56 @@ static size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, F user.parser = parser; + bool compressed_connection = false; #ifdef ENABLE_COMPRESSION - if (rpt->decompressor) - rpt->decompressor->reset(rpt->decompressor); + if(stream_has_capability(rpt, STREAM_CAP_COMPRESSION)) { + compressed_connection = true; + + if (!rpt->decompressor) + rpt->decompressor = create_decompressor(); + else + rpt->decompressor->reset(rpt->decompressor); + } #endif - char buffer[PLUGINSD_LINE_MAX + 2]; - do { - if(receiver_read(rpt, fp_in)) break; + rpt->read_buffer[0] = '\0'; + rpt->read_len = 0; - size_t pos = 0; - while(receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &pos)) { - if(unlikely(netdata_exit)) { - internal_error(true, "exiting..."); - goto done; - } - if(unlikely(rpt->shutdown)) { - internal_error(true, "parser shutdown..."); - goto done; - } - if (unlikely(parser_action(parser, buffer))) { - internal_error(true, "parser_action() failed..."); - goto done; - } + size_t read_buffer_start = 0; + char buffer[PLUGINSD_LINE_MAX + 2] = ""; + while(!netdata_exit) { + if(!receiver_next_line(rpt, buffer, PLUGINSD_LINE_MAX + 2, &read_buffer_start)) { + bool have_new_data; + if(compressed_connection) + have_new_data = receiver_read_compressed(rpt); + else + have_new_data = receiver_read_uncompressed(rpt); + + if(!have_new_data) + break; + + rpt->last_msg_t = now_realtime_sec(); + continue; } - rpt->last_msg_t = now_realtime_sec(); + if(unlikely(netdata_exit)) { + internal_error(true, "exiting..."); + goto done; + } + if(unlikely(rpt->shutdown)) { + internal_error(true, "parser shutdown..."); + goto done; + } + + if (unlikely(parser_action(parser, buffer))) { + internal_error(true, "parser_action() failed on keyword '%s'.", buffer); + break; + } } - while(!netdata_exit); done: + internal_error(true, "Streaming receiver thread stopping..."); + result = user.count; // free parser with the pop function @@ -644,42 +672,11 @@ static int rrdpush_receive(struct receiver_state *rpt) error("STREAM %s [receive from [%s]:%s]: cannot remove the non-blocking flag from socket %d", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd); struct timeval timeout; - timeout.tv_sec = 120; + timeout.tv_sec = 600; timeout.tv_usec = 0; if (unlikely(setsockopt(rpt->fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout) != 0)) error("STREAM %s [receive from [%s]:%s]: cannot set timeout for socket %d", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd); - // convert the socket to a FILE * - // It seems that the same FILE * cannot be used for both reading and writing. - // (reads and writes seem to interfere with each other, with undefined results). - - int fd_in = rpt->fd; - int fd_out = fcntl(rpt->fd, F_DUPFD_CLOEXEC, 0); - if(fd_out == -1) { - error("STREAM %s [receive from [%s]:%s]: failed to duplicate FD %d.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd); - log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rrdhost_hostname(rpt->host), "FAILED - SOCKET ERROR"); - close(fd_in); - return 0; - } - - FILE *fp_out = fdopen(fd_out, "w"); - if(!fp_out) { - error("STREAM %s [receive from [%s]:%s]: failed to get a FILE pointer for fd_out %d.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd); - log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rrdhost_hostname(rpt->host), "FAILED - SOCKET ERROR"); - close(fd_in); - close(fd_out); - return 0; - } - - FILE *fp_in = fdopen(fd_in, "r"); - if(!fp_in) { - error("STREAM %s [receive from [%s]:%s]: failed to get a FILE pointer for fd_in %d.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->fd); - log_stream_connection(rpt->client_ip, rpt->client_port, rpt->key, rpt->host->machine_guid, rrdhost_hostname(rpt->host), "FAILED - SOCKET ERROR"); - close(fd_in); - fclose(fp_out); - |