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 val) {
  61   value_t values[1];
  62   value_list_t vl = VALUE_LIST_INIT;
  63
  64   values[0].gauge = val;
  65
  66   vl.values = values;
  67   vl.values_len = 1;
  68   sstrncpy(vl.host, hostname_g, sizeof(vl.host));
  69   sstrncpy(vl.plugin, "zookeeper", sizeof(vl.plugin));
  70   sstrncpy(vl.type, type, sizeof(vl.type));
  71   if (type_inst != NULL)
  72     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
  73
  74   plugin_dispatch_values(&vl);
  75 } /* zookeeper_submit_gauge */
  76
  77 static void zookeeper_submit_derive(const char *type, const char *type_inst,
  78                                     derive_t val) {
  79   value_t values[1];
  80   value_list_t vl = VALUE_LIST_INIT;
  81
  82   values[0].derive = val;
  83
  84   vl.values = values;
  85   vl.values_len = 1;
  86   sstrncpy(vl.host, hostname_g, sizeof(vl.host));
  87   sstrncpy(vl.plugin, "zookeeper", sizeof(vl.plugin));
  88   sstrncpy(vl.type, type, sizeof(vl.type));
  89   if (type_inst != NULL)
  90     sstrncpy(vl.type_instance, type_inst, sizeof(vl.type_instance));
  91
  92   plugin_dispatch_values(&vl);
  93 } /* zookeeper_submit_derive */
  94
  95 static int zookeeper_connect(void) {
  96   int sk = -1;
  97   int status;
  98   struct addrinfo *ai_list;
  99   const char *host;
 100   const char *port;
 101
 102   host = (zk_host != NULL) ? zk_host : ZOOKEEPER_DEF_HOST;
 103   port = (zk_port != NULL) ? zk_port : ZOOKEEPER_DEF_PORT;
 104
 105   struct addrinfo ai_hints = {.ai_family = AF_UNSPEC,
 106                               .ai_socktype = SOCK_STREAM};
 107
 108   status = getaddrinfo(host, port, &ai_hints, &ai_list);
 109   if (status != 0) {
 110     char errbuf[1024];
 111     INFO("getaddrinfo failed: %s",
 112          (status == EAI_SYSTEM) ? sstrerror(errno, errbuf, sizeof(errbuf))
 113                                 : gai_strerror(status));
 114     return (-1);
 115   }
 116
 117   for (struct addrinfo *ai = ai_list; ai != NULL; ai = ai->ai_next) {
 118     sk = socket(ai->ai_family, SOCK_STREAM, 0);
 119     if (sk < 0) {
 120       char errbuf[1024];
 121       WARNING("zookeeper: socket(2) failed: %s",
 122               sstrerror(errno, errbuf, sizeof(errbuf)));
 123       continue;
 124     }
 125     status = (int)connect(sk, ai->ai_addr, ai->ai_addrlen);
 126     if (status != 0) {
 127       char errbuf[1024];
 128       close(sk);
 129       sk = -1;
 130       WARNING("zookeeper: connect(2) failed: %s",
 131               sstrerror(errno, errbuf, sizeof(errbuf)));
 132       continue;
 133     }
 134
 135     /* connected */
 136     break;
 137   }
 138
 139   freeaddrinfo(ai_list);
 140   return (sk);
 141 } /* int zookeeper_connect */
 142
 143 static int zookeeper_query(char *buffer, size_t buffer_size) {
 144   int sk, status;
 145   size_t buffer_fill;
 146
 147   sk = zookeeper_connect();
 148   if (sk < 0) {
 149     ERROR("zookeeper: Could not connect to daemon");
 150     return (-1);
 151   }
 152
 153   status = (int)swrite(sk, "mntr\r\n", strlen("mntr\r\n"));
 154   if (status != 0) {
 155     char errbuf[1024];
 156     ERROR("zookeeper: write(2) failed: %s",
 157           sstrerror(errno, errbuf, sizeof(errbuf)));
 158     close(sk);
 159     return (-1);
 160   }
 161
 162   memset(buffer, 0, buffer_size);
 163   buffer_fill = 0;
 164
 165   while ((status = (int)recv(sk, buffer + buffer_fill,
 166                              buffer_size - buffer_fill, /* flags = */ 0)) !=
 167          0) {
 168     if (status < 0) {
 169       char errbuf[1024];
 170       if ((errno == EAGAIN) || (errno == EINTR))
 171         continue;
 172       ERROR("zookeeper: Error reading from socket: %s",
 173             sstrerror(errno, errbuf, sizeof(errbuf)));
 174       close(sk);
 175       return (-1);
 176     }
 177
 178     buffer_fill += (size_t)status;
 179   } /* while (recv) */
 180
 181   status = 0;
 182   if (buffer_fill == 0) {
 183     WARNING("zookeeper: No data returned by MNTR command.");
 184     status = -1;
 185   }
 186
 187   close(sk);
 188   return (status);
 189 } /* int zookeeper_query */
 190
 191 static int zookeeper_read(void) {
 192   char buf[4096];
 193   char *ptr;
 194   char *save_ptr;
 195   char *line;
 196   char *fields[2];
 197
 198   if (zookeeper_query(buf, sizeof(buf)) < 0) {
 199     return (-1);
 200   }
 201
 202   ptr = buf;
 203   save_ptr = NULL;
 204   while ((line = strtok_r(ptr, "\n\r", &save_ptr)) != NULL) {
 205     ptr = NULL;
 206     if (strsplit(line, fields, 2) != 2) {
 207       continue;
 208     }
 209 #define FIELD_CHECK(check, expected)                                           \
 210   (strncmp(check, expected, strlen(expected)) == 0)
 211
 212     if (FIELD_CHECK(fields[0], "zk_avg_latency")) {
 213       zookeeper_submit_gauge("latency", "avg", atol(fields[1]));
 214     } else if (FIELD_CHECK(fields[0], "zk_min_latency")) {
 215       zookeeper_submit_gauge("latency", "min", atol(fields[1]));
 216     } else if (FIELD_CHECK(fields[0], "zk_max_latency")) {
 217       zookeeper_submit_gauge("latency", "max", atol(fields[1]));
 218     } else if (FIELD_CHECK(fields[0], "zk_packets_received")) {
 219       zookeeper_submit_derive("packets", "received", atol(fields[1]));
 220     } else if (FIELD_CHECK(fields[0], "zk_packets_sent")) {
 221       zookeeper_submit_derive("packets", "sent", atol(fields[1]));
 222     } else if (FIELD_CHECK(fields[0], "zk_num_alive_connections")) {
 223       zookeeper_submit_gauge("current_connections", NULL, atol(fields[1]));
 224     } else if (FIELD_CHECK(fields[0], "zk_outstanding_requests")) {
 225       zookeeper_submit_gauge("requests", "outstanding", atol(fields[1]));
 226     } else if (FIELD_CHECK(fields[0], "zk_znode_count")) {
 227       zookeeper_submit_gauge("gauge", "znode", atol(fields[1]));
 228     } else if (FIELD_CHECK(fields[0], "zk_watch_count")) {
 229       zookeeper_submit_gauge("gauge", "watch", atol(fields[1]));
 230     } else if (FIELD_CHECK(fields[0], "zk_ephemerals_count")) {
 231       zookeeper_submit_gauge("gauge", "ephemerals", atol(fields[1]));
 232     } else if (FIELD_CHECK(fields[0], "zk_ephemerals_count")) {
 233       zookeeper_submit_gauge("gauge", "ephemerals", atol(fields[1]));
 234     } else if (FIELD_CHECK(fields[0], "zk_ephemerals_count")) {
 235       zookeeper_submit_gauge("gauge", "ephemerals", atol(fields[1]));
 236     } else if (FIELD_CHECK(fields[0], "zk_approximate_data_size")) {
 237       zookeeper_submit_gauge("bytes", "approximate_data_size", atol(fields[1]));
 238     } else if (FIELD_CHECK(fields[0], "zk_followers")) {
 239       zookeeper_submit_gauge("count", "followers", atol(fields[1]));
 240     } else if (FIELD_CHECK(fields[0], "zk_synced_followers")) {
 241       zookeeper_submit_gauge("count", "synced_followers", atol(fields[1]));
 242     } else if (FIELD_CHECK(fields[0], "zk_pending_syncs")) {
 243       zookeeper_submit_gauge("count", "pending_syncs", atol(fields[1]));
 244     } else {
 245       DEBUG("Uncollected zookeeper MNTR field %s", fields[0]);
 246     }
 247   }
 248
 249   return (0);
 250 } /* zookeeper_read */
 251
 252 void module_register(void) {
 253   plugin_register_config("zookeeper", zookeeper_config, config_keys,
 254                          config_keys_num);
 255   plugin_register_read("zookeeper", zookeeper_read);
 256 } /* void module_register */