forked from netdata/netdata
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathstatsd.c
2557 lines (2117 loc) · 98.6 KB
/
statsd.c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// SPDX-License-Identifier: GPL-3.0-or-later
#include "statsd.h"
#define STATSD_CHART_PREFIX "statsd"
#define PLUGIN_STATSD_NAME "statsd.plugin"
// --------------------------------------------------------------------------------------
// #define STATSD_MULTITHREADED 1
#ifdef STATSD_MULTITHREADED
// DO NOT ENABLE MULTITHREADING - IT IS NOT WELL TESTED
#define STATSD_AVL_TREE avl_tree_lock
#define STATSD_AVL_INSERT avl_insert_lock
#define STATSD_AVL_SEARCH avl_search_lock
#define STATSD_AVL_INDEX_INIT { .avl_tree = { NULL, statsd_metric_compare }, .rwlock = AVL_LOCK_INITIALIZER }
#define STATSD_FIRST_PTR_MUTEX netdata_mutex_t first_mutex
#define STATSD_FIRST_PTR_MUTEX_INIT .first_mutex = NETDATA_MUTEX_INITIALIZER
#define STATSD_FIRST_PTR_MUTEX_LOCK(index) netdata_mutex_lock(&((index)->first_mutex))
#define STATSD_FIRST_PTR_MUTEX_UNLOCK(index) netdata_mutex_unlock(&((index)->first_mutex))
#define STATSD_DICTIONARY_OPTIONS DICTIONARY_FLAG_DEFAULT
#else
#define STATSD_AVL_TREE avl_tree
#define STATSD_AVL_INSERT avl_insert
#define STATSD_AVL_SEARCH avl_search
#define STATSD_AVL_INDEX_INIT { .root = NULL, .compar = statsd_metric_compare }
#define STATSD_FIRST_PTR_MUTEX
#define STATSD_FIRST_PTR_MUTEX_INIT
#define STATSD_FIRST_PTR_MUTEX_LOCK(index)
#define STATSD_FIRST_PTR_MUTEX_UNLOCK(index)
#define STATSD_DICTIONARY_OPTIONS DICTIONARY_FLAG_SINGLE_THREADED
#endif
#define STATSD_DECIMAL_DETAIL 1000 // floating point values get multiplied by this, with the same divisor
// --------------------------------------------------------------------------------------------------------------------
// data specific to each metric type
typedef struct statsd_metric_gauge {
LONG_DOUBLE value;
} STATSD_METRIC_GAUGE;
typedef struct statsd_metric_counter { // counter and meter
long long value;
} STATSD_METRIC_COUNTER;
typedef struct statsd_histogram_extensions {
netdata_mutex_t mutex;
// average is stored in metric->last
collected_number last_min;
collected_number last_max;
collected_number last_percentile;
collected_number last_median;
collected_number last_stddev;
collected_number last_sum;
int zeroed;
RRDDIM *rd_min;
RRDDIM *rd_max;
RRDDIM *rd_percentile;
RRDDIM *rd_median;
RRDDIM *rd_stddev;
RRDDIM *rd_sum;
size_t size;
size_t used;
LONG_DOUBLE *values; // dynamic array of values collected
} STATSD_METRIC_HISTOGRAM_EXTENSIONS;
typedef struct statsd_metric_histogram { // histogram and timer
STATSD_METRIC_HISTOGRAM_EXTENSIONS *ext;
} STATSD_METRIC_HISTOGRAM;
typedef struct statsd_metric_set {
DICTIONARY *dict;
size_t unique;
} STATSD_METRIC_SET;
// --------------------------------------------------------------------------------------------------------------------
// this is a metric - for all types of metrics
typedef enum statsd_metric_options {
STATSD_METRIC_OPTION_NONE = 0x00000000, // no options set
STATSD_METRIC_OPTION_SHOW_GAPS_WHEN_NOT_COLLECTED = 0x00000001, // do not update the chart dimension, when this metric is not collected
STATSD_METRIC_OPTION_PRIVATE_CHART_ENABLED = 0x00000002, // render a private chart for this metric
STATSD_METRIC_OPTION_PRIVATE_CHART_CHECKED = 0x00000004, // the metric has been checked if it should get private chart or not
STATSD_METRIC_OPTION_CHART_DIMENSION_COUNT = 0x00000008, // show the count of events for this private chart
STATSD_METRIC_OPTION_CHECKED_IN_APPS = 0x00000010, // set when this metric has been checked against apps
STATSD_METRIC_OPTION_USED_IN_APPS = 0x00000020, // set when this metric is used in apps
STATSD_METRIC_OPTION_CHECKED = 0x00000040, // set when the charting thread checks this metric for use in charts (its usefulness)
STATSD_METRIC_OPTION_USEFUL = 0x00000080, // set when the charting thread finds the metric useful (i.e. used in a chart)
} STATS_METRIC_OPTIONS;
typedef enum statsd_metric_type {
STATSD_METRIC_TYPE_GAUGE,
STATSD_METRIC_TYPE_COUNTER,
STATSD_METRIC_TYPE_METER,
STATSD_METRIC_TYPE_TIMER,
STATSD_METRIC_TYPE_HISTOGRAM,
STATSD_METRIC_TYPE_SET
} STATSD_METRIC_TYPE;
typedef struct statsd_metric {
avl avl; // indexing - has to be first
const char *name; // the name of the metric
uint32_t hash; // hash of the name
STATSD_METRIC_TYPE type;
// metadata about data collection
collected_number events; // the number of times this metric has been collected (never resets)
size_t count; // the number of times this metric has been collected since the last flush
// the actual collected data
union {
STATSD_METRIC_GAUGE gauge;
STATSD_METRIC_COUNTER counter;
STATSD_METRIC_HISTOGRAM histogram;
STATSD_METRIC_SET set;
};
// chart related members
STATS_METRIC_OPTIONS options; // STATSD_METRIC_OPTION_* (bitfield)
char reset; // set to 1 by the charting thread to instruct the collector thread(s) to reset this metric
collected_number last; // the last value sent to netdata
RRDSET *st; // the private chart of this metric
RRDDIM *rd_value; // the dimension of this metric value
RRDDIM *rd_count; // the dimension for the number of events received
// linking, used for walking through all metrics
struct statsd_metric *next;
struct statsd_metric *next_useful;
} STATSD_METRIC;
// --------------------------------------------------------------------------------------------------------------------
// each type of metric has its own index
typedef struct statsd_index {
char *name; // the name of the index of metrics
size_t events; // the number of events processed for this index
size_t metrics; // the number of metrics in this index
size_t useful; // the number of useful metrics in this index
STATSD_AVL_TREE index; // the AVL tree
STATSD_METRIC *first; // the linked list of metrics (new metrics are added in front)
STATSD_METRIC *first_useful; // the linked list of useful metrics (new metrics are added in front)
STATSD_FIRST_PTR_MUTEX; // when mutli-threading is enabled, a lock to protect the linked list
STATS_METRIC_OPTIONS default_options; // default options for all metrics in this index
} STATSD_INDEX;
static int statsd_metric_compare(void* a, void* b);
// --------------------------------------------------------------------------------------------------------------------
// synthetic charts
typedef enum statsd_app_chart_dimension_value_type {
STATSD_APP_CHART_DIM_VALUE_TYPE_EVENTS,
STATSD_APP_CHART_DIM_VALUE_TYPE_LAST,
STATSD_APP_CHART_DIM_VALUE_TYPE_AVERAGE,
STATSD_APP_CHART_DIM_VALUE_TYPE_SUM,
STATSD_APP_CHART_DIM_VALUE_TYPE_MIN,
STATSD_APP_CHART_DIM_VALUE_TYPE_MAX,
STATSD_APP_CHART_DIM_VALUE_TYPE_PERCENTILE,
STATSD_APP_CHART_DIM_VALUE_TYPE_MEDIAN,
STATSD_APP_CHART_DIM_VALUE_TYPE_STDDEV
} STATSD_APP_CHART_DIM_VALUE_TYPE;
typedef struct statsd_app_chart_dimension {
const char *name; // the name of this dimension
const char *metric; // the source metric name of this dimension
uint32_t metric_hash; // hash for fast string comparisons
SIMPLE_PATTERN *metric_pattern; // set when the 'metric' is a simple pattern
collected_number multiplier; // the multipler of the dimension
collected_number divisor; // the divisor of the dimension
RRDDIM_FLAGS flags; // the RRDDIM flags for this dimension
STATSD_APP_CHART_DIM_VALUE_TYPE value_type; // which value to use of the source metric
RRDDIM *rd; // a pointer to the RRDDIM that has been created for this dimension
collected_number *value_ptr; // a pointer to the source metric value
RRD_ALGORITHM algorithm; // the algorithm of this dimension
struct statsd_app_chart_dimension *next; // the next dimension for this chart
} STATSD_APP_CHART_DIM;
typedef struct statsd_app_chart {
const char *source;
const char *id;
const char *name;
const char *title;
const char *family;
const char *context;
const char *units;
long priority;
RRDSET_TYPE chart_type;
STATSD_APP_CHART_DIM *dimensions;
size_t dimensions_count;
size_t dimensions_linked_count;
RRDSET *st;
struct statsd_app_chart *next;
} STATSD_APP_CHART;
typedef struct statsd_app {
const char *name;
SIMPLE_PATTERN *metrics;
STATS_METRIC_OPTIONS default_options;
RRD_MEMORY_MODE rrd_memory_mode;
DICTIONARY *dict;
long rrd_history_entries;
const char *source;
STATSD_APP_CHART *charts;
struct statsd_app *next;
} STATSD_APP;
// --------------------------------------------------------------------------------------------------------------------
// global statsd data
struct collection_thread_status {
int status;
size_t max_sockets;
netdata_thread_t thread;
struct rusage rusage;
RRDSET *st_cpu;
RRDDIM *rd_user;
RRDDIM *rd_system;
};
static struct statsd {
STATSD_INDEX gauges;
STATSD_INDEX counters;
STATSD_INDEX timers;
STATSD_INDEX histograms;
STATSD_INDEX meters;
STATSD_INDEX sets;
size_t unknown_types;
size_t socket_errors;
size_t tcp_socket_connects;
size_t tcp_socket_disconnects;
size_t tcp_socket_connected;
size_t tcp_socket_reads;
size_t tcp_packets_received;
size_t tcp_bytes_read;
size_t udp_socket_reads;
size_t udp_packets_received;
size_t udp_bytes_read;
int enabled;
int update_every;
SIMPLE_PATTERN *charts_for;
size_t tcp_idle_timeout;
collected_number decimal_detail;
size_t private_charts;
size_t max_private_charts;
size_t max_private_charts_hard;
RRD_MEMORY_MODE private_charts_memory_mode;
long private_charts_rrd_history_entries;
unsigned int private_charts_hidden:1;
STATSD_APP *apps;
size_t recvmmsg_size;
size_t histogram_increase_step;
double histogram_percentile;
char *histogram_percentile_str;
int threads;
struct collection_thread_status *collection_threads_status;
LISTEN_SOCKETS sockets;
} statsd = {
.enabled = 1,
.max_private_charts = 200,
.max_private_charts_hard = 1000,
.private_charts_hidden = 0,
.recvmmsg_size = 10,
.decimal_detail = STATSD_DECIMAL_DETAIL,
.gauges = {
.name = "gauge",
.events = 0,
.metrics = 0,
.index = STATSD_AVL_INDEX_INIT,
.default_options = STATSD_METRIC_OPTION_NONE,
.first = NULL,
STATSD_FIRST_PTR_MUTEX_INIT
},
.counters = {
.name = "counter",
.events = 0,
.metrics = 0,
.index = STATSD_AVL_INDEX_INIT,
.default_options = STATSD_METRIC_OPTION_NONE,
.first = NULL,
STATSD_FIRST_PTR_MUTEX_INIT
},
.timers = {
.name = "timer",
.events = 0,
.metrics = 0,
.index = STATSD_AVL_INDEX_INIT,
.default_options = STATSD_METRIC_OPTION_NONE,
.first = NULL,
STATSD_FIRST_PTR_MUTEX_INIT
},
.histograms = {
.name = "histogram",
.events = 0,
.metrics = 0,
.index = STATSD_AVL_INDEX_INIT,
.default_options = STATSD_METRIC_OPTION_NONE,
.first = NULL,
STATSD_FIRST_PTR_MUTEX_INIT
},
.meters = {
.name = "meter",
.events = 0,
.metrics = 0,
.index = STATSD_AVL_INDEX_INIT,
.default_options = STATSD_METRIC_OPTION_NONE,
.first = NULL,
STATSD_FIRST_PTR_MUTEX_INIT
},
.sets = {
.name = "set",
.events = 0,
.metrics = 0,
.index = STATSD_AVL_INDEX_INIT,
.default_options = STATSD_METRIC_OPTION_NONE,
.first = NULL,
STATSD_FIRST_PTR_MUTEX_INIT
},
.tcp_idle_timeout = 600,
.apps = NULL,
.histogram_percentile = 95.0,
.histogram_increase_step = 10,
.threads = 0,
.collection_threads_status = NULL,
.sockets = {
.config = &netdata_config,
.config_section = CONFIG_SECTION_STATSD,
.default_bind_to = "udp:localhost tcp:localhost",
.default_port = STATSD_LISTEN_PORT,
.backlog = STATSD_LISTEN_BACKLOG
},
};
// --------------------------------------------------------------------------------------------------------------------
// statsd index management - add/find metrics
static int statsd_metric_compare(void* a, void* b) {
if(((STATSD_METRIC *)a)->hash < ((STATSD_METRIC *)b)->hash) return -1;
else if(((STATSD_METRIC *)a)->hash > ((STATSD_METRIC *)b)->hash) return 1;
else return strcmp(((STATSD_METRIC *)a)->name, ((STATSD_METRIC *)b)->name);
}
static inline STATSD_METRIC *statsd_metric_index_find(STATSD_INDEX *index, const char *name, uint32_t hash) {
STATSD_METRIC tmp;
tmp.name = name;
tmp.hash = (hash)?hash:simple_hash(tmp.name);
return (STATSD_METRIC *)STATSD_AVL_SEARCH(&index->index, (avl *)&tmp);
}
static inline STATSD_METRIC *statsd_find_or_add_metric(STATSD_INDEX *index, const char *name, STATSD_METRIC_TYPE type) {
debug(D_STATSD, "searching for metric '%s' under '%s'", name, index->name);
uint32_t hash = simple_hash(name);
STATSD_METRIC *m = statsd_metric_index_find(index, name, hash);
if(unlikely(!m)) {
debug(D_STATSD, "Creating new %s metric '%s'", index->name, name);
m = (STATSD_METRIC *)callocz(sizeof(STATSD_METRIC), 1);
m->name = strdupz(name);
m->hash = hash;
m->type = type;
m->options = index->default_options;
if(type == STATSD_METRIC_TYPE_HISTOGRAM || type == STATSD_METRIC_TYPE_TIMER) {
m->histogram.ext = callocz(sizeof(STATSD_METRIC_HISTOGRAM_EXTENSIONS), 1);
netdata_mutex_init(&m->histogram.ext->mutex);
}
STATSD_METRIC *n = (STATSD_METRIC *)STATSD_AVL_INSERT(&index->index, (avl *)m);
if(unlikely(n != m)) {
freez((void *)m->histogram.ext);
freez((void *)m->name);
freez((void *)m);
m = n;
}
else {
STATSD_FIRST_PTR_MUTEX_LOCK(index);
index->metrics++;
m->next = index->first;
index->first = m;
STATSD_FIRST_PTR_MUTEX_UNLOCK(index);
}
}
index->events++;
return m;
}
// --------------------------------------------------------------------------------------------------------------------
// statsd parsing numbers
static inline LONG_DOUBLE statsd_parse_float(const char *v, LONG_DOUBLE def) {
LONG_DOUBLE value;
if(likely(v && *v)) {
char *e = NULL;
value = str2ld(v, &e);
if(unlikely(e && *e))
error("STATSD: excess data '%s' after value '%s'", e, v);
}
else
value = def;
return value;
}
static inline LONG_DOUBLE statsd_parse_sampling_rate(const char *v) {
LONG_DOUBLE sampling_rate = statsd_parse_float(v, 1.0);
if(unlikely(isless(sampling_rate, 0.001))) sampling_rate = 0.001;
if(unlikely(isgreater(sampling_rate, 1.0))) sampling_rate = 1.0;
return sampling_rate;
}
static inline long long statsd_parse_int(const char *v, long long def) {
long long value;
if(likely(v && *v)) {
char *e = NULL;
value = str2ll(v, &e);
if(unlikely(e && *e))
error("STATSD: excess data '%s' after value '%s'", e, v);
}
else
value = def;
return value;
}
// --------------------------------------------------------------------------------------------------------------------
// statsd processors per metric type
static inline void statsd_reset_metric(STATSD_METRIC *m) {
m->reset = 0;
m->count = 0;
}
static inline int value_is_zinit(const char *value) {
return (value && *value == 'z' && *++value == 'i' && *++value == 'n' && *++value == 'i' && *++value == 't' && *++value == '\0');
}
#define is_metric_checked(m) ((m)->options & STATSD_METRIC_OPTION_CHECKED)
#define is_metric_useful_for_collection(m) (!is_metric_checked(m) || ((m)->options & STATSD_METRIC_OPTION_USEFUL))
static inline void statsd_process_gauge(STATSD_METRIC *m, const char *value, const char *sampling) {
if(!is_metric_useful_for_collection(m)) return;
if(unlikely(!value || !*value)) {
error("STATSD: metric '%s' of type gauge, with empty value is ignored.", m->name);
return;
}
if(unlikely(m->reset)) {
// no need to reset anything specific for gauges
statsd_reset_metric(m);
}
if(unlikely(value_is_zinit(value))) {
// magic loading of metric, without affecting anything
}
else {
if (unlikely(*value == '+' || *value == '-'))
m->gauge.value += statsd_parse_float(value, 1.0) / statsd_parse_sampling_rate(sampling);
else
m->gauge.value = statsd_parse_float(value, 1.0);
m->events++;
m->count++;
}
}
static inline void statsd_process_counter_or_meter(STATSD_METRIC *m, const char *value, const char *sampling) {
if(!is_metric_useful_for_collection(m)) return;
// we accept empty values for counters
if(unlikely(m->reset)) statsd_reset_metric(m);
if(unlikely(value_is_zinit(value))) {
// magic loading of metric, without affecting anything
}
else {
m->counter.value += llrintl((LONG_DOUBLE) statsd_parse_int(value, 1) / statsd_parse_sampling_rate(sampling));
m->events++;
m->count++;
}
}
#define statsd_process_counter(m, value, sampling) statsd_process_counter_or_meter(m, value, sampling)
#define statsd_process_meter(m, value, sampling) statsd_process_counter_or_meter(m, value, sampling)
static inline void statsd_process_histogram_or_timer(STATSD_METRIC *m, const char *value, const char *sampling, const char *type) {
if(!is_metric_useful_for_collection(m)) return;
if(unlikely(!value || !*value)) {
error("STATSD: metric of type %s, with empty value is ignored.", type);
return;
}
if(unlikely(m->reset)) {
m->histogram.ext->used = 0;
statsd_reset_metric(m);
}
if(unlikely(value_is_zinit(value))) {
// magic loading of metric, without affecting anything
}
else {
LONG_DOUBLE v = statsd_parse_float(value, 1.0);
LONG_DOUBLE sampling_rate = statsd_parse_sampling_rate(sampling);
if(unlikely(isless(sampling_rate, 0.01))) sampling_rate = 0.01;
if(unlikely(isgreater(sampling_rate, 1.0))) sampling_rate = 1.0;
long long samples = llrintl(1.0 / sampling_rate);
while(samples-- > 0) {
if(unlikely(m->histogram.ext->used == m->histogram.ext->size)) {
netdata_mutex_lock(&m->histogram.ext->mutex);
m->histogram.ext->size += statsd.histogram_increase_step;
m->histogram.ext->values = reallocz(m->histogram.ext->values, sizeof(LONG_DOUBLE) * m->histogram.ext->size);
netdata_mutex_unlock(&m->histogram.ext->mutex);
}
m->histogram.ext->values[m->histogram.ext->used++] = v;
}
m->events++;
m->count++;
}
}
#define statsd_process_timer(m, value, sampling) statsd_process_histogram_or_timer(m, value, sampling, "timer")
#define statsd_process_histogram(m, value, sampling) statsd_process_histogram_or_timer(m, value, sampling, "histogram")
static inline void statsd_process_set(STATSD_METRIC *m, const char *value) {
if(!is_metric_useful_for_collection(m)) return;
if(unlikely(!value || !*value)) {
error("STATSD: metric of type set, with empty value is ignored.");
return;
}
if(unlikely(m->reset)) {
if(likely(m->set.dict)) {
dictionary_destroy(m->set.dict);
m->set.dict = NULL;
}
statsd_reset_metric(m);
}
if (unlikely(!m->set.dict)) {
m->set.dict = dictionary_create(STATSD_DICTIONARY_OPTIONS | DICTIONARY_FLAG_VALUE_LINK_DONT_CLONE);
m->set.unique = 0;
}
if(unlikely(value_is_zinit(value))) {
// magic loading of metric, without affecting anything
}
else {
void *t = dictionary_get(m->set.dict, value);
if (unlikely(!t)) {
dictionary_set(m->set.dict, value, NULL, 1);
m->set.unique++;
}
m->events++;
m->count++;
}
}
// --------------------------------------------------------------------------------------------------------------------
// statsd parsing
static void statsd_process_metric(const char *name, const char *value, const char *type, const char *sampling, const char *tags) {
(void)tags;
debug(D_STATSD, "STATSD: raw metric '%s', value '%s', type '%s', sampling '%s', tags '%s'", name?name:"(null)", value?value:"(null)", type?type:"(null)", sampling?sampling:"(null)", tags?tags:"(null)");
if(unlikely(!name || !*name)) return;
if(unlikely(!type || !*type)) type = "m";
char t0 = type[0], t1 = type[1];
if(unlikely(t0 == 'g' && t1 == '\0')) {
statsd_process_gauge(
statsd_find_or_add_metric(&statsd.gauges, name, STATSD_METRIC_TYPE_GAUGE),
value, sampling);
}
else if(unlikely((t0 == 'c' || t0 == 'C') && t1 == '\0')) {
// etsy/statsd uses 'c'
// brubeck uses 'C'
statsd_process_counter(
statsd_find_or_add_metric(&statsd.counters, name, STATSD_METRIC_TYPE_COUNTER),
value, sampling);
}
else if(unlikely(t0 == 'm' && t1 == '\0')) {
statsd_process_meter(
statsd_find_or_add_metric(&statsd.meters, name, STATSD_METRIC_TYPE_METER),
value, sampling);
}
else if(unlikely(t0 == 'h' && t1 == '\0')) {
statsd_process_histogram(
statsd_find_or_add_metric(&statsd.histograms, name, STATSD_METRIC_TYPE_HISTOGRAM),
value, sampling);
}
else if(unlikely(t0 == 's' && t1 == '\0')) {
statsd_process_set(
statsd_find_or_add_metric(&statsd.sets, name, STATSD_METRIC_TYPE_SET),
value);
}
else if(unlikely(t0 == 'm' && t1 == 's' && type[2] == '\0')) {
statsd_process_timer(
statsd_find_or_add_metric(&statsd.timers, name, STATSD_METRIC_TYPE_TIMER),
value, sampling);
}
else {
statsd.unknown_types++;
error("STATSD: metric '%s' with value '%s' is sent with unknown metric type '%s'", name, value?value:"", type);
}
}
static inline const char *statsd_parse_skip_up_to(const char *s, char d1, char d2) {
char c;
for(c = *s; c && c != d1 && c != d2 && c != '\r' && c != '\n'; c = *++s) ;
return s;
}
const char *statsd_parse_skip_spaces(const char *s) {
char c;
for(c = *s; c && ( c == ' ' || c == '\t' || c == '\r' || c == '\n' ); c = *++s) ;
return s;
}
static inline const char *statsd_parse_field_trim(const char *start, char *end) {
if(unlikely(!start)) {
start = end;
return start;
}
while(start <= end && (*start == ' ' || *start == '\t'))
start++;
*end = '\0';
end--;
while(end >= start && (*end == ' ' || *end == '\t'))
*end-- = '\0';
return start;
}
static inline size_t statsd_process(char *buffer, size_t size, int require_newlines) {
buffer[size] = '\0';
debug(D_STATSD, "RECEIVED: %zu bytes: '%s'", size, buffer);
const char *s = buffer;
while(*s) {
const char *name = NULL, *value = NULL, *type = NULL, *sampling = NULL, *tags = NULL;
char *name_end = NULL, *value_end = NULL, *type_end = NULL, *sampling_end = NULL, *tags_end = NULL;
s = name_end = (char *)statsd_parse_skip_up_to(name = s, ':', '|');
if(name == name_end) {
s = statsd_parse_skip_spaces(s);
continue;
}
if(likely(*s == ':'))
s = value_end = (char *) statsd_parse_skip_up_to(value = ++s, '|', '|');
if(likely(*s == '|'))
s = type_end = (char *) statsd_parse_skip_up_to(type = ++s, '|', '@');
if(likely(*s == '|' || *s == '@')) {
s = sampling_end = (char *) statsd_parse_skip_up_to(sampling = ++s, '|', '#');
if(*sampling == '@') sampling++;
}
if(likely(*s == '|' || *s == '#')) {
s = tags_end = (char *) statsd_parse_skip_up_to(tags = ++s, '|', '|');
if(*tags == '#') tags++;
}
// skip everything until the end of the line
while(*s && *s != '\n') s++;
if(unlikely(require_newlines && *s != '\n' && s > buffer)) {
// move the remaining data to the beginning
size -= (name - buffer);
memmove(buffer, name, size);
return size;
}
else
s = statsd_parse_skip_spaces(s);
statsd_process_metric(
statsd_parse_field_trim(name, name_end)
, statsd_parse_field_trim(value, value_end)
, statsd_parse_field_trim(type, type_end)
, statsd_parse_field_trim(sampling, sampling_end)
, statsd_parse_field_trim(tags, tags_end)
);
}
return 0;
}
// --------------------------------------------------------------------------------------------------------------------
// statsd pollfd interface
#define STATSD_TCP_BUFFER_SIZE 65536 // minimize tcp reads
#define STATSD_UDP_BUFFER_SIZE 9000 // this should be up to MTU
typedef enum {
STATSD_SOCKET_DATA_TYPE_TCP,
STATSD_SOCKET_DATA_TYPE_UDP
} STATSD_SOCKET_DATA_TYPE;
struct statsd_tcp {
STATSD_SOCKET_DATA_TYPE type;
size_t size;
size_t len;
char buffer[];
};
#ifdef HAVE_RECVMMSG
struct statsd_udp {
int *running;
STATSD_SOCKET_DATA_TYPE type;
size_t size;
struct iovec *iovecs;
struct mmsghdr *msgs;
};
#else
struct statsd_udp {
int *running;
STATSD_SOCKET_DATA_TYPE type;
char buffer[STATSD_UDP_BUFFER_SIZE];
};
#endif
// new TCP client connected
static void *statsd_add_callback(POLLINFO *pi, short int *events, void *data) {
(void)pi;
(void)data;
*events = POLLIN;
struct statsd_tcp *t = (struct statsd_tcp *)callocz(sizeof(struct statsd_tcp) + STATSD_TCP_BUFFER_SIZE, 1);
t->type = STATSD_SOCKET_DATA_TYPE_TCP;
t->size = STATSD_TCP_BUFFER_SIZE - 1;
statsd.tcp_socket_connects++;
statsd.tcp_socket_connected++;
return t;
}
// TCP client disconnected
static void statsd_del_callback(POLLINFO *pi) {
struct statsd_tcp *t = pi->data;
if(likely(t)) {
if(t->type == STATSD_SOCKET_DATA_TYPE_TCP) {
if(t->len != 0) {
statsd.socket_errors++;
error("STATSD: client is probably sending unterminated metrics. Closed socket left with '%s'. Trying to process it.", t->buffer);
statsd_process(t->buffer, t->len, 0);
}
statsd.tcp_socket_disconnects++;
statsd.tcp_socket_connected--;
}
else
error("STATSD: internal error: received socket data type is %d, but expected %d", (int)t->type, (int)STATSD_SOCKET_DATA_TYPE_TCP);
freez(t);
}
}
// Receive data
static int statsd_rcv_callback(POLLINFO *pi, short int *events) {
*events = POLLIN;
int fd = pi->fd;
switch(pi->socktype) {
case SOCK_STREAM: {
struct statsd_tcp *d = (struct statsd_tcp *)pi->data;
if(unlikely(!d)) {
error("STATSD: internal error: expected TCP data pointer is NULL");
statsd.socket_errors++;
return -1;
}
#ifdef NETDATA_INTERNAL_CHECKS
if(unlikely(d->type != STATSD_SOCKET_DATA_TYPE_TCP)) {
error("STATSD: internal error: socket data type should be %d, but it is %d", (int)STATSD_SOCKET_DATA_TYPE_TCP, (int)d->type);
statsd.socket_errors++;
return -1;
}
#endif
int ret = 0;
ssize_t rc;
do {
rc = recv(fd, &d->buffer[d->len], d->size - d->len, MSG_DONTWAIT);
if (rc < 0) {
// read failed
if (errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR) {
error("STATSD: recv() on TCP socket %d failed.", fd);
statsd.socket_errors++;
ret = -1;
}
}
else if (!rc) {
// connection closed
debug(D_STATSD, "STATSD: client disconnected.");
ret = -1;
}
else {
// data received
d->len += rc;
statsd.tcp_socket_reads++;
statsd.tcp_bytes_read += rc;
}
if(likely(d->len > 0)) {
statsd.tcp_packets_received++;
d->len = statsd_process(d->buffer, d->len, 1);
}
if(unlikely(ret == -1))
return -1;
} while (rc != -1);
break;
}
case SOCK_DGRAM: {
struct statsd_udp *d = (struct statsd_udp *)pi->data;
if(unlikely(!d)) {
error("STATSD: internal error: expected UDP data pointer is NULL");
statsd.socket_errors++;
return -1;
}
#ifdef NETDATA_INTERNAL_CHECKS
if(unlikely(d->type != STATSD_SOCKET_DATA_TYPE_UDP)) {
error("STATSD: internal error: socket data should be %d, but it is %d", (int)d->type, (int)STATSD_SOCKET_DATA_TYPE_UDP);
statsd.socket_errors++;
return -1;
}
#endif
#ifdef HAVE_RECVMMSG
ssize_t rc;
do {
rc = recvmmsg(fd, d->msgs, (unsigned int)d->size, MSG_DONTWAIT, NULL);
if (rc < 0) {
// read failed
if (errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR) {
error("STATSD: recvmmsg() on UDP socket %d failed.", fd);
statsd.socket_errors++;
return -1;
}
} else if (rc) {
// data received
statsd.udp_socket_reads++;
statsd.udp_packets_received += rc;
size_t i;
for (i = 0; i < (size_t)rc; ++i) {
size_t len = (size_t)d->msgs[i].msg_len;
statsd.udp_bytes_read += len;
statsd_process(d->msgs[i].msg_hdr.msg_iov->iov_base, len, 0);
}
}
} while (rc != -1);
#else // !HAVE_RECVMMSG
ssize_t rc;
do {
rc = recv(fd, d->buffer, STATSD_UDP_BUFFER_SIZE - 1, MSG_DONTWAIT);
if (rc < 0) {
// read failed
if (errno != EWOULDBLOCK && errno != EAGAIN && errno != EINTR) {
error("STATSD: recv() on UDP socket %d failed.", fd);
statsd.socket_errors++;
return -1;
}
} else if (rc) {
// data received
statsd.udp_socket_reads++;
statsd.udp_packets_received++;
statsd.udp_bytes_read += rc;
statsd_process(d->buffer, (size_t) rc, 0);
}
} while (rc != -1);
#endif
break;
}
default: {
error("STATSD: internal error: unknown socktype %d on socket %d", pi->socktype, fd);
statsd.socket_errors++;
return -1;
}
}
return 0;
}
static int statsd_snd_callback(POLLINFO *pi, short int *events) {
(void)pi;
(void)events;
error("STATSD: snd_callback() called, but we never requested to send data to statsd clients.");
return -1;
}
static void statsd_timer_callback(void *timer_data) {
struct collection_thread_status *status = timer_data;
getrusage(RUSAGE_THREAD, &status->rusage);
}
// --------------------------------------------------------------------------------------------------------------------
// statsd child thread to collect metrics from network
void statsd_collector_thread_cleanup(void *data) {
struct statsd_udp *d = data;
*d->running = 0;
info("cleaning up...");
#ifdef HAVE_RECVMMSG
size_t i;
for (i = 0; i < d->size; i++)
freez(d->iovecs[i].iov_base);
freez(d->iovecs);
freez(d->msgs);
#endif
freez(d);
}
void *statsd_collector_thread(void *ptr) {
struct collection_thread_status *status = ptr;
status->status = 1;
info("STATSD collector thread started with taskid %d", gettid());
struct statsd_udp *d = callocz(sizeof(struct statsd_udp), 1);
d->running = &status->status;
netdata_thread_cleanup_push(statsd_collector_thread_cleanup, d);
#ifdef HAVE_RECVMMSG
d->type = STATSD_SOCKET_DATA_TYPE_UDP;
d->size = statsd.recvmmsg_size;
d->iovecs = callocz(sizeof(struct iovec), d->size);
d->msgs = callocz(sizeof(struct mmsghdr), d->size);