support filtering of charts during streaming; fixes #4223 (#4361)

7 files changed, 88 insertions, 37 deletions

diff --git a/src/database/rrd.h b/src/database/rrd.h
index ac3d4c3ae3..51f826c4ba 100644
--- a/src/database/rrd.h
+++ b/src/database/rrd.h
@@ -240,19 +240,21 @@ struct rrddim {
 // and may lead to missing information.
 
 typedef enum rrdset_flags {
-    RRDSET_FLAG_ENABLED          = 1 << 0, // enables or disables a chart
-    RRDSET_FLAG_DETAIL           = 1 << 1, // if set, the data set should be considered as a detail of another
+    RRDSET_FLAG_ENABLED           = 1 << 0, // enables or disables a chart
+    RRDSET_FLAG_DETAIL            = 1 << 1, // if set, the data set should be considered as a detail of another
                                          // (the master data set should be the one that has the same family and is not detail)
-    RRDSET_FLAG_DEBUG            = 1 << 2, // enables or disables debugging for a chart
-    RRDSET_FLAG_OBSOLETE         = 1 << 3, // this is marked by the collector/module as obsolete
-    RRDSET_FLAG_BACKEND_SEND     = 1 << 4, // if set, this chart should be sent to backends
-    RRDSET_FLAG_BACKEND_IGNORE   = 1 << 5, // if set, this chart should not be sent to backends
-    RRDSET_FLAG_EXPOSED_UPSTREAM = 1 << 6, // if set, we have sent this chart to netdata master (streaming)
-    RRDSET_FLAG_STORE_FIRST      = 1 << 7, // if set, do not eliminate the first collection during interpolation
-    RRDSET_FLAG_HETEROGENEOUS    = 1 << 8, // if set, the chart is not homogeneous (dimensions in it have multiple algorithms, multipliers or dividers)
-    RRDSET_FLAG_HOMEGENEOUS_CHECK= 1 << 9, // if set, the chart should be checked to determine if the dimensions as homogeneous
-    RRDSET_FLAG_HIDDEN           = 1 << 10, // if set, do not show this chart on the dashboard, but use it for backends
-    RRDSET_FLAG_SYNC_CLOCK       = 1 << 11, // if set, microseconds on next data collection will be ignored (the chart will be synced to now)
+    RRDSET_FLAG_DEBUG             = 1 << 2, // enables or disables debugging for a chart
+    RRDSET_FLAG_OBSOLETE          = 1 << 3, // this is marked by the collector/module as obsolete
+    RRDSET_FLAG_BACKEND_SEND      = 1 << 4, // if set, this chart should be sent to backends
+    RRDSET_FLAG_BACKEND_IGNORE    = 1 << 5, // if set, this chart should not be sent to backends
+    RRDSET_FLAG_UPSTREAM_SEND     = 1 << 6, // if set, this chart should be sent upstream (streaming)
+    RRDSET_FLAG_UPSTREAM_IGNORE   = 1 << 7, // if set, this chart should not be sent upstream (streaming)
+    RRDSET_FLAG_UPSTREAM_EXPOSED  = 1 << 8, // if set, we have sent this chart definition to netdata master (streaming)
+    RRDSET_FLAG_STORE_FIRST       = 1 << 9, // if set, do not eliminate the first collection during interpolation
+    RRDSET_FLAG_HETEROGENEOUS     = 1 << 10, // if set, the chart is not homogeneous (dimensions in it have multiple algorithms, multipliers or dividers)
+    RRDSET_FLAG_HOMEGENEOUS_CHECK = 1 << 11, // if set, the chart should be checked to determine if the dimensions as homogeneous
+    RRDSET_FLAG_HIDDEN            = 1 << 12, // if set, do not show this chart on the dashboard, but use it for backends
+    RRDSET_FLAG_SYNC_CLOCK        = 1 << 13, // if set, microseconds on next data collection will be ignored (the chart will be synced to now)
 } RRDSET_FLAGS;
 
 #ifdef HAVE_C___ATOMIC
@@ -302,7 +304,7 @@ struct rrdset {
     long current_entry;                             // the entry that is currently being updated
                                                     // it goes around in a round-robin fashion
 
-    uint32_t flags;                                 // configuration flags
+    RRDSET_FLAGS flags;                             // configuration flags
 
     int gap_when_lost_iterations_above;             // after how many lost iterations a gap should be stored
                                                     // netdata will interpolate values for gaps lower than this
@@ -501,7 +503,7 @@ struct rrdhost {
     const char *tags;                               // tags for this host
     const char *timezone;                           // the timezone of the host
 
-    uint32_t flags;                                 // flags about this RRDHOST
+    RRDHOST_FLAGS flags;                            // flags about this RRDHOST
 
     int rrd_update_every;                           // the update frequency of the host
     long rrd_history_entries;                       // the number of history entries for the host's charts
@@ -531,6 +533,8 @@ struct rrdhost {
     volatile unsigned int rrdpush_sender_error_shown:1; // 1 when we have logged a communication error
     volatile unsigned int rrdpush_sender_join:1;    // 1 when we have to join the sending thread
 
+    SIMPLE_PATTERN *rrdpush_send_charts_matching;   // pattern to match the charts to be sent
+
     // metrics may be collected asynchronously
     // these synchronize all the threads willing the write to our sending buffer
     netdata_mutex_t rrdpush_sender_buffer_mutex;    // exclusive access to rrdpush_sender_buffer
@@ -646,6 +650,7 @@ extern RRDHOST *rrdhost_find_or_create(
         , unsigned int rrdpush_enabled
         , char *rrdpush_destination
         , char *rrdpush_api_key
+        , char *rrdpush_send_charts_matching
 );
 
 #if defined(NETDATA_INTERNAL_CHECKS) && defined(NETDATA_VERIFY_LOCKS)
diff --git a/src/database/rrddim.c b/src/database/rrddim.c
index 53fa64dc3b..95e485106c 100644
--- a/src/database/rrddim.c
+++ b/src/database/rrddim.c
@@ -49,7 +49,7 @@ inline int rrddim_set_name(RRDSET *st, RRDDIM *rd, const char *name) {
     rd->hash_name = simple_hash(rd->name);
     rrddimvar_rename_all(rd);
     rd->exposed = 0;
-    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
     return 1;
 }
 
@@ -61,7 +61,7 @@ inline int rrddim_set_algorithm(RRDSET *st, RRDDIM *rd, RRD_ALGORITHM algorithm)
     rd->algorithm = algorithm;
     rd->exposed = 0;
     rrdset_flag_set(st, RRDSET_FLAG_HOMEGENEOUS_CHECK);
-    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
     return 1;
 }
 
@@ -73,7 +73,7 @@ inline int rrddim_set_multiplier(RRDSET *st, RRDDIM *rd, collected_number multip
     rd->multiplier = multiplier;
     rd->exposed = 0;
     rrdset_flag_set(st, RRDSET_FLAG_HOMEGENEOUS_CHECK);
-    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
     return 1;
 }
 
@@ -85,7 +85,7 @@ inline int rrddim_set_divisor(RRDSET *st, RRDDIM *rd, collected_number divisor)
     rd->divisor = divisor;
     rd->exposed = 0;
     rrdset_flag_set(st, RRDSET_FLAG_HOMEGENEOUS_CHECK);
-    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
     return 1;
 }
 
@@ -96,7 +96,7 @@ RRDDIM *rrddim_add_custom(RRDSET *st, const char *id, const char *name, collecte
     rrdset_wrlock(st);
 
     rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
-    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
 
     RRDDIM *rd = rrddim_find(st, id);
     if(unlikely(rd)) {
diff --git a/src/database/rrdhost.c b/src/database/rrdhost.c
index 203068408a..183aecfa1d 100644
--- a/src/database/rrdhost.c
+++ b/src/database/rrdhost.c
@@ -122,6 +122,7 @@ RRDHOST *rrdhost_create(const char *hostname,
                         unsigned int rrdpush_enabled,
                         char *rrdpush_destination,
                         char *rrdpush_api_key,
+                        char *rrdpush_send_charts_matching,
                         int is_localhost
 ) {
     debug(D_RRDHOST, "Host '%s': adding with guid '%s'", hostname, guid);
@@ -137,6 +138,7 @@ RRDHOST *rrdhost_create(const char *hostname,
     host->rrdpush_send_enabled     = (rrdpush_enabled && rrdpush_destination && *rrdpush_destination && rrdpush_api_key && *rrdpush_api_key) ? 1 : 0;
     host->rrdpush_send_destination = (host->rrdpush_send_enabled)?strdupz(rrdpush_destination):NULL;
     host->rrdpush_send_api_key     = (host->rrdpush_send_enabled)?strdupz(rrdpush_api_key):NULL;
+    host->rrdpush_send_charts_matching = simple_pattern_create(rrdpush_send_charts_matching, NULL, SIMPLE_PATTERN_EXACT);
 
     host->rrdpush_sender_pipe[0] = -1;
     host->rrdpush_sender_pipe[1] = -1;
@@ -329,6 +331,7 @@ RRDHOST *rrdhost_find_or_create(
         , unsigned int rrdpush_enabled
         , char *rrdpush_destination
         , char *rrdpush_api_key
+        , char *rrdpush_send_charts_matching
 ) {
     debug(D_RRDHOST, "Searching for host '%s' with guid '%s'", hostname, guid);
 
@@ -351,6 +354,7 @@ RRDHOST *rrdhost_find_or_create(
                 , rrdpush_enabled
                 , rrdpush_destination
                 , rrdpush_api_key
+                , rrdpush_send_charts_matching
                 , 0
         );
     }
@@ -463,6 +467,7 @@ void rrd_init(char *hostname) {
             , default_rrdpush_enabled
             , default_rrdpush_destination
             , default_rrdpush_api_key
+            , default_rrdpush_send_charts_matching
             , 1
     );
     rrd_unlock();
@@ -576,6 +581,7 @@ void rrdhost_free(RRDHOST *host) {
     freez(host->health_log_filename);
     freez(host->hostname);
     freez(host->registry_hostname);
+    simple_pattern_free(host->rrdpush_send_charts_matching);
     rrdhost_unlock(host);
     netdata_rwlock_destroy(&host->health_log.alarm_log_rwlock);
     netdata_rwlock_destroy(&host->rrdhost_rwlock);
diff --git a/src/database/rrdset.c b/src/database/rrdset.c
index e54bad7b96..90b3263a80 100644
--- a/src/database/rrdset.c
+++ b/src/database/rrdset.c
@@ -174,27 +174,30 @@ int rrdset_set_name(RRDSET *st, const char *name) {
     if(unlikely(rrdset_index_add_name(host, st) != st))
         error("RRDSET: INTERNAL ERROR: attempted to index duplicate chart name '%s'", st->name);
 
+    rrdset_flag_clear(st, RRDSET_FLAG_BACKEND_SEND);
+    rrdset_flag_clear(st, RRDSET_FLAG_BACKEND_IGNORE);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_SEND);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_IGNORE);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
+
     return 1;
 }
 
 inline void rrdset_is_obsolete(RRDSET *st) {
-    RRDHOST *host = st->rrdhost;
-
     if(unlikely(!(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)))) {
         rrdset_flag_set(st, RRDSET_FLAG_OBSOLETE);
-        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
 
         // the chart will not get more updates (data collection)
         // so, we have to push its definition now
-        if(unlikely(host->rrdpush_send_enabled))
-            rrdset_push_chart_definition(st);
+        rrdset_push_chart_definition_now(st);
     }
 }
 
 inline void rrdset_isnot_obsolete(RRDSET *st) {
     if(unlikely((rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)))) {
         rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
-        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
 
         // the chart will be pushed upstream automatically
         // due to data collection
@@ -480,7 +483,7 @@ RRDSET *rrdset_create_custom(
     RRDSET *st = rrdset_find_on_create(host, fullid);
     if(st) {
         rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
-        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
         return st;
     }
 
@@ -490,7 +493,7 @@ RRDSET *rrdset_create_custom(
     if(st) {
         rrdhost_unlock(host);
         rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
-        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
         return st;
     }
 
@@ -652,7 +655,11 @@ RRDSET *rrdset_create_custom(
     rrdset_flag_clear(st, RRDSET_FLAG_DETAIL);
     rrdset_flag_clear(st, RRDSET_FLAG_DEBUG);
     rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
-    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_clear(st, RRDSET_FLAG_BACKEND_SEND);
+    rrdset_flag_clear(st, RRDSET_FLAG_BACKEND_IGNORE);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_SEND);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_IGNORE);
+    rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
     rrdset_flag_set(st, RRDSET_FLAG_SYNC_CLOCK);
 
     // if(!strcmp(st->id, "disk_util.dm-0")) {
diff --git a/src/database/rrdsetvar.c b/src/database/rrdsetvar.c
index 19a61bbd39..45f661903b 100644
--- a/src/database/rrdsetvar.c
+++ b/src/database/rrdsetvar.c
@@ -183,7 +183,7 @@ void rrdsetvar_custom_chart_variable_set(RRDSETVAR *rs, calculated_number value)
             *v = value;
 
             // mark the chart to be sent upstream
-            rrdset_flag_clear(rs->rrdset, RRDSET_FLAG_EXPOSED_UPSTREAM);
+            rrdset_flag_clear(rs->rrdset, RRDSET_FLAG_UPSTREAM_EXPOSED);
         }
     }
 }
diff --git a/src/streaming/rrdpush.c b/src/streaming/rrdpush.c
index bdd1871496..5d28206049 100644
--- a/src/streaming/rrdpush.c
+++ b/src/streaming/rrdpush.c
@@ -35,11 +35,13 @@ typedef enum {
 unsigned int default_rrdpush_enabled = 0;
 char *default_rrdpush_destination = NULL;
 char *default_rrdpush_api_key = NULL;
+char *default_rrdpush_send_charts_matching = NULL;
 
 int rrdpush_init() {
     default_rrdpush_enabled     = (unsigned int)appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "enabled", default_rrdpush_enabled);
     default_rrdpush_destination = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "destination", "");
     default_rrdpush_api_key     = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "api key", "");
+    default_rrdpush_send_charts_matching      = appconfig_get(&stream_config, CONFIG_SECTION_STREAM, "send charts matching", "*");
     rrdhost_free_orphan_time    = config_get_number(CONFIG_SECTION_GLOBAL, "cleanup orphan hosts after seconds", rrdhost_free_orphan_time);
 
     if(default_rrdpush_enabled && (!default_rrdpush_destination || !*default_rrdpush_destination || !default_rrdpush_api_key || !*default_rrdpush_api_key)) {
@@ -69,11 +71,33 @@ unsigned int remote_clock_resync_iterations = 60;
 #define rrdpush_buffer_lock(host) netdata_mutex_lock(&((host)->rrdpush_sender_buffer_mutex))
 #define rrdpush_buffer_unlock(host) netdata_mutex_unlock(&((host)->rrdpush_sender_buffer_mutex))
 
+static inline int should_send_chart_matching(RRDSET *st) {
+    if(unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED))) {
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_SEND);
+        rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_IGNORE);
+    }
+    else if(!rrdset_flag_check(st, RRDSET_FLAG_UPSTREAM_SEND|RRDSET_FLAG_UPSTREAM_IGNORE)) {
+        RRDHOST *host = st->rrdhost;
+
+        if(simple_pattern_matches(host->rrdpush_send_charts_matching, st->id) ||
+            simple_pattern_matches(host->rrdpush_send_charts_matching, st->name)) {
+            rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_IGNORE);
+            rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_SEND);
+        }
+        else {
+            rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_SEND);
+            rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_IGNORE);
+        }
+    }
+
+    return(rrdset_flag_check(st, RRDSET_FLAG_UPSTREAM_SEND));
+}
+
 // checks if the current chart definition has been sent
 static inline int need_to_send_chart_definition(RRDSET *st) {
     rrdset_check_rdlock(st);
 
-    if(unlikely(!(rrdset_flag_check(st, RRDSET_FLAG_EXPOSED_UPSTREAM))))
+    if(unlikely(!(rrdset_flag_check(st, RRDSET_FLAG_UPSTREAM_EXPOSED))))
         return 1;
 
     RRDDIM *rd;
@@ -93,7 +117,7 @@ static inline int need_to_send_chart_definition(RRDSET *st) {
 static inline void rrdpush_send_chart_definition_nolock(RRDSET *st) {
     RRDHOST *host = st->rrdhost;
 
-    rrdset_flag_set(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+    rrdset_flag_set(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
 
     // send the chart
     buffer_sprintf(
@@ -171,9 +195,12 @@ static inline void rrdpush_send_chart_metrics_nolock(RRDSET *st) {
 
 static void rrdpush_sender_thread_spawn(RRDHOST *host);
 
-void rrdset_push_chart_definition(RRDSET *st) {
+void rrdset_push_chart_definition_now(RRDSET *st) {
     RRDHOST *host = st->rrdhost;
 
+    if(unlikely(!host->rrdpush_send_enabled || !should_send_chart_matching(st)))
+        return;
+
     rrdset_rdlock(st);
     rrdpush_buffer_lock(host);
     rrdpush_send_chart_definition_nolock(st);
@@ -182,11 +209,11 @@ void rrdset_push_chart_definition(RRDSET *st) {
 }
 
 void rrdset_done_push(RRDSET *st) {
-    RRDHOST *host = st->rrdhost;
-
-    if(unlikely(!rrdset_flag_check(st, RRDSET_FLAG_ENABLED)))
+    if(unlikely(!should_send_chart_matching(st)))
         return;
 
+    RRDHOST *host = st->rrdhost;
+
     rrdpush_buffer_lock(host);
 
     if(unlikely(host->rrdpush_send_enabled && !host->rrdpush_sender_spawn))
@@ -269,7 +296,7 @@ static void rrdpush_sender_thread_reset_all_charts(RRDHOST *host) {
 
     RRDSET *st;
     rrdset_foreach_read(st, host) {
-        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+        rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
 
         st->upstream_resync_time = 0;
 
@@ -750,6 +777,7 @@ static int rrdpush_receive(int fd
     int rrdpush_enabled = default_rrdpush_enabled;
     char *rrdpush_destination = default_rrdpush_destination;
     char *rrdpush_api_key = default_rrdpush_api_key;
+    char *rrdpush_send_charts_matching = default_rrdpush_send_charts_matching;
     time_t alarms_delay = 60;
     RRDPUSH_MULTIPLE_CONNECTIONS_STRATEGY rrdpush_multiple_connections_strategy = RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW;
 
@@ -781,6 +809,9 @@ static int rrdpush_receive(int fd
     rrdpush_multiple_connections_strategy = get_multiple_connections_strategy(&stream_config, key, "multiple connections", rrdpush_multiple_connections_strategy);
     rrdpush_multiple_connections_strategy = get_multiple_connections_strategy(&stream_config, machine_guid, "multiple connections", rrdpush_multiple_connections_strategy);
 
+    rrdpush_send_charts_matching = appconfig_get(&stream_config, key, "default proxy send charts matching", rrdpush_send_charts_matching);
+    rrdpush_send_charts_matching = appconfig_get(&stream_config, machine_guid, "proxy send charts matching", rrdpush_send_charts_matching);
+
     tags = appconfig_set_default(&stream_config, machine_guid, "host tags", (tags)?tags:"");
     if(tags && !*tags) tags = NULL;
 
@@ -803,6 +834,7 @@ static int rrdpush_receive(int fd
                 , (unsigned int)(rrdpush_enabled && rrdpush_destination && *rrdpush_destination && rrdpush_api_key && *rrdpush_api_key)
                 , rrdpush_destination
                 , rrdpush_api_key
+                , rrdpush_send_charts_matching
         );
 
     if(!host) {
diff --git a/src/streaming/rrdpush.h b/src/streaming/rrdpush.h
index 84e3f684c3..6fc272b916 100644
--- a/src/streaming/rrdpush.h
+++ b/src/streaming/rrdpush.h
@@ -9,11 +9,12 @@
 extern unsigned int default_rrdpush_enabled;
 extern char *default_rrdpush_destination;
 extern char *default_rrdpush_api_key;
+extern char *default_rrdpush_send_charts_matching;
 extern unsigned int remote_clock_resync_iterations;
 
 extern int rrdpush_init();
 extern void rrdset_done_push(RRDSET *st);
-extern void rrdset_push_chart_definition(RRDSET *st);
+extern void rrdset_push_chart_definition_now(RRDSET *st);
 extern void *rrdpush_sender_thread(void *ptr);
 
 extern int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url);