daemon/plugin.c: fix two more stylistic issues
[collectd.git] / src / write_prometheus.c
1 /**
2  * collectd - src/write_prometheus.c
3  * Copyright (C) 2016       Florian octo Forster
4  *
5  * Permission is hereby granted, free of charge, to any person obtaining a copy
6  * of this software and associated documentation files (the "Software"), to deal
7  * in the Software without restriction, including without limitation the rights
8  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9  * copies of the Software, and to permit persons to whom the Software is
10  * furnished to do so, subject to the following conditions:
11  *
12  * The above copyright notice and this permission notice shall be included in
13  * all copies or substantial portions of the Software.
14  *
15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21  * SOFTWARE.
22  *
23  * Authors:
24  *   Florian octo Forster <octo at collectd.org>
25  */
26
27 #include "collectd.h"
28
29 #include "common.h"
30 #include "plugin.h"
31 #include "utils_avltree.h"
32 #include "utils_complain.h"
33 #include "utils_time.h"
34
35 #include "prometheus.pb-c.h"
36
37 #include <microhttpd.h>
38
39 #include <netdb.h>
40 #include <sys/socket.h>
41 #include <sys/types.h>
42
43 #ifndef PROMETHEUS_DEFAULT_STALENESS_DELTA
44 #define PROMETHEUS_DEFAULT_STALENESS_DELTA TIME_T_TO_CDTIME_T_STATIC(300)
45 #endif
46
47 #define VARINT_UINT32_BYTES 5
48
49 #define CONTENT_TYPE_PROTO                                                     \
50   "application/vnd.google.protobuf; proto=io.prometheus.client.MetricFamily; " \
51   "encoding=delimited"
52 #define CONTENT_TYPE_TEXT "text/plain; version=0.0.4"
53
54 static c_avl_tree_t *metrics;
55 static pthread_mutex_t metrics_lock = PTHREAD_MUTEX_INITIALIZER;
56
57 static char *httpd_host = NULL;
58 static unsigned short httpd_port = 9103;
59 static struct MHD_Daemon *httpd;
60
61 static cdtime_t staleness_delta = PROMETHEUS_DEFAULT_STALENESS_DELTA;
62
63 /* Unfortunately, protoc-c doesn't export its implementation of varint, so we
64  * need to implement our own. */
65 static size_t varint(uint8_t buffer[static VARINT_UINT32_BYTES],
66                      uint32_t value) {
67   for (size_t i = 0; i < VARINT_UINT32_BYTES; i++) {
68     buffer[i] = (uint8_t)(value & 0x7f);
69     value >>= 7;
70
71     if (value == 0)
72       return i + 1;
73
74     buffer[i] |= 0x80;
75   }
76
77   return 0;
78 }
79
80 /* format_protobuf iterates over all metric families in "metrics" and adds them
81  * to a buffer in ProtoBuf format. It prefixes each protobuf with its encoded
82  * size, the so called "delimited" format. */
83 static void format_protobuf(ProtobufCBuffer *buffer) {
84   pthread_mutex_lock(&metrics_lock);
85
86   char *unused_name;
87   Io__Prometheus__Client__MetricFamily *fam;
88   c_avl_iterator_t *iter = c_avl_get_iterator(metrics);
89   while (c_avl_iterator_next(iter, (void *)&unused_name, (void *)&fam) == 0) {
90     /* Prometheus uses a message length prefix to determine where one
91      * MetricFamily ends and the next begins. This delimiter is encoded as a
92      * "varint", which is common in Protobufs. */
93     uint8_t delim[VARINT_UINT32_BYTES] = {0};
94     size_t delim_len = varint(
95         delim,
96         (uint32_t)io__prometheus__client__metric_family__get_packed_size(fam));
97     buffer->append(buffer, delim_len, delim);
98
99     io__prometheus__client__metric_family__pack_to_buffer(fam, buffer);
100   }
101   c_avl_iterator_destroy(iter);
102
103   pthread_mutex_unlock(&metrics_lock);
104 }
105
106 static char const *escape_label_value(char *buffer, size_t buffer_size,
107                                       char const *value) {
108   /* shortcut for values that don't need escaping. */
109   if (strpbrk(value, "\n\"\\") == NULL)
110     return value;
111
112   size_t value_len = strlen(value);
113   size_t buffer_len = 0;
114
115   for (size_t i = 0; i < value_len; i++) {
116     switch (value[i]) {
117     case '\n':
118     case '"':
119     case '\\':
120       if ((buffer_size - buffer_len) < 3) {
121         break;
122       }
123       buffer[buffer_len] = '\\';
124       buffer[buffer_len + 1] = (value[i] == '\n') ? 'n' : value[i];
125       buffer_len += 2;
126       break;
127
128     default:
129       if ((buffer_size - buffer_len) < 2) {
130         break;
131       }
132       buffer[buffer_len] = value[i];
133       buffer_len++;
134       break;
135     }
136   }
137
138   assert(buffer_len < buffer_size);
139   buffer[buffer_len] = 0;
140   return buffer;
141 }
142
143 /* format_labels formats a metric's labels in Prometheus-compatible format. This
144  * format looks like this:
145  *
146  *   key0="value0",key1="value1"
147  */
148 static char *format_labels(char *buffer, size_t buffer_size,
149                            Io__Prometheus__Client__Metric const *m) {
150   /* our metrics always have at least one and at most three labels. */
151   assert(m->n_label >= 1);
152   assert(m->n_label <= 3);
153
154 #define LABEL_KEY_SIZE DATA_MAX_NAME_LEN
155 #define LABEL_VALUE_SIZE (2 * DATA_MAX_NAME_LEN - 1)
156 #define LABEL_BUFFER_SIZE (LABEL_KEY_SIZE + LABEL_VALUE_SIZE + 4)
157
158   char *labels[3] = {
159       (char[LABEL_BUFFER_SIZE]){0}, (char[LABEL_BUFFER_SIZE]){0},
160       (char[LABEL_BUFFER_SIZE]){0},
161   };
162
163   /* N.B.: the label *names* are hard-coded by this plugin and therefore we
164    * know that they are sane. */
165   for (size_t i = 0; i < m->n_label; i++) {
166     char value[LABEL_VALUE_SIZE];
167     snprintf(labels[i], LABEL_BUFFER_SIZE, "%s=\"%s\"", m->label[i]->name,
168              escape_label_value(value, sizeof(value), m->label[i]->value));
169   }
170
171   strjoin(buffer, buffer_size, labels, m->n_label, ",");
172   return buffer;
173 }
174
175 /* format_protobuf iterates over all metric families in "metrics" and adds them
176  * to a buffer in plain text format. */
177 static void format_text(ProtobufCBuffer *buffer) {
178   pthread_mutex_lock(&metrics_lock);
179
180   char *unused_name;
181   Io__Prometheus__Client__MetricFamily *fam;
182   c_avl_iterator_t *iter = c_avl_get_iterator(metrics);
183   while (c_avl_iterator_next(iter, (void *)&unused_name, (void *)&fam) == 0) {
184     char line[1024]; /* 4x DATA_MAX_NAME_LEN? */
185
186     snprintf(line, sizeof(line), "# HELP %s %s\n", fam->name, fam->help);
187     buffer->append(buffer, strlen(line), (uint8_t *)line);
188
189     snprintf(line, sizeof(line), "# TYPE %s %s\n", fam->name,
190              (fam->type == IO__PROMETHEUS__CLIENT__METRIC_TYPE__GAUGE)
191                  ? "gauge"
192                  : "counter");
193     buffer->append(buffer, strlen(line), (uint8_t *)line);
194
195     for (size_t i = 0; i < fam->n_metric; i++) {
196       Io__Prometheus__Client__Metric *m = fam->metric[i];
197
198       char labels[1024];
199
200       char timestamp_ms[24] = "";
201       if (m->has_timestamp_ms)
202         snprintf(timestamp_ms, sizeof(timestamp_ms), " %" PRIi64,
203                  m->timestamp_ms);
204
205       if (fam->type == IO__PROMETHEUS__CLIENT__METRIC_TYPE__GAUGE)
206         snprintf(line, sizeof(line), "%s{%s} " GAUGE_FORMAT "%s\n", fam->name,
207                  format_labels(labels, sizeof(labels), m), m->gauge->value,
208                  timestamp_ms);
209       else /* if (fam->type == IO__PROMETHEUS__CLIENT__METRIC_TYPE__COUNTER) */
210         snprintf(line, sizeof(line), "%s{%s} %.0f%s\n", fam->name,
211                  format_labels(labels, sizeof(labels), m), m->counter->value,
212                  timestamp_ms);
213
214       buffer->append(buffer, strlen(line), (uint8_t *)line);
215     }
216   }
217   c_avl_iterator_destroy(iter);
218
219   char server[1024];
220   snprintf(server, sizeof(server), "\n# collectd/write_prometheus %s at %s\n",
221            PACKAGE_VERSION, hostname_g);
222   buffer->append(buffer, strlen(server), (uint8_t *)server);
223
224   pthread_mutex_unlock(&metrics_lock);
225 }
226
227 /* http_handler is the callback called by the microhttpd library. It essentially
228  * handles all HTTP request aspects and creates an HTTP response. */
229 static int http_handler(void *cls, struct MHD_Connection *connection,
230                         const char *url, const char *method,
231                         const char *version, const char *upload_data,
232                         size_t *upload_data_size, void **connection_state) {
233   if (strcmp(method, MHD_HTTP_METHOD_GET) != 0) {
234     return MHD_NO;
235   }
236
237   /* On the first call for each connection, return without anything further.
238    * Apparently not everything has been initialized yet or so; the docs are not
239    * very specific on the issue. */
240   if (*connection_state == NULL) {
241     /* set to a random non-NULL pointer. */
242     *connection_state = &(int){42};
243     return MHD_YES;
244   }
245
246   char const *accept = MHD_lookup_connection_value(connection, MHD_HEADER_KIND,
247                                                    MHD_HTTP_HEADER_ACCEPT);
248   bool want_proto = (accept != NULL) &&
249                     (strstr(accept, "application/vnd.google.protobuf") != NULL);
250
251   uint8_t scratch[4096] = {0};
252   ProtobufCBufferSimple simple = PROTOBUF_C_BUFFER_SIMPLE_INIT(scratch);
253   ProtobufCBuffer *buffer = (ProtobufCBuffer *)&simple;
254
255   if (want_proto)
256     format_protobuf(buffer);
257   else
258     format_text(buffer);
259
260 #if defined(MHD_VERSION) && MHD_VERSION >= 0x00090500
261   struct MHD_Response *res = MHD_create_response_from_buffer(
262       simple.len, simple.data, MHD_RESPMEM_MUST_COPY);
263 #else
264   struct MHD_Response *res = MHD_create_response_from_data(
265       simple.len, simple.data, /* must_free = */ 0, /* must_copy = */ 1);
266 #endif
267   MHD_add_response_header(res, MHD_HTTP_HEADER_CONTENT_TYPE,
268                           want_proto ? CONTENT_TYPE_PROTO : CONTENT_TYPE_TEXT);
269
270   int status = MHD_queue_response(connection, MHD_HTTP_OK, res);
271
272   MHD_destroy_response(res);
273   PROTOBUF_C_BUFFER_SIMPLE_CLEAR(&simple);
274   return status;
275 }
276
277 /*
278  * Functions for manipulating the global state in "metrics". This is organized
279  * in two tiers: the global "metrics" tree holds "metric families", which are
280  * identified by a name (a string). Each metric family has one or more
281  * "metrics", which are identified by a unique set of key-value-pairs. For
282  * example:
283  *
284  * collectd_cpu_total
285  *   {cpu="0",type="idle"}
286  *   {cpu="0",type="user"}
287  *   ...
288  * collectd_memory
289  *   {memory="used"}
290  *   {memory="free"}
291  *   ...
292  * {{{ */
293 /* label_pair_destroy frees the memory used by a label pair. */
294 static void label_pair_destroy(Io__Prometheus__Client__LabelPair *msg) {
295   if (msg == NULL)
296     return;
297
298   sfree(msg->name);
299   sfree(msg->value);
300
301   sfree(msg);
302 }
303
304 /* label_pair_clone allocates and initializes a new label pair. */
305 static Io__Prometheus__Client__LabelPair *
306 label_pair_clone(Io__Prometheus__Client__LabelPair const *orig) {
307   Io__Prometheus__Client__LabelPair *copy = calloc(1, sizeof(*copy));
308   if (copy == NULL)
309     return NULL;
310   io__prometheus__client__label_pair__init(copy);
311
312   copy->name = strdup(orig->name);
313   copy->value = strdup(orig->value);
314   if ((copy->name == NULL) || (copy->value == NULL)) {
315     label_pair_destroy(copy);
316     return NULL;
317   }
318
319   return copy;
320 }
321
322 /* metric_destroy frees the memory used by a metric. */
323 static void metric_destroy(Io__Prometheus__Client__Metric *msg) {
324   if (msg == NULL)
325     return;
326
327   for (size_t i = 0; i < msg->n_label; i++) {
328     label_pair_destroy(msg->label[i]);
329   }
330   sfree(msg->label);
331
332   sfree(msg->gauge);
333   sfree(msg->counter);
334
335   sfree(msg);
336 }
337
338 /* metric_cmp compares two metrics. It's prototype makes it easy to use with
339  * qsort(3) and bsearch(3). */
340 static int metric_cmp(void const *a, void const *b) {
341   Io__Prometheus__Client__Metric const *m_a =
342       *((Io__Prometheus__Client__Metric **)a);
343   Io__Prometheus__Client__Metric const *m_b =
344       *((Io__Prometheus__Client__Metric **)b);
345
346   if (m_a->n_label < m_b->n_label)
347     return -1;
348   else if (m_a->n_label > m_b->n_label)
349     return 1;
350
351   /* Prometheus does not care about the order of labels. All labels in this
352    * plugin are created by METRIC_ADD_LABELS(), though, and therefore always
353    * appear in the same order. We take advantage of this and simplify the check
354    * by making sure all labels are the same in each position.
355    *
356    * We also only need to check the label values, because the label names are
357    * the same for all metrics in a metric family.
358    *
359    * 3 labels:
360    * [0] $plugin="$plugin_instance" => $plugin is the same within a family
361    * [1] type="$type_instance"      => "type" is a static string
362    * [2] instance="$host"           => "instance" is a static string
363    *
364    * 2 labels, variant 1:
365    * [0] $plugin="$plugin_instance" => $plugin is the same within a family
366    * [1] instance="$host"           => "instance" is a static string
367    *
368    * 2 labels, variant 2:
369    * [0] $plugin="$type_instance"   => $plugin is the same within a family
370    * [1] instance="$host"           => "instance" is a static string
371    *
372    * 1 label:
373    * [1] instance="$host"           => "instance" is a static string
374    */
375   for (size_t i = 0; i < m_a->n_label; i++) {
376     int status = strcmp(m_a->label[i]->value, m_b->label[i]->value);
377     if (status != 0)
378       return status;
379
380 #if COLLECT_DEBUG
381     assert(strcmp(m_a->label[i]->name, m_b->label[i]->name) == 0);
382 #endif
383   }
384
385   return 0;
386 }
387
388 #define METRIC_INIT                                                            \
389   &(Io__Prometheus__Client__Metric) {                                          \
390     .label =                                                                   \
391         (Io__Prometheus__Client__LabelPair *[]){                               \
392             &(Io__Prometheus__Client__LabelPair){                              \
393                 .name = NULL,                                                  \
394             },                                                                 \
395             &(Io__Prometheus__Client__LabelPair){                              \
396                 .name = NULL,                                                  \
397             },                                                                 \
398             &(Io__Prometheus__Client__LabelPair){                              \
399                 .name = NULL,                                                  \
400             },                                                                 \
401         },                                                                     \
402     .n_label = 0,                                                              \
403   }
404
405 #define METRIC_ADD_LABELS(m, vl)                                               \
406   do {                                                                         \
407     if (strlen((vl)->plugin_instance) != 0) {                                  \
408       (m)->label[(m)->n_label]->name = (char *)(vl)->plugin;                   \
409       (m)->label[(m)->n_label]->value = (char *)(vl)->plugin_instance;         \
410       (m)->n_label++;                                                          \
411     }                                                                          \
412                                                                                \
413     if (strlen((vl)->type_instance) != 0) {                                    \
414       (m)->label[(m)->n_label]->name = "type";                                 \
415       if (strlen((vl)->plugin_instance) == 0)                                  \
416         (m)->label[(m)->n_label]->name = (char *)(vl)->plugin;                 \
417       (m)->label[(m)->n_label]->value = (char *)(vl)->type_instance;           \
418       (m)->n_label++;                                                          \
419     }                                                                          \
420                                                                                \
421     (m)->label[(m)->n_label]->name = "instance";                               \
422     (m)->label[(m)->n_label]->value = (char *)(vl)->host;                      \
423     (m)->n_label++;                                                            \
424   } while (0)
425
426 /* metric_clone allocates and initializes a new metric based on orig. */
427 static Io__Prometheus__Client__Metric *
428 metric_clone(Io__Prometheus__Client__Metric const *orig) {
429   Io__Prometheus__Client__Metric *copy = calloc(1, sizeof(*copy));
430   if (copy == NULL)
431     return NULL;
432   io__prometheus__client__metric__init(copy);
433
434   copy->n_label = orig->n_label;
435   copy->label = calloc(copy->n_label, sizeof(*copy->label));
436   if (copy->label == NULL) {
437     sfree(copy);
438     return NULL;
439   }
440
441   for (size_t i = 0; i < copy->n_label; i++) {
442     copy->label[i] = label_pair_clone(orig->label[i]);
443     if (copy->label[i] == NULL) {
444       metric_destroy(copy);
445       return NULL;
446     }
447   }
448
449   return copy;
450 }
451
452 /* metric_update stores the new value and timestamp in m. */
453 static int metric_update(Io__Prometheus__Client__Metric *m, value_t value,
454                          int ds_type, cdtime_t t, cdtime_t interval) {
455   if (ds_type == DS_TYPE_GAUGE) {
456     sfree(m->counter);
457     if (m->gauge == NULL) {
458       m->gauge = calloc(1, sizeof(*m->gauge));
459       if (m->gauge == NULL)
460         return ENOMEM;
461       io__prometheus__client__gauge__init(m->gauge);
462     }
463
464     m->gauge->value = (double)value.gauge;
465     m->gauge->has_value = 1;
466   } else { /* not gauge */
467     sfree(m->gauge);
468     if (m->counter == NULL) {
469       m->counter = calloc(1, sizeof(*m->counter));
470       if (m->counter == NULL)
471         return ENOMEM;
472       io__prometheus__client__counter__init(m->counter);
473     }
474
475     switch (ds_type) {
476     case DS_TYPE_ABSOLUTE:
477       m->counter->value = (double)value.absolute;
478       break;
479     case DS_TYPE_COUNTER:
480       m->counter->value = (double)value.counter;
481       break;
482     default:
483       m->counter->value = (double)value.derive;
484       break;
485     }
486     m->counter->has_value = 1;
487   }
488
489   /* Prometheus has a globally configured timeout after which metrics are
490    * considered stale. This causes problems when metrics have an interval
491    * exceeding that limit. We emulate the behavior of "pushgateway" and *not*
492    * send a timestamp value â€“ Prometheus will fill in the current time. */
493   if (interval <= staleness_delta) {
494     m->timestamp_ms = CDTIME_T_TO_MS(t);
495     m->has_timestamp_ms = 1;
496   } else {
497     static c_complain_t long_metric = C_COMPLAIN_INIT_STATIC;
498     c_complain(
499         LOG_NOTICE, &long_metric,
500         "write_prometheus plugin: You have metrics with an interval exceeding "
501         "\"StalenessDelta\" setting (%.3fs). This is suboptimal, please check "
502         "the collectd.conf(5) manual page to understand what's going on.",
503         CDTIME_T_TO_DOUBLE(staleness_delta));
504
505     m->timestamp_ms = 0;
506     m->has_timestamp_ms = 0;
507   }
508
509   return 0;
510 }
511
512 /* metric_family_add_metric adds m to the metric list of fam. */
513 static int metric_family_add_metric(Io__Prometheus__Client__MetricFamily *fam,
514                                     Io__Prometheus__Client__Metric *m) {
515   Io__Prometheus__Client__Metric **tmp =
516       realloc(fam->metric, (fam->n_metric + 1) * sizeof(*fam->metric));
517   if (tmp == NULL)
518     return ENOMEM;
519   fam->metric = tmp;
520
521   fam->metric[fam->n_metric] = m;
522   fam->n_metric++;
523
524   /* Sort the metrics so that lookup is fast. */
525   qsort(fam->metric, fam->n_metric, sizeof(*fam->metric), metric_cmp);
526
527   return 0;
528 }
529
530 /* metric_family_delete_metric looks up and deletes the metric corresponding to
531  * vl. */
532 static int
533 metric_family_delete_metric(Io__Prometheus__Client__MetricFamily *fam,
534                             value_list_t const *vl) {
535   Io__Prometheus__Client__Metric *key = METRIC_INIT;
536   METRIC_ADD_LABELS(key, vl);
537
538   size_t i;
539   for (i = 0; i < fam->n_metric; i++) {
540     if (metric_cmp(&key, &fam->metric[i]) == 0)
541       break;
542   }
543
544   if (i >= fam->n_metric)
545     return ENOENT;
546
547   metric_destroy(fam->metric[i]);
548   if ((fam->n_metric - 1) > i)
549     memmove(&fam->metric[i], &fam->metric[i + 1],
550             ((fam->n_metric - 1) - i) * sizeof(fam->metric[i]));
551   fam->n_metric--;
552
553   if (fam->n_metric == 0) {
554     sfree(fam->metric);
555     return 0;
556   }
557
558   Io__Prometheus__Client__Metric **tmp =
559       realloc(fam->metric, fam->n_metric * sizeof(*fam->metric));
560   if (tmp != NULL)
561     fam->metric = tmp;
562
563   return 0;
564 }
565
566 /* metric_family_get_metric looks up the matching metric in a metric family,
567  * allocating it if necessary. */
568 static Io__Prometheus__Client__Metric *
569 metric_family_get_metric(Io__Prometheus__Client__MetricFamily *fam,
570                          value_list_t const *vl) {
571   Io__Prometheus__Client__Metric *key = METRIC_INIT;
572   METRIC_ADD_LABELS(key, vl);
573
574   /* Metrics are sorted in metric_family_add_metric() so that we can do a binary
575    * search here. */
576   Io__Prometheus__Client__Metric **m = bsearch(
577       &key, fam->metric, fam->n_metric, sizeof(*fam->metric), metric_cmp);
578
579   if (m != NULL) {
580     return *m;
581   }
582
583   Io__Prometheus__Client__Metric *new_metric = metric_clone(key);
584   if (new_metric == NULL)
585     return NULL;
586
587   DEBUG("write_prometheus plugin: created new metric in family");
588   int status = metric_family_add_metric(fam, new_metric);
589   if (status != 0) {
590     metric_destroy(new_metric);
591     return NULL;
592   }
593
594   return new_metric;
595 }
596
597 /* metric_family_update looks up the matching metric in a metric family,
598  * allocating it if necessary, and updates the metric to the latest value. */
599 static int metric_family_update(Io__Prometheus__Client__MetricFamily *fam,
600                                 data_set_t const *ds, value_list_t const *vl,
601                                 size_t ds_index) {
602   Io__Prometheus__Client__Metric *m = metric_family_get_metric(fam, vl);
603   if (m == NULL)
604     return -1;
605
606   return metric_update(m, vl->values[ds_index], ds->ds[ds_index].type, vl->time,
607                        vl->interval);
608 }
609
610 /* metric_family_destroy frees the memory used by a metric family. */
611 static void metric_family_destroy(Io__Prometheus__Client__MetricFamily *msg) {
612   if (msg == NULL)
613     return;
614
615   sfree(msg->name);
616   sfree(msg->help);
617
618   for (size_t i = 0; i < msg->n_metric; i++) {
619     metric_destroy(msg->metric[i]);
620   }
621   sfree(msg->metric);
622
623   sfree(msg);
624 }
625
626 /* metric_family_create allocates and initializes a new metric family. */
627 static Io__Prometheus__Client__MetricFamily *
628 metric_family_create(char *name, data_set_t const *ds, value_list_t const *vl,
629                      size_t ds_index) {
630   Io__Prometheus__Client__MetricFamily *msg = calloc(1, sizeof(*msg));
631   if (msg == NULL)
632     return NULL;
633   io__prometheus__client__metric_family__init(msg);
634
635   msg->name = name;
636
637   char help[1024];
638   snprintf(
639       help, sizeof(help),
640       "write_prometheus plugin: '%s' Type: '%s', Dstype: '%s', Dsname: '%s'",
641       vl->plugin, vl->type, DS_TYPE_TO_STRING(ds->ds[ds_index].type),
642       ds->ds[ds_index].name);
643   msg->help = strdup(help);
644
645   msg->type = (ds->ds[ds_index].type == DS_TYPE_GAUGE)
646                   ? IO__PROMETHEUS__CLIENT__METRIC_TYPE__GAUGE
647                   : IO__PROMETHEUS__CLIENT__METRIC_TYPE__COUNTER;
648   msg->has_type = 1;
649
650   return msg;
651 }
652
653 /* metric_family_name creates a metric family's name from a data source. This is
654  * done in the same way as done by the "collectd_exporter" for best possible
655  * compatibility. In essence, the plugin, type and data source name go in the
656  * metric family name, while hostname, plugin instance and type instance go into
657  * the labels of a metric. */
658 static char *metric_family_name(data_set_t const *ds, value_list_t const *vl,
659                                 size_t ds_index) {
660   char const *fields[5] = {"collectd"};
661   size_t fields_num = 1;
662
663   if (strcmp(vl->plugin, vl->type) != 0) {
664     fields[fields_num] = vl->plugin;
665     fields_num++;
666   }
667   fields[fields_num] = vl->type;
668   fields_num++;
669
670   if (strcmp("value", ds->ds[ds_index].name) != 0) {
671     fields[fields_num] = ds->ds[ds_index].name;
672     fields_num++;
673   }
674
675   /* Prometheus best practices:
676    * cumulative metrics should have a "total" suffix. */
677   if ((ds->ds[ds_index].type == DS_TYPE_COUNTER) ||
678       (ds->ds[ds_index].type == DS_TYPE_DERIVE)) {
679     fields[fields_num] = "total";
680     fields_num++;
681   }
682
683   char name[5 * DATA_MAX_NAME_LEN];
684   strjoin(name, sizeof(name), (char **)fields, fields_num, "_");
685   return strdup(name);
686 }
687
688 /* metric_family_get looks up the matching metric family, allocating it if
689  * necessary. */
690 static Io__Prometheus__Client__MetricFamily *
691 metric_family_get(data_set_t const *ds, value_list_t const *vl, size_t ds_index,
692                   bool allocate) {
693   char *name = metric_family_name(ds, vl, ds_index);
694   if (name == NULL) {
695     ERROR("write_prometheus plugin: Allocating metric family name failed.");
696     return NULL;
697   }
698
699   Io__Prometheus__Client__MetricFamily *fam = NULL;
700   if (c_avl_get(metrics, name, (void *)&fam) == 0) {
701     sfree(name);
702     assert(fam != NULL);
703     return fam;
704   }
705
706   if (!allocate) {
707     sfree(name);
708     return NULL;
709   }
710
711   fam = metric_family_create(name, ds, vl, ds_index);
712   if (fam == NULL) {
713     ERROR("write_prometheus plugin: Allocating metric family failed.");
714     sfree(name);
715     return NULL;
716   }
717
718   /* If successful, "name" is owned by "fam", i.e. don't free it here. */
719   DEBUG("write_prometheus plugin: metric family \"%s\" has been created.",
720         name);
721   name = NULL;
722
723   int status = c_avl_insert(metrics, fam->name, fam);
724   if (status != 0) {
725     ERROR("write_prometheus plugin: Adding \"%s\" failed.", name);
726     metric_family_destroy(fam);
727     return NULL;
728   }
729
730   return fam;
731 }
732 /* }}} */
733
734 static void prom_logger(__attribute__((unused)) void *arg, char const *fmt,
735                         va_list ap) {
736   /* {{{ */
737   char errbuf[1024];
738   vsnprintf(errbuf, sizeof(errbuf), fmt, ap);
739
740   ERROR("write_prometheus plugin: %s", errbuf);
741 } /* }}} prom_logger */
742
743 #if MHD_VERSION >= 0x00090000
744 static int prom_open_socket(int addrfamily) {
745   /* {{{ */
746   char service[NI_MAXSERV];
747   snprintf(service, sizeof(service), "%hu", httpd_port);
748
749   struct addrinfo *res;
750   int status = getaddrinfo(httpd_host, service,
751                            &(struct addrinfo){
752                                .ai_flags = AI_PASSIVE | AI_ADDRCONFIG,
753                                .ai_family = addrfamily,
754                                .ai_socktype = SOCK_STREAM,
755                            },
756                            &res);
757   if (status != 0) {
758     return -1;
759   }
760
761   int fd = -1;
762   for (struct addrinfo *ai = res; ai != NULL; ai = ai->ai_next) {
763     fd = socket(ai->ai_family, ai->ai_socktype | SOCK_CLOEXEC, 0);
764     if (fd == -1)
765       continue;
766
767     if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &(int){1}, sizeof(int)) != 0) {
768       WARNING("write_prometheus plugin: setsockopt(SO_REUSEADDR) failed: %s",
769               STRERRNO);
770       close(fd);
771       fd = -1;
772       continue;
773     }
774
775     if (bind(fd, ai->ai_addr, ai->ai_addrlen) != 0) {
776       close(fd);
777       fd = -1;
778       continue;
779     }
780
781     if (listen(fd, /* backlog = */ 16) != 0) {
782       close(fd);
783       fd = -1;
784       continue;
785     }
786
787     char str_node[NI_MAXHOST];
788     char str_service[NI_MAXSERV];
789
790     getnameinfo(ai->ai_addr, ai->ai_addrlen, str_node, sizeof(str_node),
791                 str_service, sizeof(str_service),
792                 NI_NUMERICHOST | NI_NUMERICSERV);
793
794     INFO("write_prometheus plugin: Listening on [%s]:%s.", str_node,
795          str_service);
796     break;
797   }
798
799   freeaddrinfo(res);
800
801   return fd;
802 } /* }}} int prom_open_socket */
803
804 static struct MHD_Daemon *prom_start_daemon() {
805   /* {{{ */
806   int fd = prom_open_socket(PF_INET6);
807   if (fd == -1)
808     fd = prom_open_socket(PF_INET);
809   if (fd == -1) {
810     ERROR("write_prometheus plugin: Opening a listening socket for [%s]:%hu "
811           "failed.",
812           (httpd_host != NULL) ? httpd_host : "::", httpd_port);
813     return NULL;
814   }
815
816   unsigned int flags = MHD_USE_THREAD_PER_CONNECTION | MHD_USE_DEBUG;
817 #if MHD_VERSION >= 0x00095300
818   flags |= MHD_USE_INTERNAL_POLLING_THREAD;
819 #endif
820
821   struct MHD_Daemon *d = MHD_start_daemon(
822       flags, httpd_port,
823       /* MHD_AcceptPolicyCallback = */ NULL,
824       /* MHD_AcceptPolicyCallback arg = */ NULL, http_handler, NULL,
825       MHD_OPTION_LISTEN_SOCKET, fd, MHD_OPTION_EXTERNAL_LOGGER, prom_logger,
826       NULL, MHD_OPTION_END);
827   if (d == NULL) {
828     ERROR("write_prometheus plugin: MHD_start_daemon() failed.");
829     close(fd);
830     return NULL;
831   }
832
833   return d;
834 } /* }}} struct MHD_Daemon *prom_start_daemon */
835 #else /* if MHD_VERSION < 0x00090000 */
836 static struct MHD_Daemon *prom_start_daemon() {
837   /* {{{ */
838   struct MHD_Daemon *d = MHD_start_daemon(
839       MHD_USE_THREAD_PER_CONNECTION | MHD_USE_DEBUG, httpd_port,
840       /* MHD_AcceptPolicyCallback = */ NULL,
841       /* MHD_AcceptPolicyCallback arg = */ NULL, http_handler, NULL,
842       MHD_OPTION_EXTERNAL_LOGGER, prom_logger, NULL, MHD_OPTION_END);
843   if (d == NULL) {
844     ERROR("write_prometheus plugin: MHD_start_daemon() failed.");
845     return NULL;
846   }
847
848   return d;
849 } /* }}} struct MHD_Daemon *prom_start_daemon */
850 #endif
851
852 /*
853  * collectd callbacks
854  */
855 static int prom_config(oconfig_item_t *ci) {
856   for (int i = 0; i < ci->children_num; i++) {
857     oconfig_item_t *child = ci->children + i;
858
859     if (strcasecmp("Host", child->key) == 0) {
860 #if MHD_VERSION >= 0x00090000
861       cf_util_get_string(child, &httpd_host);
862 #else
863       ERROR("write_prometheus plugin: Option `Host' not supported. Please "
864             "upgrade libmicrohttpd to at least 0.9.0");
865       return -1;
866 #endif
867     } else if (strcasecmp("Port", child->key) == 0) {
868       int status = cf_util_get_port_number(child);
869       if (status > 0)
870         httpd_port = (unsigned short)status;
871     } else if (strcasecmp("StalenessDelta", child->key) == 0) {
872       cf_util_get_cdtime(child, &staleness_delta);
873     } else {
874       WARNING("write_prometheus plugin: Ignoring unknown configuration option "
875               "\"%s\".",
876               child->key);
877     }
878   }
879
880   return 0;
881 }
882
883 static int prom_init() {
884   if (metrics == NULL) {
885     metrics = c_avl_create((void *)strcmp);
886     if (metrics == NULL) {
887       ERROR("write_prometheus plugin: c_avl_create() failed.");
888       return -1;
889     }
890   }
891
892   if (httpd == NULL) {
893     httpd = prom_start_daemon();
894     if (httpd == NULL) {
895       return -1;
896     }
897     DEBUG("write_prometheus plugin: Successfully started microhttpd %s",
898           MHD_get_version());
899   }
900
901   return 0;
902 }
903
904 static int prom_write(data_set_t const *ds, value_list_t const *vl,
905                       __attribute__((unused)) user_data_t *ud) {
906   pthread_mutex_lock(&metrics_lock);
907
908   for (size_t i = 0; i < ds->ds_num; i++) {
909     Io__Prometheus__Client__MetricFamily *fam =
910         metric_family_get(ds, vl, i, /* allocate = */ true);
911     if (fam == NULL)
912       continue;
913
914     int status = metric_family_update(fam, ds, vl, i);
915     if (status != 0) {
916       ERROR("write_prometheus plugin: Updating metric \"%s\" failed with "
917             "status %d",
918             fam->name, status);
919       continue;
920     }
921   }
922
923   pthread_mutex_unlock(&metrics_lock);
924   return 0;
925 }
926
927 static int prom_missing(value_list_t const *vl,
928                         __attribute__((unused)) user_data_t *ud) {
929   data_set_t const *ds = plugin_get_ds(vl->type);
930   if (ds == NULL)
931     return ENOENT;
932
933   pthread_mutex_lock(&metrics_lock);
934
935   for (size_t i = 0; i < ds->ds_num; i++) {
936     Io__Prometheus__Client__MetricFamily *fam =
937         metric_family_get(ds, vl, i, /* allocate = */ false);
938     if (fam == NULL)
939       continue;
940
941     int status = metric_family_delete_metric(fam, vl);
942     if (status != 0) {
943       ERROR("write_prometheus plugin: Deleting a metric in family \"%s\" "
944             "failed with status %d",
945             fam->name, status);
946
947       continue;
948     }
949
950     if (fam->n_metric == 0) {
951       int status = c_avl_remove(metrics, fam->name, NULL, NULL);
952       if (status != 0) {
953         ERROR("write_prometheus plugin: Deleting metric family \"%s\" failed "
954               "with status %d",
955               fam->name, status);
956         continue;
957       }
958       metric_family_destroy(fam);
959     }
960   }
961
962   pthread_mutex_unlock(&metrics_lock);
963   return 0;
964 }
965
966 static int prom_shutdown() {
967   if (httpd != NULL) {
968     MHD_stop_daemon(httpd);
969     httpd = NULL;
970   }
971
972   pthread_mutex_lock(&metrics_lock);
973   if (metrics != NULL) {
974     char *name;
975     Io__Prometheus__Client__MetricFamily *fam;
976     while (c_avl_pick(metrics, (void *)&name, (void *)&fam) == 0) {
977       assert(name == fam->name);
978       name = NULL;
979
980       metric_family_destroy(fam);
981     }
982     c_avl_destroy(metrics);
983     metrics = NULL;
984   }
985   pthread_mutex_unlock(&metrics_lock);
986
987   sfree(httpd_host);
988
989   return 0;
990 }
991
992 void module_register() {
993   plugin_register_complex_config("write_prometheus", prom_config);
994   plugin_register_init("write_prometheus", prom_init);
995   plugin_register_write("write_prometheus", prom_write,
996                         /* user data = */ NULL);
997   plugin_register_missing("write_prometheus", prom_missing,
998                           /* user data = */ NULL);
999   plugin_register_shutdown("write_prometheus", prom_shutdown);
1000 }