src/zookeeper.c

   1 /**
   2  * collectd - src/zookeeper.c
   3  * Copyright (C) 2014       Google, Inc.
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a
   6  * copy of this software and associated documentation files (the "Software"),
   7  * to deal in the Software without restriction, including without limitation
   8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9  * and/or sell copies of the Software, and to permit persons to whom the
  10  * Software is 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
  20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  21  * DEALINGS IN THE SOFTWARE.
  22  *
  23  * Authors:
  24  *   Jeremy Katz <jeremy at katzbox.net>
  25  **/
  26
  27 #include "collectd.h"
  28
  29 #include "common.h"
  30 #include "plugin.h"
  31
  32 #include <netdb.h>
  33 #include <netinet/in.h>
  34 #include <netinet/tcp.h>
  35 #include <sys/un.h>
  36
  37 #define ZOOKEEPER_DEF_HOST "127.0.0.1"
  38 #define ZOOKEEPER_DEF_PORT "2181"
  39
  40 static char *zk_host = NULL;
  41 static char *zk_port = NULL;
  42
  43 static const char *config_keys[] = {"Host", "Port"};
  44 static int config_keys_num = STATIC_ARRAY_SIZE(config_keys);
  45
  46 static int zookeeper_config(const char *key, const char *value) {
  47   if (strncmp(key, "Host", strlen("Host")) == 0) {
  48     sfree(zk_host);
  49     zk_host = strdup(value);
  50   } else if (strncmp(key, "Port", strlen("Port")) == 0) {
  51     sfree(zk_port);
  52     zk_port = strdup(value);
  53   } else {
  54     return -1;
  55   }
  56   return 0;
  57 }
  58
  59 static void zookeeper_submit_gauge(const char *type, const char *type_inst,
  60                                    gauge_t value) {
  61   value_list_t vl = VALUE_LIST_INIT;
  62
  63   vl.values = &(value_t){.gauge = value};
  64   vl.values_len = 1;
  65   sstrncpy(vl.plugin, "zookeeper", sizeof(vl.plugin));
  66   sstrncpy(vl.type, type, sizeof(vl.type));
  67   if (type_inst != NULL)
  68     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
  69
  70   plugin_dispatch_values(&vl);
  71 } /* zookeeper_submit_gauge */
  72
  73 static void zookeeper_submit_derive(const char *type, const char *type_inst,
  74                                     derive_t value) {
  75   value_list_t vl = VALUE_LIST_INIT;
  76
  77   vl.values = &(value_t){.derive = value};
  78   vl.values_len = 1;
  79   sstrncpy(vl.plugin, "zookeeper", sizeof(vl.plugin));
  80   sstrncpy(vl.type, type, sizeof(vl.type));
  81   if (type_inst != NULL)
  82     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
  83
  84   plugin_dispatch_values(&vl);
  85 } /* zookeeper_submit_derive */
  86
  87 static int zookeeper_connect(void) {
  88   int sk = -1;
  89   int status;
  90   struct addrinfo *ai_list;
  91   const char *host;
  92   const char *port;
  93
  94   host = (zk_host != NULL) ? zk_host : ZOOKEEPER_DEF_HOST;
  95   port = (zk_port != NULL) ? zk_port : ZOOKEEPER_DEF_PORT;
  96
  97   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
  98                               .ai_socktype = SOCK_STREAM};
  99
 100   status = getaddrinfo(host, port, &ai_hints, &ai_list);
 101   if (status != 0) {
 102     char errbuf[1024];
 103     INFO("getaddrinfo failed: %s",
 104          (status == EAI_SYSTEM) ? sstrerror(errno, errbuf, sizeof(errbuf))
 105                                 : gai_strerror(status));
 106     return -1;
 107   }
 108
 109   for (struct addrinfo *ai = ai_list; ai != NULL; ai = ai->ai_next) {
 110     sk = socket(ai->ai_family, SOCK_STREAM, 0);
 111     if (sk < 0) {
 112       char errbuf[1024];
 113       WARNING("zookeeper: socket(2) failed: %s",
 114               sstrerror(errno, errbuf, sizeof(errbuf)));
 115       continue;
 116     }
 117     status = (int)connect(sk, ai->ai_addr, ai->ai_addrlen);
 118     if (status != 0) {
 119       char errbuf[1024];
 120       close(sk);
 121       sk = -1;
 122       WARNING("zookeeper: connect(2) failed: %s",
 123               sstrerror(errno, errbuf, sizeof(errbuf)));
 124       continue;
 125     }
 126
 127     /* connected */
 128     break;
 129   }
 130
 131   freeaddrinfo(ai_list);
 132   return sk;
 133 } /* int zookeeper_connect */
 134
 135 static int zookeeper_query(char *buffer, size_t buffer_size) {
 136   int sk, status;
 137   size_t buffer_fill;
 138
 139   sk = zookeeper_connect();
 140   if (sk < 0) {
 141     ERROR("zookeeper: Could not connect to daemon");
 142     return -1;
 143   }
 144
 145   status = (int)swrite(sk, "mntr\r\n", strlen("mntr\r\n"));
 146   if (status != 0) {
 147     char errbuf[1024];
 148     ERROR("zookeeper: write(2) failed: %s",
 149           sstrerror(errno, errbuf, sizeof(errbuf)));
 150     close(sk);
 151     return -1;
 152   }
 153
 154   memset(buffer, 0, buffer_size);
 155   buffer_fill = 0;
 156
 157   while ((status = (int)recv(sk, buffer + buffer_fill,
 158                              buffer_size - buffer_fill, /* flags = */ 0)) !=
 159          0) {
 160     if (status < 0) {
 161       char errbuf[1024];
 162       if ((errno == EAGAIN) || (errno == EINTR))
 163         continue;
 164       ERROR("zookeeper: Error reading from socket: %s",
 165             sstrerror(errno, errbuf, sizeof(errbuf)));
 166       close(sk);
 167       return -1;
 168     }
 169
 170     buffer_fill += (size_t)status;
 171   } /* while (recv) */
 172
 173   status = 0;
 174   if (buffer_fill == 0) {
 175     WARNING("zookeeper: No data returned by MNTR command.");
 176     status = -1;
 177   }
 178
 179   close(sk);
 180   return status;
 181 } /* int zookeeper_query */
 182
 183 static int zookeeper_read(void) {
 184   char buf[4096];
 185   char *ptr;
 186   char *save_ptr;
 187   char *line;
 188   char *fields[2];
 189
 190   if (zookeeper_query(buf, sizeof(buf)) < 0) {
 191     return -1;
 192   }
 193
 194   ptr = buf;
 195   save_ptr = NULL;
 196   while ((line = strtok_r(ptr, "\n\r", &save_ptr)) != NULL) {
 197     ptr = NULL;
 198     if (strsplit(line, fields, 2) != 2) {
 199       continue;
 200     }
 201 #define FIELD_CHECK(check, expected)                                           \
 202   (strncmp(check, expected, strlen(expected)) == 0)
 203
 204     if (FIELD_CHECK(fields[0], "zk_avg_latency")) {
 205       zookeeper_submit_gauge("latency", "avg", atol(fields[1]));
 206     } else if (FIELD_CHECK(fields[0], "zk_min_latency")) {
 207       zookeeper_submit_gauge("latency", "min", atol(fields[1]));
 208     } else if (FIELD_CHECK(fields[0], "zk_max_latency")) {
 209       zookeeper_submit_gauge("latency", "max", atol(fields[1]));
 210     } else if (FIELD_CHECK(fields[0], "zk_packets_received")) {
 211       zookeeper_submit_derive("packets", "received", atol(fields[1]));
 212     } else if (FIELD_CHECK(fields[0], "zk_packets_sent")) {
 213       zookeeper_submit_derive("packets", "sent", atol(fields[1]));
 214     } else if (FIELD_CHECK(fields[0], "zk_num_alive_connections")) {
 215       zookeeper_submit_gauge("current_connections", NULL, atol(fields[1]));
 216     } else if (FIELD_CHECK(fields[0], "zk_outstanding_requests")) {
 217       zookeeper_submit_gauge("requests", "outstanding", atol(fields[1]));
 218     } else if (FIELD_CHECK(fields[0], "zk_znode_count")) {
 219       zookeeper_submit_gauge("gauge", "znode", atol(fields[1]));
 220     } else if (FIELD_CHECK(fields[0], "zk_watch_count")) {
 221       zookeeper_submit_gauge("gauge", "watch", atol(fields[1]));
 222     } else if (FIELD_CHECK(fields[0], "zk_ephemerals_count")) {
 223       zookeeper_submit_gauge("gauge", "ephemerals", atol(fields[1]));
 224     } else if (FIELD_CHECK(fields[0], "zk_ephemerals_count")) {
 225       zookeeper_submit_gauge("gauge", "ephemerals", atol(fields[1]));
 226     } else if (FIELD_CHECK(fields[0], "zk_ephemerals_count")) {
 227       zookeeper_submit_gauge("gauge", "ephemerals", atol(fields[1]));
 228     } else if (FIELD_CHECK(fields[0], "zk_approximate_data_size")) {
 229       zookeeper_submit_gauge("bytes", "approximate_data_size", atol(fields[1]));
 230     } else if (FIELD_CHECK(fields[0], "zk_followers")) {
 231       zookeeper_submit_gauge("count", "followers", atol(fields[1]));
 232     } else if (FIELD_CHECK(fields[0], "zk_synced_followers")) {
 233       zookeeper_submit_gauge("count", "synced_followers", atol(fields[1]));
 234     } else if (FIELD_CHECK(fields[0], "zk_pending_syncs")) {
 235       zookeeper_submit_gauge("count", "pending_syncs", atol(fields[1]));
 236     } else {
 237       DEBUG("Uncollected zookeeper MNTR field %s", fields[0]);
 238     }
 239   }
 240
 241   return 0;
 242 } /* zookeeper_read */
 243
 244 void module_register(void) {
 245   plugin_register_config("zookeeper", zookeeper_config, config_keys,
 246                          config_keys_num);
 247   plugin_register_read("zookeeper", zookeeper_read);
 248 } /* void module_register */