rename variable "opentsdb host tags" to "host tags"; move it from the backends code to the rrdhost code; allow streaming to pass over the host tags; allow host tags to be overwritten in [MACHINE_GUID] sections of stream.conf; fixes #2304
Costa Tsaousis (ktsaou) committed
Jun 17, 2017 at 22:10 UTC
12cd57e2ab559c42ca80d40eb5cf4f1632c00a3d
5 files changed
+46
-24
src/backends.c
+4
-10
@@ -204,8 +204,6 @@ static inline int process_graphite_response(BUFFER *b) {
204
// ----------------------------------------------------------------------------
205
// opentsdb backend
206
207
-static const char *opentsdb_host_tags = NULL;
208
-
207
static inline int format_dimension_collected_opentsdb_telnet(
208
BUFFER *b // the buffer to write data to
209
, const char *prefix // the prefix to use
@@ -231,8 +229,8 @@ static inline int format_dimension_collected_opentsdb_telnet(
229
, (uint32_t)rd->last_collected_time.tv_sec
230
, rd->last_collected_value
231
, hostname
234
- , (opentsdb_host_tags)?" ":""
235
- , (opentsdb_host_tags)?opentsdb_host_tags:""
232
+ , (host->tags)?" ":""
233
+ , (host->tags)?host->tags:""
234
);
235
236
return 1;
@@ -264,8 +262,8 @@ static inline int format_dimension_stored_opentsdb_telnet(
262
, (uint32_t) before
263
, value
264
, hostname
267
- , (opentsdb_host_tags)?" ":""
268
- , (opentsdb_host_tags)?opentsdb_host_tags:""
265
+ , (host->tags)?" ":""
266
+ , (host->tags)?host->tags:""
267
);
268
269
return 1;
@@ -511,10 +509,6 @@ void *backends_main(void *ptr) {
509
default_port = 4242;
510
backend_response_checker = process_opentsdb_response;
511
514
- const char *tags = config_get(CONFIG_SECTION_BACKEND, "opentsdb host tags", "");
515
- if(tags && *tags)
516
- opentsdb_host_tags = tags;
517
-
512
if(options & BACKEND_SOURCE_DATA_AS_COLLECTED)
513
backend_request_formatter = format_dimension_collected_opentsdb_telnet;
514
else
src/main.c
+3
@@ -410,6 +410,9 @@ static void backwards_compatible_config() {
410
411
config_move(CONFIG_SECTION_GLOBAL, "web files group",
412
CONFIG_SECTION_WEB, "web files group");
413
+
414
+ config_move(CONFIG_SECTION_BACKEND, "opentsdb host tags",
415
+ CONFIG_SECTION_BACKEND, "host tags");
416
}
417
418
static void get_netdata_configured_variables() {
src/rrd.h
+3
-1
@@ -378,7 +378,8 @@ struct rrdhost {
378
char machine_guid[GUID_LEN + 1]; // the unique ID of this host
379
uint32_t hash_machine_guid; // the hash of the unique ID
380
381
- char *os; // the O/S type of the host
381
+ const char *os; // the O/S type of the host
382
+ const char *tags; // tags for this host
383
384
uint32_t flags; // flags about this RRDHOST
385
@@ -498,6 +499,7 @@ extern RRDHOST *rrdhost_find_or_create(
499
, const char *registry_hostname
500
, const char *guid
501
, const char *os
502
+ , const char *tags
503
, int update_every
504
, long history
505
, RRD_MEMORY_MODE mode
src/rrdhost.c
+15
-2
@@ -58,6 +58,11 @@ RRDHOST *rrdhost_find_by_hostname(const char *hostname, uint32_t hash) {
58
// ----------------------------------------------------------------------------
59
// RRDHOST - internal helpers
60
61
+static inline void rrdhost_init_tags(RRDHOST *host, const char *tags) {
62
+ freez((void *)host->tags);
63
+ host->tags = strdupz(tags);
64
+}
65
+
66
static inline void rrdhost_init_hostname(RRDHOST *host, const char *hostname) {
67
freez(host->hostname);
68
host->hostname = strdupz(hostname);
@@ -65,7 +70,7 @@ static inline void rrdhost_init_hostname(RRDHOST *host, const char *hostname) {
70
}
71
72
static inline void rrdhost_init_os(RRDHOST *host, const char *os) {
68
- freez(host->os);
73
+ freez((void *)host->os);
74
host->os = strdupz(os?os:"unknown");
75
}
76
@@ -83,6 +88,7 @@ RRDHOST *rrdhost_create(const char *hostname,
88
const char *registry_hostname,
89
const char *guid,
90
const char *os,
91
+ const char *tags,
92
int update_every,
93
long entries,
94
RRD_MEMORY_MODE memory_mode,
@@ -116,6 +122,7 @@ RRDHOST *rrdhost_create(const char *hostname,
122
rrdhost_init_hostname(host, hostname);
123
rrdhost_init_machine_guid(host, guid);
124
rrdhost_init_os(host, os);
125
+ rrdhost_init_tags(host, tags);
126
host->registry_hostname = strdupz((registry_hostname && *registry_hostname)?registry_hostname:hostname);
127
128
avl_init_lock(&(host->rrdset_root_index), rrdset_compare);
@@ -233,6 +240,7 @@ RRDHOST *rrdhost_create(const char *hostname,
240
else {
241
info("Host '%s' (at registry as '%s') with guid '%s' initialized"
242
", os %s"
243
+ ", tags '%s'"
244
", update every %d"
245
", memory mode %s"
246
", history entries %ld"
@@ -248,6 +256,7 @@ RRDHOST *rrdhost_create(const char *hostname,
256
, host->registry_hostname
257
, host->machine_guid
258
, host->os
259
+ , (host->tags)?host->tags:""
260
, host->rrd_update_every
261
, rrd_memory_mode_name(host->rrd_memory_mode)
262
, host->rrd_history_entries
@@ -273,6 +282,7 @@ RRDHOST *rrdhost_find_or_create(
282
, const char *registry_hostname
283
, const char *guid
284
, const char *os
285
+ , const char *tags
286
, int update_every
287
, long history
288
, RRD_MEMORY_MODE mode
@@ -291,6 +301,7 @@ RRDHOST *rrdhost_find_or_create(
301
, registry_hostname
302
, guid
303
, os
304
+ , tags
305
, update_every
306
, history
307
, mode
@@ -380,6 +391,7 @@ void rrd_init(char *hostname) {
391
, registry_get_this_machine_hostname()
392
, registry_get_this_machine_guid()
393
, os_type
394
+ , config_get(CONFIG_SECTION_BACKEND, "opentsdb host tags", "")
395
, default_rrd_update_every
396
, default_rrd_history_entries
397
, default_rrd_memory_mode
@@ -479,7 +491,8 @@ void rrdhost_free(RRDHOST *host) {
491
// ------------------------------------------------------------------------
492
// free it
493
482
- freez(host->os);
494
+ freez((void *)host->tags);
495
+ freez((void *)host->os);
496
freez(host->cache_dir);
497
freez(host->varlib_dir);
498
freez(host->rrdpush_api_key);
src/rrdpush.c
+21
-11
@@ -119,7 +119,7 @@ static inline void send_chart_metrics(RRDSET *st) {
119
buffer_strcat(st->rrdhost->rrdpush_buffer, "END\n");
120
}
121
122
-void rrdpush_sender_thread_spawn(RRDHOST *host);
122
+static void rrdpush_sender_thread_spawn(RRDHOST *host);
123
124
void rrdset_done_push(RRDSET *st) {
125
RRDHOST *host = st->rrdhost;
@@ -305,17 +305,19 @@ void *rrdpush_sender_thread(void *ptr) {
305
306
info("STREAM %s [send to %s]: initializing communication...", host->hostname, connected_to);
307
308
- char http[1000 + 1];
309
- snprintfz(http, 1000,
310
- "STREAM key=%s&hostname=%s®istry_hostname=%s&machine_guid=%s&os=%s&update_every=%d HTTP/1.1\r\n"
308
+ #define HTTP_HEADER_SIZE 8192
309
+ char http[HTTP_HEADER_SIZE + 1];
310
+ snprintfz(http, HTTP_HEADER_SIZE,
311
+ "STREAM key=%s&hostname=%s®istry_hostname=%s&machine_guid=%s&update_every=%d&os=%s&tags=%s HTTP/1.1\r\n"
312
"User-Agent: netdata-push-service/%s\r\n"
313
"Accept: */*\r\n\r\n"
314
, host->rrdpush_api_key
315
, host->hostname
316
, host->registry_hostname
317
, host->machine_guid
317
- , host->os
318
, default_rrd_update_every
319
+ , host->os
320
+ , (host->tags)?host->tags:""
321
, program_version
322
);
323
@@ -329,7 +331,7 @@ void *rrdpush_sender_thread(void *ptr) {
331
332
info("STREAM %s [send to %s]: waiting response from remote netdata...", host->hostname, connected_to);
333
332
- if(recv_timeout(host->rrdpush_socket, http, 1000, 0, timeout) == -1) {
334
+ if(recv_timeout(host->rrdpush_socket, http, HTTP_HEADER_SIZE, 0, timeout) == -1) {
335
close(host->rrdpush_socket);
336
host->rrdpush_socket = -1;
337
error("STREAM %s [send to %s]: failed to initialize communication", host->hostname, connected_to);
@@ -509,7 +511,7 @@ cleanup:
511
// ----------------------------------------------------------------------------
512
// rrdpush receiver thread
513
512
-int rrdpush_receive(int fd, const char *key, const char *hostname, const char *registry_hostname, const char *machine_guid, const char *os, int update_every, char *client_ip, char *client_port) {
514
+static int rrdpush_receive(int fd, const char *key, const char *hostname, const char *registry_hostname, const char *machine_guid, const char *os, const char *tags, int update_every, char *client_ip, char *client_port) {
515
RRDHOST *host;
516
int history = default_rrd_history_entries;
517
RRD_MEMORY_MODE mode = default_rrd_memory_mode;
@@ -544,6 +546,8 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *r
546
rrdpush_api_key = appconfig_get(&stream_config, key, "default proxy api key", rrdpush_api_key);
547
rrdpush_api_key = appconfig_get(&stream_config, machine_guid, "proxy api key", rrdpush_api_key);
548
549
+ tags = appconfig_get(&stream_config, machine_guid, "host tags", tags);
550
+
551
if(!strcmp(machine_guid, "localhost"))
552
host = localhost;
553
else
@@ -552,6 +556,7 @@ int rrdpush_receive(int fd, const char *key, const char *hostname, const char *r
556
, registry_hostname
557
, machine_guid
558
, os
559
+ , tags
560
, update_every
561
, history
562
, mode
@@ -654,12 +659,13 @@ struct rrdpush_thread {
659
char *registry_hostname;
660
char *machine_guid;
661
char *os;
662
+ char *tags;
663
char *client_ip;
664
char *client_port;
665
int update_every;
666
};
667
662
-void *rrdpush_receiver_thread(void *ptr) {
668
+static void *rrdpush_receiver_thread(void *ptr) {
669
struct rrdpush_thread *rpt = (struct rrdpush_thread *)ptr;
670
671
if (pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
@@ -670,7 +676,7 @@ void *rrdpush_receiver_thread(void *ptr) {
676
677
678
info("STREAM %s [%s]:%s: receive thread created (task id %d)", rpt->hostname, rpt->client_ip, rpt->client_port, gettid());
673
- rrdpush_receive(rpt->fd, rpt->key, rpt->hostname, rpt->registry_hostname, rpt->machine_guid, rpt->os, rpt->update_every, rpt->client_ip, rpt->client_port);
679
+ rrdpush_receive(rpt->fd, rpt->key, rpt->hostname, rpt->registry_hostname, rpt->machine_guid, rpt->os, rpt->tags, rpt->update_every, rpt->client_ip, rpt->client_port);
680
info("STREAM %s [receive from [%s]:%s]: receive thread ended (task id %d)", rpt->hostname, rpt->client_ip, rpt->client_port, gettid());
681
682
freez(rpt->key);
@@ -678,6 +684,7 @@ void *rrdpush_receiver_thread(void *ptr) {
684
freez(rpt->registry_hostname);
685
freez(rpt->machine_guid);
686
freez(rpt->os);
687
+ freez(rpt->tags);
688
freez(rpt->client_ip);
689
freez(rpt->client_port);
690
freez(rpt);
@@ -686,7 +693,7 @@ void *rrdpush_receiver_thread(void *ptr) {
693
return NULL;
694
}
695
689
-void rrdpush_sender_thread_spawn(RRDHOST *host) {
696
+static void rrdpush_sender_thread_spawn(RRDHOST *host) {
697
rrdhost_wrlock(host);
698
699
if(!host->rrdpush_spawn) {
@@ -707,7 +714,7 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
714
715
info("STREAM [receive from [%s]:%s]: new client connection.", w->client_ip, w->client_port);
716
710
- char *key = NULL, *hostname = NULL, *registry_hostname = NULL, *machine_guid = NULL, *os = "unknown";
717
+ char *key = NULL, *hostname = NULL, *registry_hostname = NULL, *machine_guid = NULL, *os = "unknown", *tags = NULL;
718
int update_every = default_rrd_update_every;
719
char buf[GUID_LEN + 1];
720
@@ -731,6 +738,8 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
738
update_every = (int)strtoul(value, NULL, 0);
739
else if(!strcmp(name, "os"))
740
os = value;
741
+ else if(!strcmp(name, "tags"))
742
+ tags = value;
743
else
744
info("STREAM [receive from [%s]:%s]: request has parameter '%s' = '%s', which is not used.", w->client_ip, w->client_port, key, value);
745
}
@@ -791,6 +800,7 @@ int rrdpush_receiver_thread_spawn(RRDHOST *host, struct web_client *w, char *url
800
rpt->registry_hostname = strdupz((registry_hostname && *registry_hostname)?registry_hostname:hostname);
801
rpt->machine_guid = strdupz(machine_guid);
802
rpt->os = strdupz(os);
803
+ rpt->tags = (tags)?strdupz(tags):NULL;
804
rpt->client_ip = strdupz(w->client_ip);
805
rpt->client_port = strdupz(w->client_port);
806
rpt->update_every = update_every;