summaryrefslogtreecommitdiffstats
path: root/streaming/rrdpush.h
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2022-01-26 18:05:15 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2022-01-26 18:05:42 +0000
commit112b5b91647c3dea45cc1c9bc364df526c8012f1 (patch)
tree450af925135ec664c4310a1eb28b69481094ee2a /streaming/rrdpush.h
parentReleasing debian version 1.32.1-2. (diff)
downloadnetdata-112b5b91647c3dea45cc1c9bc364df526c8012f1.tar.xz
netdata-112b5b91647c3dea45cc1c9bc364df526c8012f1.zip
Merging upstream version 1.33.0.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'streaming/rrdpush.h')
-rw-r--r--streaming/rrdpush.h69
1 files changed, 62 insertions, 7 deletions
diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index 027ccd102..937ead6fa 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -10,10 +10,16 @@
#define CONNECTED_TO_SIZE 100
-#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)4
#define STREAM_VERSION_CLAIM 3
#define STREAM_VERSION_CLABELS 4
-#define VERSION_GAP_FILLING 5
+#define STREAM_VERSION_COMPRESSION 5
+#define VERSION_GAP_FILLING 6
+
+#ifdef ENABLE_COMPRESSION
+#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)(STREAM_VERSION_COMPRESSION)
+#else
+#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)(STREAM_VERSION_CLABELS)
+#endif //ENABLE_COMPRESSION
#define STREAMING_PROTOCOL_VERSION "1.1"
#define START_STREAMING_PROMPT "Hit me baby, push them over..."
@@ -35,6 +41,38 @@ typedef struct {
char *kernel_version;
} stream_encoded_t;
+#ifdef ENABLE_COMPRESSION
+struct compressor_state {
+ char *buffer;
+ size_t buffer_size;
+ struct compressor_data *data; // Compression API specific data
+ void (*reset)(struct compressor_state *state);
+ size_t (*compress)(struct compressor_state *state, const char *data, size_t size, char **buffer);
+ void (*destroy)(struct compressor_state **state);
+};
+
+struct decompressor_state {
+ char *buffer;
+ size_t buffer_size;
+ size_t buffer_len;
+ size_t buffer_pos;
+ char *out_buffer;
+ size_t out_buffer_len;
+ size_t out_buffer_pos;
+ size_t total_compressed;
+ size_t total_uncompressed;
+ size_t packet_count;
+ struct decompressor_data *data; // Deompression API specific data
+ void (*reset)(struct decompressor_state *state);
+ size_t (*start)(struct decompressor_state *state, const char *header, size_t header_size);
+ size_t (*put)(struct decompressor_state *state, const char *data, size_t size);
+ size_t (*decompress)(struct decompressor_state *state);
+ size_t (*decompressed_bytes_in_buffer)(struct decompressor_state *state);
+ size_t (*get)(struct decompressor_state *state, char *data, size_t size);
+ void (*destroy)(struct decompressor_state **state);
+};
+#endif
+
// Thread-local storage
// Metric transmission: collector threads asynchronously fill the buffer, sender thread uses it.
@@ -60,6 +98,10 @@ struct sender_state {
char read_buffer[512];
int read_len;
int32_t version;
+#ifdef ENABLE_COMPRESSION
+ unsigned int rrdpush_compression;
+ struct compressor_state *compressor;
+#endif
};
struct receiver_state {
@@ -75,9 +117,9 @@ struct receiver_state {
char *abbrev_timezone;
int32_t utc_offset;
char *tags;
- char *client_ip; // Duplicated in pluginsd
- char *client_port; // Duplicated in pluginsd
- char *program_name; // Duplicated in pluginsd
+ char *client_ip; // Duplicated in pluginsd
+ char *client_port; // Duplicated in pluginsd
+ char *program_name; // Duplicated in pluginsd
char *program_version;
struct rrdhost_system_info *system_info;
int update_every;
@@ -85,15 +127,22 @@ struct receiver_state {
time_t last_msg_t;
char read_buffer[1024]; // Need to allow RRD_ID_LENGTH_MAX * 4 + the other fields
int read_len;
+ unsigned int shutdown:1; // Tell the thread to exit
+ unsigned int exited; // Indicates that the thread has exited (NOT A BITFIELD!)
#ifdef ENABLE_HTTPS
struct netdata_ssl ssl;
#endif
- unsigned int shutdown:1; // Tell the thread to exit
- unsigned int exited; // Indicates that the thread has exited (NOT A BITFIELD!)
+#ifdef ENABLE_COMPRESSION
+ unsigned int rrdpush_compression;
+ struct decompressor_state *decompressor;
+#endif
};
extern unsigned int default_rrdpush_enabled;
+#ifdef ENABLE_COMPRESSION
+extern unsigned int default_compression_enabled;
+#endif
extern char *default_rrdpush_destination;
extern char *default_rrdpush_api_key;
extern char *default_rrdpush_send_charts_matching;
@@ -116,4 +165,10 @@ extern void rrdpush_sender_thread_stop(RRDHOST *host);
extern void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, RRDVAR *rv);
extern void log_stream_connection(const char *client_ip, const char *client_port, const char *api_key, const char *machine_guid, const char *host, const char *msg);
+#ifdef ENABLE_COMPRESSION
+struct compressor_state *create_compressor();
+struct decompressor_state *create_decompressor();
+size_t is_compressed_data(const char *data, size_t data_size);
+#endif
+
#endif //NETDATA_RRDPUSH_H