Add code for LZ4 streaming data compression (#11821)
* Add code for LZ4 streaming data compression * Fix LGTM alert * Add lz4 library for link when compression enabled * Add LZ4_resetStream_fast presence detection * Disable compression for older LZ4 libraries * Correct LZ4 API check * [Testing Stream Compression] Debug msgs and report.md * Add LZ4 library version using LZ4_initStream * Fixed bug in SSL mode * [Testing compression] - Add compression info messages * Set compression enabled by default, update doc * Update streaming/README.md Co-authored-by: DShreve2 <david@netdata.cloud> * [Agent Negotiation] Compression as separate capability * [Agent Negotiation] Compression as separate capability - default compression variable always active * Add code to negotiate compression * [Agent Negotiation] Based on stream version * [Agent Negotiation] Version based - fix compilation error * [Agent Negotiation] Fix glob var default_compression_enbaled=0 affects all the connections - Handle compression - stream version based * [Agent Negotiation - Compression] - Add control flag in 1. sender/receiver state & 2. stream.conf per child * [Agent Negotiation - Compression] Fix stream.conf key, mguid control * [Agent Negotiate Compression] Fine control on stream.conf per key,mguid for each child * [Agent Negotiation Compression] Stop destroying compressor for runtime configuration + Update Readme.md * [Agent Negotiation Compression] Use stream_version 4 if compression is disabled * Correct child's compression check * [Agent Negotiation Compression] Create streaming compression section in docs. * [Agent Negotiation Compresion] Remove redundant debug msgs * [Stream Compression] - integrate compression build info & config info in api/v1/info endpoint. * [Agent Negotiation] Finalize README.md * [Agent Stream Compression] Fix buildinfo json, Finalize readme.md * [Agent Stream Compression] Negotiate compression based on stream version * [Agent Stream Compression] Stream compression control per child in stream.conf | per AP_KEY, MACHINE_GUID * [Agent Stream Compression] Avoid destroying compressor enabling runtime configuration + Update Readme.md * [Agent Stream Compression] - Provide stream compression build info & config info in api/v1/info endpoint + Update Readme.md * [Agent Stream Compression] Fix rebase conflicts * [Agent Stream Compression] Fix more rebase conflicts * [Agent Stream Compression] 1. Stream version based negotiation 2. per child stream.conf control 3. finalize docs 4. stream compression build info in web api * [Agent Stream Compression] 1. Stream version based negotiation 2. per child stream.conf control 3. finalize docs 4. stream compression build info in web api * [Agent Stream Compression] Change unsuccessful buffer check to error * [Agent Stream Compression] Readme.md proof-read corrections, downgrade to stream_version_clabels, add shields for supported versions, EOF lint * [Agent Stream Compression] Fix missed lz4 library on Alpine Linux * Phrasal review Co-authored-by: odynik <odynik.ee@gmail.com> Co-authored-by: DShreve2 <david@netdata.cloud> Co-authored-by: Tina Lüdtke <tina@kickoke.com>
avstrakhov committed
Jan 19, 2022 at 18:57 UTC
b003e5fd40c3b42dfacc87db5a7730b76eff0e92
13 files changed
+883
-103
CMakeLists.txt
+1
@@ -740,6 +740,7 @@ set(API_PLUGIN_FILES
740
set(STREAMING_PLUGIN_FILES
741
streaming/rrdpush.c
742
streaming/rrdpush.h
743
+ streaming/compression.c
744
streaming/receiver.c
745
streaming/sender.c
746
)
Makefile.am
+1
@@ -543,6 +543,7 @@ API_PLUGIN_FILES = \
543
544
STREAMING_PLUGIN_FILES = \
545
streaming/rrdpush.c \
546
+ streaming/compression.c \
547
streaming/sender.c \
548
streaming/receiver.c \
549
streaming/rrdpush.h \
configure.ac
+43
-1
@@ -154,6 +154,12 @@ AC_ARG_ENABLE(
154
,
155
[enable_https="detect"]
156
)
157
+AC_ARG_ENABLE(
158
+ [compression],
159
+ [AS_HELP_STRING([--enable-compression], [Enable LZ4 compression support @<:@default autodetect@:>@])],
160
+ ,
161
+ [enable_compression="detect"]
162
+)
163
AC_ARG_ENABLE(
164
[dbengine],
165
[AS_HELP_STRING([--disable-dbengine], [disable netdata dbengine @<:@default autodetect@:>@])],
@@ -355,13 +361,18 @@ OPTIONAL_UV_LIBS="${UV_LIBS}"
361
# -----------------------------------------------------------------------------
362
# lz4 Extremely Fast Compression algorithm
363
364
+AC_CHECK_LIB(
365
+ [lz4],
366
+ [LZ4_initStream],
367
+ [LZ4_LIBS_FAST="-llz4"]
368
+)
369
+
370
AC_CHECK_LIB(
371
[lz4],
372
[LZ4_compress_default],
373
[LZ4_LIBS="-llz4"]
374
)
375
364
-
376
# -----------------------------------------------------------------------------
377
# zlib
378
@@ -517,6 +528,37 @@ fi
528
AC_MSG_RESULT([${enable_https}])
529
AM_CONDITIONAL([ENABLE_HTTPS], [test "${enable_https}" = "yes"])
530
531
+AC_MSG_CHECKING([if netdata compression should be used])
532
+if test "${enable_compression}" != "no"; then
533
+ if test "${LZ4_LIBS_FAST}"; then
534
+ LIBS_BKP="${LIBS}"
535
+ LIBS="${LZ4_LIBS_FAST}"
536
+ AC_TRY_LINK(
537
+ [ #include <lz4.h> ],
538
+ [
539
+ LZ4_stream_t* stream = LZ4_initStream(NULL, 0);
540
+ ],
541
+ [ enable_compression="yes"],
542
+ [ enable_compression="no" ]
543
+ )
544
+ LIBS="${LIBS_BKP}"
545
+ if test "${enable_compression}" == "yes"; then
546
+ OPTIONAL_LZ4_LIBS="${LZ4_LIBS_FAST}"
547
+ AC_DEFINE([ENABLE_COMPRESSION], [1], [netdata compression usability])
548
+ fi
549
+ else
550
+ if test "${enable_compression}" == "yes"; then
551
+ AC_MSG_ERROR([liblz4 with version >= 1.9.0 required to enable_compression. Try installing the required version of 'liblz4-dev' or 'liblz4-devel'.])
552
+ enable_compression="no"
553
+ fi
554
+ enable_compression="no"
555
+ fi
556
+else
557
+ enable_compression="no"
558
+fi
559
+AC_MSG_RESULT([${enable_compression}])
560
+AM_CONDITIONAL([ENABLE_COMPRESSION], [test "${enable_compression}" = "yes"])
561
+
562
# -----------------------------------------------------------------------------
563
# JSON-C
564
daemon/buildinfo.c
+10
@@ -49,6 +49,13 @@
49
#define FEAT_ML 0
50
#endif
51
52
+#ifdef ENABLE_COMPRESSION
53
+#define FEAT_STREAM_COMPRESSION 1
54
+#else
55
+#define FEAT_STREAM_COMPRESSION 0
56
+#endif //ENABLE_COMPRESSION
57
+
58
+
59
// Optional libraries
60
61
#ifdef HAVE_PROTOBUF
@@ -212,6 +219,7 @@ void print_build_info(void) {
219
printf(" ACLK Legacy: %s\n", FEAT_YES_NO(0));
220
printf(" TLS Host Verification: %s\n", FEAT_YES_NO(FEAT_TLS_HOST_VERIFY));
221
printf(" Machine Learning: %s\n", FEAT_YES_NO(FEAT_ML));
222
+ printf(" Stream Compression: %s\n", FEAT_YES_NO(FEAT_STREAM_COMPRESSION));
223
224
printf("Libraries:\n");
225
printf(" protobuf: %s%s\n", FEAT_YES_NO(FEAT_PROTOBUF), FEAT_PROTOBUF_BUNDLED);
@@ -265,6 +273,7 @@ void print_build_info_json(void) {
273
274
printf(" \"tls-host-verify\": %s,\n", FEAT_JSON_BOOL(FEAT_TLS_HOST_VERIFY));
275
printf(" \"machine-learning\": %s\n", FEAT_JSON_BOOL(FEAT_ML));
276
+ printf(" \"stream-compression\": %s\n", FEAT_JSON_BOOL(FEAT_STREAM_COMPRESSION));
277
printf(" },\n");
278
279
printf(" \"libs\": {\n");
@@ -311,6 +320,7 @@ void analytics_build_info(BUFFER *b) {
320
if(NEW_CLOUD_PROTO) buffer_strcat (b, "|New Cloud Protocol Support");
321
if(FEAT_TLS_HOST_VERIFY) buffer_strcat (b, "|TLS Host Verification");
322
if(FEAT_ML) buffer_strcat (b, "|Machine Learning");
323
+ if(FEAT_STREAM_COMPRESSION) buffer_strcat (b, "|Stream Compression");
324
325
if(FEAT_PROTOBUF) buffer_strcat (b, "|protobuf");
326
if(FEAT_JEMALLOC) buffer_strcat (b, "|jemalloc");
database/rrdhost.c
+4
@@ -851,6 +851,10 @@ void rrdhost_free(RRDHOST *host) {
851
rrdpush_sender_thread_stop(host); // stop a possibly running thread
852
cbuffer_free(host->sender->buffer);
853
buffer_free(host->sender->build);
854
+#ifdef ENABLE_COMPRESSION
855
+ if (host->sender->compressor)
856
+ host->sender->compressor->destroy(&host->sender->compressor);
857
+#endif
858
freez(host->sender);
859
host->sender = NULL;
860
if (netdata_exit) {
streaming/README.md
+159
-60
@@ -1,89 +1,90 @@
1
-<!--
1
+---
2
title: "Streaming and replication"
3
description: "Replicate and mirror Netdata's metrics through real-time streaming from child to parent nodes. Then combine, correlate, and export."
4
custom_edit_url: https://github.com/netdata/netdata/edit/master/streaming/README.md
5
--->
5
+---
6
7
-# Streaming and replication
7
9
-Each Netdata is able to replicate/mirror its database to another Netdata, by streaming the collected
10
-metrics in real-time to it. This is quite different to [data archiving to third party time-series
8
+Each Netdata node is able to replicate/mirror its database to another Netdata node, by streaming the collected
9
+metrics in real-time. This is quite different to [data archiving to third party time-series
10
databases](/exporting/README.md).
11
+The nodes that send metrics are called **child** nodes, and the nodes that receive metrics are called **parent** nodes.
12
+
13
+There are also **proxy** nodes, which collect metrics from a child and sends it to a parent.
14
13
-When Netdata streams metrics to another Netdata, the receiving one is able to perform everything a Netdata instance is
14
-capable of. This includes the following:
15
+When one Netdata node streams metrics another, the receiving instance can use the data for all features of a typical Netdata node, for example:
16
17
- Visualize metrics with a dashboard
18
- Run health checks that trigger alarms and send alarm notifications
19
- Export metrics to an external time-series database
20
20
-The nodes that send metrics are called **child** nodes, and the nodes that receive metrics are called **parent** nodes.
21
-There are also **proxy** nodes, which collects metrics from a child and sends it to a parent.
21
+
22
+
23
24
## Supported configurations
25
26
### Netdata without a database or web API (headless collector)
27
27
-Local Netdata (child), **without any database or alarms**, collects metrics and sends them to another Netdata
28
+A local Netdata Agent (child), **without any database or alarms**, collects metrics and sends them to another Netdata node
29
(parent).
30
+The same parent can collect data for any number of child nodes and serves alerts for each child.
31
32
The node menu shows a list of all "databases streamed to" the parent. Clicking one of those links allows the user to
33
view the full dashboard of the child node. The URL has the form
34
`http://parent-host:parent-port/host/child-host/`.
35
34
-Alarms for the child are served by the parent.
36
36
-In this mode the child is just a plain data collector. It spawns all external plugins, but instead of maintaining a
37
-local database and accepting dashboard requests, it streams all metrics to the parent. The memory footprint is reduced
38
-significantly, to between 6 MiB and 40 MiB, depending on the enabled plugins. To reduce the memory usage as much as
37
+In a headless setup, the child acts as a plain data collector. It spawns all external plugins, but instead of maintaining a
38
+local database and accepting dashboard requests, it streams all metrics to the parent.
39
+
40
+This setup works great to reduce the memory footprint. Depending on the enabled plugins, memory usage is between 6 MiB and 40 MiB. To reduce the memory usage as much as
41
possible, refer to the [performance optimization guide](/docs/guides/configure/performance.md).
42
41
-The same parent can collect data for any number of child nodes.
43
44
### Database Replication
45
45
-Local Netdata (child), **with a local database (and possibly alarms)**, collects metrics and
46
-sends them to another Netdata (parent).
46
+The local Netdata Agent (child), **with a local database (and possibly alarms)**, collects metrics and
47
+sends them to another Netdata node (parent).
48
49
The user can use all the functions **at both** `http://child-ip:child-port/` and
50
`http://parent-host:parent-port/host/child-host/`.
51
52
The child and the parent may have different data retention policies for the same metrics.
53
53
-Alarms for the child are triggered by **both** the child and the parent (and actually
54
-each can have different alarms configurations or have alarms disabled).
54
+Alerts for the child are triggered by **both** the child and the parent.
55
+It is possible to enable different alert configurations on the parent and the child.
56
56
-Take a note, that custom chart names, configured on the child, should be in the form `type.name` to work correctly. The parent will truncate the `type` part and substitute the original chart `type` to store the name in the database.
57
+In order for custom chart names on the child to work correctly, follow the form `type.name`. The parent will truncate the `type` part and substitute the original chart `type` to store the name in the database.
58
59
### Netdata proxies
60
60
-Local Netdata (child), with or without a database, collects metrics and sends them to another
61
-Netdata (**proxy**), which may or may not maintain a database, which forwards them to another
61
+The local Netdata Agent(child), with or without a database, collects metrics and sends them to another
62
+Netdata node(**proxy**), which may or may not maintain a database, which forwards them to another
63
Netdata (parent).
64
64
-Alarms for the child can be triggered by any of the involved hosts that maintains a database.
65
+Alerts for the child can be triggered by any of the involved hosts that maintains a database.
66
66
-Any number of daisy chaining Netdata servers are supported, each with or without a database and
67
-with or without alarms for the child metrics.
67
+You can daisy-chain any number of Netdata, each with or without a database and
68
+with or without alerts for the child metrics.
69
69
-### mix and match with backends
70
+### Mix and match with backends
71
72
All nodes that maintain a database can also send their data to a backend database.
73
This allows quite complex setups.
74
75
Example:
76
76
-1. Netdata `A`, `B` do not maintain a database and stream metrics to Netdata `C`(live streaming functionality, i.e. this PR)
77
-2. Netdata `C` maintains a database for `A`, `B`, `C` and archives all metrics to `graphite` with 10 second detail (backends functionality)
78
-3. Netdata `C` also streams data for `A`, `B`, `C` to Netdata `D`, which also collects data from `E`, `F` and `G` from another DMZ (live streaming functionality, i.e. this PR)
79
-4. Netdata `D` is just a proxy, without a database, that streams all data to a remote site at Netdata `H`
80
-5. Netdata `H` maintains a database for `A`, `B`, `C`, `D`, `E`, `F`, `G`, `H` and sends all data to `opentsdb` with 5 seconds detail (backends functionality)
81
-6. alarms are triggered by `H` for all hosts
82
-7. users can use all the Netdata that maintain a database to view metrics (i.e. at `H` all hosts can be viewed).
77
+1. Netdata nodes `A` and `B` do not maintain a database and stream metrics to Netdata node `C`(live streaming functionality).
78
+2. Netdata node `C` maintains a database for `A`, `B`, `C` and archives all metrics to `graphite` with 10 second detail (backends functionality).
79
+3. Netdata node `C` also streams data for `A`, `B`, `C` to Netdata `D`, which also collects data from `E`, `F` and `G` from another DMZ (live streaming functionality).
80
+4. Netdata node `D` is just a proxy, without a database, that streams all data to a remote site at Netdata `H`.
81
+5. Netdata node `H` maintains a database for `A`, `B`, `C`, `D`, `E`, `F`, `G`, `H` and sends all data to `opentsdb` with 5 seconds detail (backends functionality)
82
+6. Alerts are triggered by `H` for all hosts.
83
+7. Users can use all Netdata nodes that maintain a database to view metrics (i.e. at `H` all hosts can be viewed).
84
85
## Configuration
86
86
-These are options that affect the operation of Netdata in this area:
87
+The following options affect how Netdata streams:
88
89
```
90
[global]
@@ -91,7 +92,7 @@ These are options that affect the operation of Netdata in this area:
92
```
93
94
`[global].memory mode = none` disables the database at this host. This also disables health
94
-monitoring (there cannot be health monitoring without a database).
95
+monitoring because a node can't have health monitoring without a database.
96
97
```
98
[web]
@@ -117,22 +118,27 @@ entries "... too busy to accept new streaming request. Will be allowed in X secs
118
`[backend]` configures data archiving to a backend (it archives all databases maintained on
119
this host).
120
120
-### streaming configuration
121
+### Streaming configuration
122
+
123
+The new file `stream.conf` contains streaming configuration for a sending and a receiving Netdata node.
124
+
125
+To configure streaming on your system:
126
+1. Generate an API key using `uuidgen`. Note: API keys are just random GUIDs. You can use the same API key on all your Netdata, or use a different API key for any pair of sending-receiving Netdata nodes.
127
122
-A new file is introduced: `stream.conf` (to edit it on your system run
123
-`/etc/netdata/edit-config stream.conf`). This file holds streaming configuration for both the
124
-sending and the receiving Netdata.
128
+2. Authorize the communication between a pair of sending-receiving Netdata nodes using the generated API key.
129
+Once the communication is authorized, the sending Netdata node can push metrics for any number of hosts.
130
126
-API keys are used to authorize the communication of a pair of sending-receiving Netdata.
127
-Once the communication is authorized, the sending Netdata can push metrics for any number of hosts.
131
+3. To edit `stream.conf`, run `/etc/netdata/edit-config stream.conf`
132
129
-You can generate an API key with the command `uuidgen`. API keys are just random GUIDs.
130
-You can use the same API key on all your Netdata, or use a different API key for any pair of
131
-sending-receiving Netdata.
133
+The following sections describe how you can configure sending and receiving Netdata nodes.
134
133
-##### options for the sending node
135
135
-This is the section for the sending Netdata. On the receiving node, `[stream].enabled` can be `no`.
136
+
137
+
138
+
139
+##### Options for the sending node
140
+
141
+This is the section for the sending Netdata node. On the receiving node, `[stream].enabled` can be `no`.
142
If it is `yes`, the receiving node will also stream the metrics to another node (i.e. it will be
143
a proxy).
144
@@ -141,8 +147,13 @@ a proxy).
147
enabled = yes | no
148
destination = IP:PORT[:SSL] ...
149
api key = XXXXXXXXXXX
144
-```
150
151
+[API_KEY]
152
+ enabled = yes | no
153
+
154
+[MACHINE_GUID]
155
+ enabled = yes | no
156
+```
157
This is an overview of how these options can be combined:
158
159
| target|memory<br/>mode|web<br/>mode|stream<br/>enabled|backend|alarms|dashboard|
@@ -154,9 +165,10 @@ This is an overview of how these options can be combined:
165
166
For the options to encrypt the data stream between the child and the parent, refer to [securing the communication](#securing-streaming-communications)
167
157
-##### options for the receiving node
168
159
-`stream.conf` looks like this:
169
+##### Options for the receiving node
170
+
171
+For a receiving Netdata node, the `stream.conf` looks like this:
172
173
```sh
174
# replace API_KEY with your uuidgen generated GUID
@@ -192,7 +204,7 @@ You can also use `default memory mode = dbengine` for an API key or `memory mode
204
a single host. The additional `page cache size` and `dbengine multihost disk space` configuration options
205
are inherited from the global Netdata configuration.
206
195
-##### allow from
207
+##### Allow from
208
209
`allow from` settings are [Netdata simple patterns](/libnetdata/simple_pattern/README.md): string matches
210
that use `*` as wildcard (any number of times) and a `!` prefix for a negative match.
@@ -201,7 +213,7 @@ important: left to right, the first positive or negative match is used.
213
214
`allow from` is available in Netdata v1.9+
215
204
-##### tracing
216
+##### Tracing
217
218
When a child is trying to push metrics to a parent or proxy, it logs entries like these:
219
@@ -229,7 +241,10 @@ For Netdata v1.9+, streaming can also be monitored via `access.log`.
241
242
### Securing streaming communications
243
232
-Netdata does not activate TLS encryption by default. To encrypt streaming connections, you first need to [enable TLS support](/web/server/README.md#enabling-tls-support) on the parent. With encryption enabled on the receiving side, you need to instruct the child to use TLS/SSL as well. On the child's `stream.conf`, configure the destination as follows:
244
+Netdata does not activate TLS encryption by default. To encrypt streaming connections:
245
+1. On the parent node (receiving node), [enable TLS support](/web/server/README.md#enabling-tls-support).
246
+2. On the child node (sending node), [enable TLS support](/web/server/README.md#enabling-tls-support).
247
+3. On the child's `stream.conf`, configure the destination as follows:
248
249
```
250
[stream]
@@ -316,6 +331,80 @@ With the introduction of TLS/SSL, the parent-child communication behaves as show
331
| Yes|-/force/optional|Yes|no|The parent-child stream is encrypted, provided that the parent has a valid TLS/SSL certificate. Otherwise, the child refuses to connect.|
332
| Yes|-/force/optional|Yes|yes|The parent-child stream is encrypted.|
333
334
+### Streaming compression
335
+<a href="https://github.com/netdata/netdata/releases/latest"><img src="https://img.shields.io/badge/Supported%20Netdata%20Agent-v1.33%2B-brightgreen" alt="Supported version Netdata agent release"></a>
336
+
337
+<a href="https://github.com/netdata/netdata/releases/latest"><img src="https://img.shields.io/badge/Supported%20Netdata%20stream%20version-v5%2B-blue" alt="Supported version Netdata agent release"></a>
338
+
339
+#### OS dependencies
340
+* Streaming compression is based on [lz4 v1.9.0+](https://github.com/lz4/lz4). The [lz4 v1.9.0+](https://github.com/lz4/lz4) library must be installed in your OS in order to enable streaming compression. Any lower version will disable Netdata streaming compression for compatibility purposes between the older versions of Netdata agents.
341
+
342
+To check if your Netdata Agent supports stream compression run the following GET request in your browser or terminal:
343
+
344
+```
345
+curl -X GET http://localhost:19999/api/v1/info | grep 'Stream Compression'
346
+```
347
+
348
+**Output**
349
+```
350
+"buildinfo": "dbengine|Native HTTPS|Netdata Cloud|ACLK Next Generation|New Cloud Protocol Support|ACLK Legacy|TLS Host Verification|Machine Learning|Stream Compression|protobuf|JSON-C|libcrypto|libm|LWS v3.2.2|mosquitto|zlib|apps|cgroup Network Tracking|EBPF|perf|slabinfo",
351
+```
352
+> Note: If your OS doesn't support Netdata compression the `buildinfo` will not contain the `Stream Compression` statement.
353
+
354
+To check if your Netdata Agent has stream compression enabled, run the following GET request in your browser or terminal:
355
+
356
+```
357
+ curl -X GET http://localhost:19999/api/v1/info | grep 'stream-compression'
358
+```
359
+**Output**
360
+```
361
+"stream-compression": "enabled"
362
+```
363
+Note: The `stream-compression` status can be `"enabled" | "disabled" | "N/A"`.
364
+
365
+Stream data compression is enabled by default on systems where LZ4 library v1.9.0+ is installed. A compressed data packet is determined and decompressed on the fly.
366
+
367
+#### Limitations
368
+The current implementation of streaming data compression has the limitation that the size of single data block transmitted must not exceed 16384 bytes. If single data block size exceeds this limit, stream data compression should be disabled.
369
+
370
+#### How to enable stream compression
371
+Netdata Agents are shipped with data compression enabled by default. You can also configure which streams will use compression.
372
+
373
+With enabled stream compression, a Netdata Agent can negotiate streaming compression with other Netdata Agents. During the negotiation of streaming compression both Netdata Agents should support and enable compression in order to communicate over a compressed stream. The negotiation will result into an uncompressed stream, if one of the Netdata Agents doesn't support **or** has compression disabled.
374
+
375
+To enable stream compression:
376
+
377
+1. Edit `stream.conf` by using the `edit-config` script:
378
+`/etc/netdata/edit-config stream.conf`.
379
+
380
+2. In the `[stream]` section, set `enable compression` to `yes`.
381
+```
382
+# This is the default stream compression flag for an agent.
383
+
384
+[stream]
385
+ enable compression = yes | no
386
+```
387
+
388
+
389
+| Parent | Stream compression | Child |
390
+|----------------------|--------------------|----------------------|
391
+| Supported & Enabled | compressed | Supported & Enabled |
392
+| (Supported & Disabled)/Not supported | uncompressed | Supported & Enabled |
393
+| Supported & Enabled | uncompressed | (Supported & Disabled)/Not supported |
394
+| (Supported & Disabled)/Not supported | uncompressed | (Supported & Disabled)/Not supported |
395
+
396
+In case of parents with multiple children you can select which streams will be compressed by using the same configuration under the `[API_KEY]`, `[MACHINE_GUID]` section.
397
+
398
+This configuration uses AND logic with the default stream compression configuration under the `[stream]` section. This means the stream compression from child to parent will be enabled only if the outcome of the AND logic operation is true (`default compression enabled` && `api key compression enabled`). So both should be enabled to get stream compression otherwise stream compression is disabled.
399
+```
400
+[API_KEY]
401
+ enable compression = yes | no
402
+```
403
+Same thing applies with the `[MACHINE_GUID]` configuration.
404
+```
405
+[MACHINE_GUID]
406
+ enable compression = yes | no
407
+```
408
## Viewing remote host dashboards, using mirrored databases
409
410
On any receiving Netdata, that maintains remote databases and has its web server enabled,
@@ -385,7 +474,12 @@ I will use this API_KEY: `11111111-2222-3333-4444-555555555555`. Replace it with
474
475
#### Configuring the parent
476
388
-On the parent, edit `/etc/netdata/stream.conf` (to edit it on your system run `/etc/netdata/edit-config stream.conf`) and set these:
477
+To configure the parent node:
478
+
479
+1. On the parent node, edit `stream.conf` by using the `edit-config` script:
480
+`/etc/netdata/edit-config stream.conf`
481
+
482
+2. Set the following parameters:
483
484
```bash
485
[11111111-2222-3333-4444-555555555555]
@@ -414,7 +508,12 @@ the `netdata` process, but a system power issue would leave the connection open
508
509
#### Configuring the child nodes
510
417
-On each of the child nodes, edit `/etc/netdata/stream.conf` (to edit it on your system run `/etc/netdata/edit-config stream.conf`) and set these:
511
+To configure the child node:
512
+
513
+1. On the child node, edit `stream.conf` by using the `edit-config` script:
514
+`/etc/netdata/edit-config stream.conf`
515
+
516
+2. Set the following parameters:
517
518
```bash
519
[stream]
@@ -446,9 +545,9 @@ _`netdata.conf` configuration on child nodes, to disable the local database and
545
546
Keep in mind that setting `memory mode = none` will also force `[health].enabled = no` (health checks require access to a local database). But you can keep the database and disable health checks if you need to. You are however sending all the metrics to the parent node, which can handle the health checking (`[health].enabled = yes`)
547
449
-#### Netdata unique id
548
+#### Netdata unique ID
549
451
-The file `/var/lib/netdata/registry/netdata.public.unique.id` contains a random GUID that **uniquely identifies each Netdata**. This file is automatically generated, by Netdata, the first time it is started and remains unaltered forever.
550
+The file `/var/lib/netdata/registry/netdata.public.unique.id` contains a random GUID that **uniquely identifies each Netdata Agent**. This file is automatically generated, by Netdata, the first time it is started and remains unaltered forever.
551
552
> If you are building an image to be used for automated provisioning of autoscaled VMs, it important to delete that file from the image, so that each instance of your image will generate its own.
553
@@ -456,7 +555,7 @@ The file `/var/lib/netdata/registry/netdata.public.unique.id` contains a random
555
556
Both parent and child nodes log information at `/var/log/netdata/error.log`.
557
459
-Run the following on both the parent and child nodes:
558
+To obtain the error logs, run the following on both the parent and child nodes:
559
560
```
561
tail -f /var/log/netdata/error.log | grep STREAM
@@ -511,7 +610,7 @@ This means a setup like the following is also possible:
610
611
## Proxies
612
514
-A proxy is a Netdata instance that is receiving metrics from a Netdata, and streams them to another Netdata.
613
+A proxy is a Netdata node that is receiving metrics from a Netdata node, and streams them to another Netdata node.
614
615
Netdata proxies may or may not maintain a database for the metrics passing through them.
616
When they maintain a database, they can also run health checks (alarms and notifications)
@@ -571,11 +670,11 @@ down), you will see the following in the child's `error.log`.
670
ERROR : STREAM_SENDER[HOSTNAME] : Failed to connect to 'PARENT IP', port 'PARENT PORT' (errno 113, No route to host)
671
```
672
574
-### 'Is this a Netdata?'
673
+### 'Is this a Netdata node?'
674
675
This question can appear when Netdata starts the stream and receives an unexpected response. This error can appear when
676
the parent is using SSL and the child tries to connect using plain text. You will also see this message when
578
-Netdata connects to another server that isn't Netdata. The complete error message will look like this:
677
+Netdata connects to another server that isn't a Netdata node. The complete error message will look like this:
678
679
```
680
ERROR : STREAM_SENDER[CHILD HOSTNAME] : STREAM child HOSTNAME [send to PARENT HOSTNAME:PARENT PORT]: server is not replying properly (is it a netdata?).
streaming/compression.c
new
+345
@@ -0,0 +1,345 @@
1
+#include "rrdpush.h"
2
+#include "lz4.h"
3
+
4
+#ifdef ENABLE_COMPRESSION
5
+
6
+#define LZ4_MAX_MSG_SIZE 0x4000
7
+#define LZ4_STREAM_BUFFER_SIZE (0x10000 + LZ4_MAX_MSG_SIZE)
8
+
9
+#define SIGNATURE ((uint32_t)('z' | 0x80) | (0x80 << 8) | (0x80 << 16) | ('\n' << 24))
10
+#define SIGNATURE_MASK ((uint32_t)0xff | (0x80 << 8) | (0x80 << 16) | (0xff << 24))
11
+#define SIGNATURE_SIZE 4
12
+
13
+
14
+/*
15
+ * LZ4 streaming API compressor specific data
16
+ */
17
+struct compressor_data {
18
+ LZ4_stream_t *stream;
19
+ char *stream_buffer;
20
+ size_t stream_buffer_pos;
21
+};
22
+
23
+
24
+/*
25
+ * Reset compressor state for a new stream
26
+ */
27
+static void lz4_compressor_reset(struct compressor_state *state)
28
+{
29
+ if (state->data) {
30
+ if (state->data->stream) {
31
+ LZ4_resetStream_fast(state->data->stream);
32
+ info("STREAM_COMPRESSION: Compressor resets stream fast!");
33
+ }
34
+ state->data->stream_buffer_pos = 0;
35
+ }
36
+}
37
+
38
+/*
39
+ * Destroy compressor state and all related data
40
+ */
41
+static void lz4_compressor_destroy(struct compressor_state **state)
42
+{
43
+ if (state && *state) {
44
+ struct compressor_state *s = *state;
45
+ if (s->data) {
46
+ if (s->data->stream)
47
+ LZ4_freeStream(s->data->stream);
48
+ freez(s->data->stream_buffer);
49
+ }
50
+ freez(s->buffer);
51
+ freez(s);
52
+ *state = NULL;
53
+ debug(D_STREAM, "STREAM_COMPRESSION: Compressor destroyed!");
54
+ }
55
+}
56
+
57
+/*
58
+ * Compress the given block of data
59
+ * Comprecced data will remain in the internal buffer until the next invokation
60
+ * Return the size of compressed data block as result and the pointer to internal buffer using the last argument
61
+ * or 0 in case of error
62
+ */
63
+static size_t lz4_compressor_compress(struct compressor_state *state, const char *data, size_t size, char **out)
64
+{
65
+ if (!state || !size || !out)
66
+ return 0;
67
+ if (size > LZ4_MAX_MSG_SIZE) {
68
+ error("Message size above limit: %lu", size);
69
+ return 0;
70
+ }
71
+ size_t max_dst_size = LZ4_COMPRESSBOUND(size);
72
+ size_t data_size = max_dst_size + SIGNATURE_SIZE;
73
+
74
+ if (!state->buffer) {
75
+ state->buffer = mallocz(data_size);
76
+ state->buffer_size = data_size;
77
+ } else if (state->buffer_size < data_size) {
78
+ state->buffer = reallocz(state->buffer, data_size);
79
+ state->buffer_size = data_size;
80
+ }
81
+
82
+ memcpy(state->data->stream_buffer + state->data->stream_buffer_pos, data, size);
83
+ long int compressed_data_size = LZ4_compress_fast_continue(state->data->stream,
84
+ state->data->stream_buffer + state->data->stream_buffer_pos,
85
+ state->buffer + SIGNATURE_SIZE, size, max_dst_size, 1);
86
+ if (compressed_data_size < 0) {
87
+ error("Date compression error: %ld", compressed_data_size);
88
+ return 0;
89
+ }
90
+ state->data->stream_buffer_pos += size;
91
+ if (state->data->stream_buffer_pos >= LZ4_STREAM_BUFFER_SIZE - LZ4_MAX_MSG_SIZE)
92
+ state->data->stream_buffer_pos = 0;
93
+ uint32_t len = ((compressed_data_size & 0x7f) | 0x80 | (((compressed_data_size & (0x7f << 7)) << 1) | 0x8000)) << 8;
94
+ *(uint32_t *)state->buffer = len | SIGNATURE;
95
+ *out = state->buffer;
96
+ debug(D_STREAM, "STREAM: Compressed data header: %ld", compressed_data_size);
97
+ return compressed_data_size + SIGNATURE_SIZE;
98
+}
99
+
100
+/*
101
+ * Create and initalize compressor state
102
+ * Return the pointer to compressor_state structure created
103
+ */
104
+struct compressor_state *create_compressor()
105
+{
106
+ struct compressor_state *state = callocz(1, sizeof(struct compressor_state));
107
+
108
+ state->reset = lz4_compressor_reset;
109
+ state->compress = lz4_compressor_compress;
110
+ state->destroy = lz4_compressor_destroy;
111
+
112
+ state->data = callocz(1, sizeof(struct compressor_data));
113
+ state->data->stream = LZ4_createStream();
114
+ state->data->stream_buffer = callocz(1, LZ4_DECODER_RING_BUFFER_SIZE(LZ4_MAX_MSG_SIZE));
115
+ state->buffer_size = LZ4_STREAM_BUFFER_SIZE;
116
+ state->reset(state);
117
+ debug(D_STREAM, "STREAM_COMPRESSION: Initialize streaming compression!");
118
+ return state;
119
+}
120
+
121
+/*
122
+ * LZ4 streaming API decompressor specific data
123
+ */
124
+struct decompressor_data {
125
+ LZ4_streamDecode_t *stream;
126
+ char *stream_buffer;
127
+ size_t stream_buffer_size;
128
+ size_t stream_buffer_pos;
129
+};
130
+
131
+/*
132
+ * Reset decompressor state for a new stream
133
+ */
134
+static void lz4_decompressor_reset(struct decompressor_state *state)
135
+{
136
+ if (state->data) {
137
+ if (state->data->stream)
138
+ LZ4_setStreamDecode(state->data->stream, NULL, 0);
139
+ state->data->stream_buffer_pos = 0;
140
+ state->buffer_len = 0;
141
+ state->out_buffer_len = 0;
142
+ }
143
+}
144
+
145
+/*
146
+ * Destroy decompressor state and all related data
147
+ */
148
+static void lz4_decompressor_destroy(struct decompressor_state **state)
149
+{
150
+ if (state && *state) {
151
+ struct decompressor_state *s = *state;
152
+ if (s->data) {
153
+ debug(D_STREAM, "STREAM_COMPRESSION: Destroying decompressor.");
154
+ if (s->data->stream)
155
+ LZ4_freeStreamDecode(s->data->stream);
156
+ freez(s->data->stream_buffer);
157
+ }
158
+ freez(s->buffer);
159
+ freez(s);
160
+ *state = NULL;
161
+ }
162
+}
163
+
164
+static size_t decode_compress_header(const char *data, size_t data_size)
165
+{
166
+ if (!data || !data_size)
167
+ return 0;
168
+ if (data_size < SIGNATURE_SIZE)
169
+ return 0;
170
+ uint32_t sign = *(uint32_t *)data;
171
+ if ((sign & SIGNATURE_MASK) != SIGNATURE)
172
+ return 0;
173
+ size_t length = ((sign >> 8) & 0x7f) | ((sign >> 9) & (0x7f << 7));
174
+ return length;
175
+}
176
+
177
+/*
178
+ * Check input data for the compression header
179
+ * Return the size of compressed data or 0 for uncompressed data
180
+ */
181
+size_t is_compressed_data(const char *data, size_t data_size)
182
+{
183
+ return decode_compress_header(data, data_size);
184
+}
185
+
186
+/*
187
+ * Start the collection of compressed data in an internal buffer
188
+ * Return the size of compressed data or 0 for uncompressed data
189
+ */
190
+static size_t lz4_decompressor_start(struct decompressor_state *state, const char *header, size_t header_size)
191
+{
192
+ size_t length = decode_compress_header(header, header_size);
193
+ if (!length)
194
+ return 0;
195
+
196
+ if (!state->buffer) {
197
+ state->buffer = mallocz(length);
198
+ state->buffer_size = length;
199
+ } else if (state->buffer_size < length) {
200
+ state->buffer = reallocz(state->buffer, length);
201
+ state->buffer_size = length;
202
+ }
203
+ state->buffer_len = length;
204
+ state->buffer_pos = 0;
205
+ state->out_buffer_pos = 0;
206
+ state->out_buffer_len = 0;
207
+ return length;
208
+}
209
+
210
+/*
211
+ * Add a chunk of compressed data to the internal buffer
212
+ * Return the current size of compressed data or 0 for error
213
+ */
214
+static size_t lz4_decompressor_put(struct decompressor_state *state, const char *data, size_t size)
215
+{
216
+ if (!state || !size || !data)
217
+ return 0;
218
+ if (!state->buffer)
219
+ fatal("STREAM: No decompressor buffer allocated");
220
+
221
+ if (state->buffer_pos + size > state->buffer_len) {
222
+ error("STREAM: Decompressor buffer overflow %lu + %lu > %lu",
223
+ state->buffer_pos, size, state->buffer_len);
224
+ size = state->buffer_len - state->buffer_pos;
225
+ }
226
+ memcpy(state->buffer + state->buffer_pos, data, size);
227
+ state->buffer_pos += size;
228
+ return state->buffer_pos;
229
+}
230
+
231
+static size_t saving_percent(size_t comp_len, size_t src_len)
232
+{
233
+ if (comp_len > src_len)
234
+ comp_len = src_len;
235
+ if (!src_len)
236
+ return 0;
237
+ return 100 - comp_len * 100 / src_len;
238
+}
239
+
240
+/*
241
+ * Decompress the compressed data in the internal buffer
242
+ * Return the size of uncompressed data or 0 for error
243
+ */
244
+static size_t lz4_decompressor_decompress(struct decompressor_state *state)
245
+{
246
+ if (!state)
247
+ return 0;
248
+ if (!state->buffer) {
249
+ error("STREAM: No decompressor buffer allocated");
250
+ return 0;
251
+ }
252
+
253
+ long int decompressed_size = LZ4_decompress_safe_continue(state->data->stream, state->buffer,
254
+ state->data->stream_buffer + state->data->stream_buffer_pos,
255
+ state->buffer_len, state->data->stream_buffer_size - state->data->stream_buffer_pos);
256
+ if (decompressed_size < 0) {
257
+ error("STREAM: Decompressor error %ld", decompressed_size);
258
+ return 0;
259
+ }
260
+
261
+ state->out_buffer = state->data->stream_buffer + state->data->stream_buffer_pos;
262
+ state->data->stream_buffer_pos += decompressed_size;
263
+ if (state->data->stream_buffer_pos >= state->data->stream_buffer_size - LZ4_MAX_MSG_SIZE)
264
+ state->data->stream_buffer_pos = 0;
265
+ state->out_buffer_len = decompressed_size;
266
+ state->out_buffer_pos = 0;
267
+
268
+ // Some compression statistics
269
+ size_t old_avg_saving = saving_percent(state->total_compressed, state->total_uncompressed);
270
+ size_t old_avg_size = state->packet_count ? state->total_uncompressed / state->packet_count : 0;
271
+
272
+ state->total_compressed += state->buffer_len + SIGNATURE_SIZE;
273
+ state->total_uncompressed += decompressed_size;
274
+ state->packet_count++;
275
+
276
+ size_t saving = saving_percent(state->buffer_len, decompressed_size);
277
+ size_t avg_saving = saving_percent(state->total_compressed, state->total_uncompressed);
278
+ size_t avg_size = state->total_uncompressed / state->packet_count;
279
+
280
+ if (old_avg_saving != avg_saving || old_avg_size != avg_size){
281
+ debug(D_STREAM, "STREAM: Saving: %lu%% (avg. %lu%%), avg.size: %lu", saving, avg_saving, avg_size);
282
+ }
283
+ return decompressed_size;
284
+}
285
+
286
+/*
287
+ * Return the size of uncompressed data left in the internal buffer or 0 for error
288
+ */
289
+static size_t lz4_decompressor_decompressed_bytes_in_buffer(struct decompressor_state *state)
290
+{
291
+ return state->out_buffer_len ?
292
+ state->out_buffer_len - state->out_buffer_pos : 0;
293
+}
294
+
295
+/*
296
+ * Fill the buffer provided with uncompressed data from the internal buffer
297
+ * Return the size of uncompressed data copied or 0 for error
298
+ */
299
+static size_t lz4_decompressor_get(struct decompressor_state *state, char *data, size_t size)
300
+{
301
+ if (!state || !size || !data)
302
+ return 0;
303
+ if (!state->out_buffer)
304
+ fatal("STREAM: No decompressor output buffer allocated");
305
+ if (state->out_buffer_pos + size > state->out_buffer_len)
306
+ size = state->out_buffer_len - state->out_buffer_pos;
307
+
308
+ char *p = state->out_buffer + state->out_buffer_pos, *endp = p + size, *last_lf = NULL;
309
+ for (; p < endp; ++p)
310
+ if (*p == '\n' || *p == 0)
311
+ last_lf = p;
312
+ if (last_lf)
313
+ size = last_lf + 1 - (state->out_buffer + state->out_buffer_pos);
314
+
315
+ memcpy(data, state->out_buffer + state->out_buffer_pos, size);
316
+ state->out_buffer_pos += size;
317
+ return size;
318
+}
319
+
320
+/*
321
+ * Create and initalize decompressor state
322
+ * Return the pointer to decompressor_state structure created
323
+ */
324
+struct decompressor_state *create_decompressor()
325
+{
326
+ struct decompressor_state *state = callocz(1, sizeof(struct decompressor_state));
327
+ state->reset = lz4_decompressor_reset;
328
+ state->start = lz4_decompressor_start;
329
+ state->put = lz4_decompressor_put;
330
+ state->decompress = lz4_decompressor_decompress;
331
+ state->get = lz4_decompressor_get;
332
+ state->decompressed_bytes_in_buffer = lz4_decompressor_decompressed_bytes_in_buffer;
333
+ state->destroy = lz4_decompressor_destroy;
334
+
335
+ state->data = callocz(1, sizeof(struct decompressor_data));
336
+ fatal_assert(state->data);
337
+ state->data->stream = LZ4_createStreamDecode();
338
+ state->data->stream_buffer_size = LZ4_decoderRingBufferSize(LZ4_MAX_MSG_SIZE);
339
+ state->data->stream_buffer = mallocz(state->data->stream_buffer_size);
340
+ fatal_assert(state->data->stream_buffer);
341
+ state->reset(state);
342
+ debug(D_STREAM, "STREAM_COMPRESSION: Initialize streaming decompression!");
343
+ return state;
344
+}
345
+#endif
streaming/receiver.c
+152
-4
@@ -21,6 +21,10 @@ void destroy_receiver_state(struct receiver_state *rpt) {
21
if(rpt->ssl.conn){
22
SSL_free(rpt->ssl.conn);
23
}
24
+#endif
25
+#ifdef ENABLE_COMPRESSION
26
+ if (rpt->decompressor)
27
+ rpt->decompressor->destroy(&rpt->decompressor);
28
#endif
29
freez(rpt);
30
}
@@ -159,6 +163,8 @@ PARSER_RC streaming_claimed_id(char **words, void *user, PLUGINSD_ACTION *plugin
163
return PARSER_RC_OK;
164
}
165
166
+
167
+#ifndef ENABLE_COMPRESSION
168
/* The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
169
*/
170
static int receiver_read(struct receiver_state *r, FILE *fp) {
@@ -186,6 +192,130 @@ static int receiver_read(struct receiver_state *r, FILE *fp) {
192
r->read_len = strlen(r->read_buffer);
193
return 0;
194
}
195
+#else
196
+/*
197
+ * The receiver socket is blocking, perform a single read into a buffer so that we can reassemble lines for parsing.
198
+ * if SSL encryption is on, then use SSL API for reading stream data.
199
+ * Use line oriented fgets() in buffer from receiver_state is provided.
200
+ * In other cases use fread to read binary data from socket.
201
+ * Return zero on success and the number of bytes were read using pointer in the last argument.
202
+ */
203
+static int read_stream(struct receiver_state *r, FILE *fp, char* buffer, size_t size, int* ret) {
204
+ if (!ret)
205
+ return 1;
206
+ *ret = 0;
207
+#ifdef ENABLE_HTTPS
208
+ if (r->ssl.conn && !r->ssl.flags) {
209
+ ERR_clear_error();
210
+ if (buffer != r->read_buffer + r->read_len) {
211
+ *ret = SSL_read(r->ssl.conn, buffer, size);
212
+ if (*ret > 0 )
213
+ return 0;
214
+ } else {
215
+ // we need to receive data with LF to parse compression header
216
+ size_t ofs = 0;
217
+ int res = 0;
218
+ while (ofs < size) {
219
+ do {
220
+ res = SSL_read(r->ssl.conn, buffer + ofs, 1);
221
+ } while (res == 0);
222
+
223
+ if (res < 0)
224
+ break;
225
+ if (buffer[ofs] == '\n')
226
+ break;
227
+ ofs += res;
228
+ }
229
+ if (res > 0) {
230
+ ofs += res;
231
+ *ret = ofs;
232
+ buffer[ofs] = 0;
233
+ return 0;
234
+ }
235
+ }
236
+ // Don't treat SSL_ERROR_WANT_READ or SSL_ERROR_WANT_WRITE differently on blocking socket
237
+ u_long err;
238
+ char buf[256];
239
+ while ((err = ERR_get_error()) != 0) {
240
+ ERR_error_string_n(err, buf, sizeof(buf));
241
+ error("STREAM %s [receive from %s] ssl error: %s", r->hostname, r->client_ip, buf);
242
+ }
243
+ return 1;
244
+ }
245
+#endif
246
+ if (buffer != r->read_buffer + r->read_len) {
247
+ // read to external buffer
248
+ *ret = fread(buffer, 1, size, fp);
249
+ if (!*ret)
250
+ return 1;
251
+ } else {
252
+ if (!fgets(r->read_buffer, sizeof(r->read_buffer), fp))
253
+ return 1;
254
+ *ret = strlen(r->read_buffer);
255
+ }
256
+ return 0;
257
+}
258
+
259
+/*
260
+ * Get the next line of data for parsing.
261
+ * Return data from the decompressor buffer if available.
262
+ * Otherwise read next line from the socket and check for compression header.
263
+ * Return the line was read If no compression header was found.
264
+ * Otherwise read the entire block of compressed data, decompress it
265
+ * and return it in receiver_state buffer.
266
+ * Return zero on success.
267
+ */
268
+static int receiver_read(struct receiver_state *r, FILE *fp) {
269
+ // check any decompressed data present
270
+ if (r->decompressor &&
271
+ r->decompressor->decompressed_bytes_in_buffer(r->decompressor)) {
272
+ size_t available = sizeof(r->read_buffer) - r->read_len;
273
+ if (available) {
274
+ size_t len = r->decompressor->get(r->decompressor,
275
+ r->read_buffer + r->read_len, available);
276
+ if (!len)
277
+ return 1;
278
+ r->read_len += len;
279
+ }
280
+ return 0;
281
+ }
282
+ int ret = 0;
283
+ if (read_stream(r, fp, r->read_buffer + r->read_len, sizeof(r->read_buffer) - r->read_len - 1, &ret))
284
+ return 1;
285
+
286
+ if (!is_compressed_data(r->read_buffer, ret)) {
287
+ r->read_len += ret;
288
+ return 0;
289
+ }
290
+
291
+ if (unlikely(!r->decompressor))
292
+ r->decompressor = create_decompressor();
293
+
294
+ size_t bytes_to_read = r->decompressor->start(r->decompressor,
295
+ r->read_buffer, ret);
296
+
297
+ // Read the entire block of compressed data because
298
+ // we're unable to decompress incomplete block
299
+ char compressed[bytes_to_read];
300
+ do {
301
+ if (read_stream(r, fp, compressed, bytes_to_read, &ret))
302
+ return 1;
303
+ // Send input data to decompressor
304
+ if (ret)
305
+ r->decompressor->put(r->decompressor, compressed, ret);
306
+ bytes_to_read -= ret;
307
+ } while (bytes_to_read > 0);
308
+ // Decompress
309
+ size_t bytes_to_parse = r->decompressor->decompress(r->decompressor);
310
+ if (!bytes_to_parse)
311
+ return 1;
312
+ // Fill read buffer with decompressed data
313
+ r->read_len = r->decompressor->get(r->decompressor,
314
+ r->read_buffer, sizeof(r->read_buffer));
315
+ return 0;
316
+}
317
+
318
+#endif
319
320
/* Produce a full line if one exists, statefully return where we start next time.
321
* When we hit the end of the buffer with a partial line move it to the beginning for the next fill.
@@ -208,7 +338,6 @@ static char *receiver_next_line(struct receiver_state *r, int *pos) {
338
return NULL;
339
}
340
211
-
341
size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp) {
342
size_t result;
343
PARSER_USER_OBJECT *user = callocz(1, sizeof(*user));
@@ -239,12 +368,14 @@ size_t streaming_parser(struct receiver_state *rpt, struct plugind *cd, FILE *fp
368
parser->plugins_action->overwrite_action = &pluginsd_overwrite_action;
369
parser->plugins_action->chart_action = &pluginsd_chart_action;
370
parser->plugins_action->set_action = &pluginsd_set_action;
242
- parser->plugins_action->clabel_commit_action = &pluginsd_clabel_commit_action;
243
- parser->plugins_action->clabel_action = &pluginsd_clabel_action;
371
372
user->parser = parser;
373
247
- do {
374
+#ifdef ENABLE_COMPRESSION
375
+ if (rpt->decompressor)
376
+ rpt->decompressor->reset(rpt->decompressor);
377
+#endif
378
+ do{
379
if (receiver_read(rpt, fp))
380
break;
381
int pos = 0;
@@ -311,6 +442,13 @@ static int rrdpush_receive(struct receiver_state *rpt)
442
rrdpush_send_charts_matching = appconfig_get(&stream_config, rpt->key, "default proxy send charts matching", rrdpush_send_charts_matching);
443
rrdpush_send_charts_matching = appconfig_get(&stream_config, rpt->machine_guid, "proxy send charts matching", rrdpush_send_charts_matching);
444
445
+#ifdef ENABLE_COMPRESSION
446
+ unsigned int rrdpush_compression = default_compression_enabled;
447
+ rrdpush_compression = appconfig_get_boolean(&stream_config, rpt->key, "enable compression", rrdpush_compression);
448
+ rrdpush_compression = appconfig_get_boolean(&stream_config, rpt->machine_guid, "enable compression", rrdpush_compression);
449
+ rpt->rrdpush_compression = (rrdpush_compression && default_compression_enabled);
450
+#endif //ENABLE_COMPRESSION
451
+
452
(void)appconfig_set_default(&stream_config, rpt->machine_guid, "host tags", (rpt->tags)?rpt->tags:"");
453
454
if (strcmp(rpt->machine_guid, localhost->machine_guid) == 0) {
@@ -432,6 +570,16 @@ static int rrdpush_receive(struct receiver_state *rpt)
570
info("STREAM %s [receive from [%s]:%s]: initializing communication...", rpt->host->hostname, rpt->client_ip, rpt->client_port);
571
char initial_response[HTTP_HEADER_SIZE];
572
if (rpt->stream_version > 1) {
573
+ if(rpt->stream_version >= STREAM_VERSION_COMPRESSION){
574
+#ifdef ENABLE_COMPRESSION
575
+ if(!rpt->rrdpush_compression)
576
+ rpt->stream_version = STREAM_VERSION_CLABELS;
577
+#else
578
+ if(STREAMING_PROTOCOL_CURRENT_VERSION < rpt->stream_version) {
579
+ rpt->stream_version = STREAMING_PROTOCOL_CURRENT_VERSION;
580
+ }
581
+#endif
582
+ }
583
info("STREAM %s [receive from [%s]:%s]: Netdata is using the stream version %u.", rpt->host->hostname, rpt->client_ip, rpt->client_port, rpt->stream_version);
584
sprintf(initial_response, "%s%u", START_STREAMING_PROMPT_VN, rpt->stream_version);
585
} else if (rpt->stream_version == 1) {
streaming/rrdpush.c
+8
-2
@@ -40,6 +40,9 @@ struct config stream_config = {
40
};
41
42
unsigned int default_rrdpush_enabled = 0;
43
+#ifdef ENABLE_COMPRESSION
44
+unsigned int default_compression_enabled = 1;
45
+#endif
46
char *default_rrdpush_destination = NULL;
47
char *default_rrdpush_api_key = NULL;
48
char *default_rrdpush_send_charts_matching = NULL;
@@ -73,7 +76,10 @@ int rrdpush_init() {
76
default_rrdpush_api_key = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "api key", "");
77
default_rrdpush_send_charts_matching = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "send charts matching", "*");
78
rrdhost_free_orphan_time = config_get_number(CONFIG_SECTION_GLOBAL, "cleanup orphan hosts after seconds", rrdhost_free_orphan_time);
76
-
79
+#ifdef ENABLE_COMPRESSION
80
+ default_compression_enabled = (unsigned int)appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM,
81
+ "enable compression", default_compression_enabled);
82
+#endif
83
84
if(default_rrdpush_enabled && (!default_rrdpush_destination || !*default_rrdpush_destination || !default_rrdpush_api_key || !*default_rrdpush_api_key)) {
85
error("STREAM [send]: cannot enable sending thread - information is missing.");
@@ -775,4 +781,4 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *url) {
781
782
buffer_flush(w->response.data);
783
return 200;
778
-}
784
+}
\ No newline at end of file
streaming/rrdpush.h
+62
-7
@@ -10,10 +10,16 @@
10
11
#define CONNECTED_TO_SIZE 100
12
13
-#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)4
13
#define STREAM_VERSION_CLAIM 3
14
#define STREAM_VERSION_CLABELS 4
16
-#define VERSION_GAP_FILLING 5
15
+#define STREAM_VERSION_COMPRESSION 5
16
+#define VERSION_GAP_FILLING 6
17
+
18
+#ifdef ENABLE_COMPRESSION
19
+#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)(STREAM_VERSION_COMPRESSION)
20
+#else
21
+#define STREAMING_PROTOCOL_CURRENT_VERSION (uint32_t)(STREAM_VERSION_CLABELS)
22
+#endif //ENABLE_COMPRESSION
23
24
#define STREAMING_PROTOCOL_VERSION "1.1"
25
#define START_STREAMING_PROMPT "Hit me baby, push them over..."
@@ -35,6 +41,38 @@ typedef struct {
41
char *kernel_version;
42
} stream_encoded_t;
43
44
+#ifdef ENABLE_COMPRESSION
45
+struct compressor_state {
46
+ char *buffer;
47
+ size_t buffer_size;
48
+ struct compressor_data *data; // Compression API specific data
49
+ void (*reset)(struct compressor_state *state);
50
+ size_t (*compress)(struct compressor_state *state, const char *data, size_t size, char **buffer);
51
+ void (*destroy)(struct compressor_state **state);
52
+};
53
+
54
+struct decompressor_state {
55
+ char *buffer;
56
+ size_t buffer_size;
57
+ size_t buffer_len;
58
+ size_t buffer_pos;
59
+ char *out_buffer;
60
+ size_t out_buffer_len;
61
+ size_t out_buffer_pos;
62
+ size_t total_compressed;
63
+ size_t total_uncompressed;
64
+ size_t packet_count;
65
+ struct decompressor_data *data; // Deompression API specific data
66
+ void (*reset)(struct decompressor_state *state);
67
+ size_t (*start)(struct decompressor_state *state, const char *header, size_t header_size);
68
+ size_t (*put)(struct decompressor_state *state, const char *data, size_t size);
69
+ size_t (*decompress)(struct decompressor_state *state);
70
+ size_t (*decompressed_bytes_in_buffer)(struct decompressor_state *state);
71
+ size_t (*get)(struct decompressor_state *state, char *data, size_t size);
72
+ void (*destroy)(struct decompressor_state **state);
73
+};
74
+#endif
75
+
76
// Thread-local storage
77
// Metric transmission: collector threads asynchronously fill the buffer, sender thread uses it.
78
@@ -60,6 +98,10 @@ struct sender_state {
98
char read_buffer[512];
99
int read_len;
100
int32_t version;
101
+#ifdef ENABLE_COMPRESSION
102
+ unsigned int rrdpush_compression;
103
+ struct compressor_state *compressor;
104
+#endif
105
};
106
107
struct receiver_state {
@@ -75,9 +117,9 @@ struct receiver_state {
117
char *abbrev_timezone;
118
int32_t utc_offset;
119
char *tags;
78
- char *client_ip; // Duplicated in pluginsd
79
- char *client_port; // Duplicated in pluginsd
80
- char *program_name; // Duplicated in pluginsd
120
+ char *client_ip; // Duplicated in pluginsd
121
+ char *client_port; // Duplicated in pluginsd
122
+ char *program_name; // Duplicated in pluginsd
123
char *program_version;
124
struct rrdhost_system_info *system_info;
125
int update_every;
@@ -85,15 +127,22 @@ struct receiver_state {
127
time_t last_msg_t;
128
char read_buffer[1024]; // Need to allow RRD_ID_LENGTH_MAX * 4 + the other fields
129
int read_len;
130
+ unsigned int shutdown:1; // Tell the thread to exit
131
+ unsigned int exited; // Indicates that the thread has exited (NOT A BITFIELD!)
132
#ifdef ENABLE_HTTPS
133
struct netdata_ssl ssl;
134
#endif
91
- unsigned int shutdown:1; // Tell the thread to exit
92
- unsigned int exited; // Indicates that the thread has exited (NOT A BITFIELD!)
135
+#ifdef ENABLE_COMPRESSION
136
+ unsigned int rrdpush_compression;
137
+ struct decompressor_state *decompressor;
138
+#endif
139
};
140
141
142
extern unsigned int default_rrdpush_enabled;
143
+#ifdef ENABLE_COMPRESSION
144
+extern unsigned int default_compression_enabled;
145
+#endif
146
extern char *default_rrdpush_destination;
147
extern char *default_rrdpush_api_key;
148
extern char *default_rrdpush_send_charts_matching;
@@ -116,4 +165,10 @@ extern void rrdpush_sender_thread_stop(RRDHOST *host);
165
extern void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, RRDVAR *rv);
166
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);
167
168
+#ifdef ENABLE_COMPRESSION
169
+struct compressor_state *create_compressor();
170
+struct decompressor_state *create_decompressor();
171
+size_t is_compressed_data(const char *data, size_t data_size);
172
+#endif
173
+
174
#endif //NETDATA_RRDPUSH_H
streaming/sender.c
+71
-29
@@ -15,9 +15,26 @@ void sender_start(struct sender_state *s) {
15
16
// Collector thread finishing a transmission
17
void sender_commit(struct sender_state *s) {
18
- if(cbuffer_add_unsafe(s->host->sender->buffer, buffer_tostring(s->host->sender->build),
19
- s->host->sender->build->len))
18
+ char *src = (char *)buffer_tostring(s->host->sender->build);
19
+ size_t src_len = s->host->sender->build->len;
20
+#ifdef ENABLE_COMPRESSION
21
+ do {
22
+ if (src && src_len) {
23
+ if (s->compressor && s->rrdpush_compression) {
24
+ src_len = s->compressor->compress(s->compressor, src, src_len, &src);
25
+ if (!src_len) {
26
+ error("Compression error - data discarded");
27
+ break;
28
+ }
29
+ }
30
+ if(cbuffer_add_unsafe(s->host->sender->buffer, src, src_len))
31
+ s->overflow = 1;
32
+ }
33
+ } while (0);
34
+#else
35
+ if(cbuffer_add_unsafe(s->host->sender->buffer, src, src_len))
36
s->overflow = 1;
37
+#endif
38
buffer_flush(s->build);
39
netdata_mutex_unlock(&s->mutex);
40
}
@@ -147,6 +164,35 @@ void rrdpush_clean_encoded(stream_encoded_t *se)
164
freez(se->kernel_version);
165
}
166
167
+static inline long int parse_stream_version(RRDHOST *host, char *http)
168
+{
169
+ long int stream_version = -1;
170
+ int answer = -1;
171
+ char *stream_version_start = strchr(http, '=');
172
+ if (stream_version_start) {
173
+ stream_version_start++;
174
+ stream_version = strtol(stream_version_start, NULL, 10);
175
+ answer = memcmp(http, START_STREAMING_PROMPT_VN, (size_t)(stream_version_start - http));
176
+ if (!answer) {
177
+ rrdpush_set_flags_to_newest_stream(host);
178
+ }
179
+ } else {
180
+ answer = memcmp(http, START_STREAMING_PROMPT_V2, strlen(START_STREAMING_PROMPT_V2));
181
+ if (!answer) {
182
+ stream_version = 1;
183
+ rrdpush_set_flags_to_newest_stream(host);
184
+ } else {
185
+ answer = memcmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT));
186
+ if (!answer) {
187
+ stream_version = 0;
188
+ host->labels.labels_flag |= LABEL_FLAG_STOP_STREAM;
189
+ host->labels.labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
190
+ }
191
+ }
192
+ }
193
+ return stream_version;
194
+}
195
+
196
static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_port, int timeout,
197
struct sender_state *s) {
198
@@ -355,32 +401,7 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po
401
402
http[received] = '\0';
403
debug(D_STREAM, "Response to sender from far end: %s", http);
358
- int answer = -1;
359
- char *version_start = strchr(http, '=');
360
- int32_t version = -1;
361
- if(version_start) {
362
- version_start++;
363
- version = (int32_t)strtol(version_start, NULL, 10);
364
- answer = memcmp(http, START_STREAMING_PROMPT_VN, (size_t)(version_start - http));
365
- if(!answer) {
366
- rrdpush_set_flags_to_newest_stream(host);
367
- }
368
- } else {
369
- answer = memcmp(http, START_STREAMING_PROMPT_V2, strlen(START_STREAMING_PROMPT_V2));
370
- if(!answer) {
371
- version = 1;
372
- rrdpush_set_flags_to_newest_stream(host);
373
- }
374
- else {
375
- answer = memcmp(http, START_STREAMING_PROMPT, strlen(START_STREAMING_PROMPT));
376
- if(!answer) {
377
- version = 0;
378
- host->labels.labels_flag |= LABEL_FLAG_STOP_STREAM;
379
- host->labels.labels_flag &= ~LABEL_FLAG_UPDATE_STREAM;
380
- }
381
- }
382
- }
383
-
404
+ int32_t version = (int32_t)parse_stream_version(host, http);
405
if(version == -1) {
406
error("STREAM %s [send to %s]: server is not replying properly (is it a netdata?).", host->hostname, s->connected_to);
407
rrdpush_sender_thread_close_socket(host);
@@ -388,10 +409,26 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po
409
}
410
s->version = version;
411
412
+#ifdef ENABLE_COMPRESSION
413
+ s->rrdpush_compression = (default_compression_enabled && (s->version >= STREAM_VERSION_COMPRESSION));
414
+ if(s->rrdpush_compression)
415
+ {
416
+ // parent supports compression
417
+ if(s->compressor)
418
+ s->compressor->reset(s->compressor);
419
+ }
420
+ else {
421
+ //parent does not support compression or has compression disabled
422
+ debug(D_STREAM, "Stream is uncompressed! One of the agents (%s <-> %s) does not support compression OR compression is disabled.", s->connected_to, s->host->hostname);
423
+ s->version = STREAM_VERSION_CLABELS;
424
+ }
425
+#endif //ENABLE_COMPRESSION
426
+
427
+
428
info("STREAM %s [send to %s]: established communication with a parent using protocol version %d - ready to send metrics..."
429
, host->hostname
430
, s->connected_to
394
- , version);
431
+ , s->version);
432
433
if(sock_setnonblock(host->rrdpush_sender_socket) < 0)
434
error("STREAM %s [send to %s]: cannot set non-blocking mode for socket.", host->hostname, s->connected_to);
@@ -581,6 +618,11 @@ void sender_init(struct sender_state *s, RRDHOST *parent) {
618
s->host = parent;
619
s->buffer = cbuffer_new(1024, 1024*1024);
620
s->build = buffer_create(1);
621
+#ifdef ENABLE_COMPRESSION
622
+ s->rrdpush_compression = default_compression_enabled;
623
+ if (default_compression_enabled)
624
+ s->compressor = create_compressor();
625
+#endif
626
netdata_mutex_init(&s->mutex);
627
}
628
streaming/stream.conf
+19
@@ -60,6 +60,12 @@
60
# The API_KEY to use (as the sender)
61
api key =
62
63
+ # Stream Compresssion
64
+ #
65
+ # The netdata child is configurated to enable stream compression by default.
66
+ # You can control stream compression in this agent with options: yes | no
67
+ #enable compression = yes
68
+
69
# The timeout to connect and send metrics
70
timeout seconds = 60
71
@@ -156,6 +162,12 @@
162
#default proxy api key = API_KEY
163
#default proxy send charts matching = *
164
165
+ # Stream Compresssion
166
+ #
167
+ # The stream with the child can be configurated to enable stream compression.
168
+ # You can control stream compression in this parent agent stream with options: yes | no
169
+ #enable compression = yes
170
+
171
172
# -----------------------------------------------------------------------------
173
# 3. PER SENDING HOST SETTINGS, ON PARENT NETDATA
@@ -203,3 +215,10 @@
215
#proxy destination = IP:PORT IP:PORT ...
216
#proxy api key = API_KEY
217
#proxy send charts matching = *
218
+
219
+ # Stream Compresssion
220
+ #
221
+ # The stream with the child can be configurated to enable stream compression.
222
+ # You can control stream compression in this parent agent stream with options: yes | no
223
+ #enable compression = yes
224
+
\ No newline at end of file
web/api/web_api_v1.c
+8
@@ -1035,6 +1035,14 @@ inline int web_client_api_request_v1_info_fill_buffer(RRDHOST *host, BUFFER *wb)
1035
analytics_get_data(analytics_data.netdata_config_stream_enabled, wb);
1036
buffer_strcat(wb, ",\n");
1037
1038
+#ifdef ENABLE_COMPRESSION
1039
+ buffer_strcat(wb, "\t\"stream-compression\": ");
1040
+ buffer_strcat(wb, (default_compression_enabled ? "\"enabled\"" : "\"disabled\""));
1041
+ buffer_strcat(wb, ",\n");
1042
+#else
1043
+ buffer_strcat(wb, "\t\"stream-compression\": \"N/A\",\n");
1044
+#endif //ENABLE_COMPRESSION
1045
+
1046
buffer_strcat(wb, "\t\"hosts-available\": ");
1047
analytics_get_data(analytics_data.netdata_config_hosts_available, wb);
1048
buffer_strcat(wb, ",\n");