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