src/pinba.c

   1 /**
   2  * collectd - src/pinba.c (based on code from pinba_engine 0.0.5)
   3  * Copyright (c) 2007-2009  Antony Dovgal
   4  * Copyright (C) 2010       Phoenix Kayo
   5  *
   6  * This program is free software; you can redistribute it and/or modify it
   7  * under the terms of the GNU General Public License as published by the
   8  * Free Software Foundation; only version 2 of the License is applicable.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18  *
  19  * Authors:
  20  *   Antony Dovgal <tony at daylessday.org>
  21  *   Phoenix Kayo <kayo.k11.4 at gmail.com>
  22  **/
  23
  24 #define _XOPEN_SOURCE 500
  25
  26 #include "collectd.h"
  27 #include "common.h"
  28 #include "plugin.h"
  29 #include "configfile.h"
  30
  31 #include <pthread.h>
  32 #include <sys/socket.h>
  33 #include <netinet/in.h>
  34 #include <arpa/inet.h>
  35
  36 #include "pinba.pb-c.h"
  37
  38 typedef uint8_t u_char;
  39
  40 #include <event.h>
  41
  42 /*
  43  *  Service declaration section
  44  */
  45 #ifndef PINBA_UDP_BUFFER_SIZE
  46 # define PINBA_UDP_BUFFER_SIZE 65536
  47 #endif
  48
  49 #ifndef PINBA_DEFAULT_ADDRESS
  50 # define PINBA_DEFAULT_ADDRESS "127.0.0.1" /* FIXME */
  51 #endif
  52
  53 #ifndef PINBA_DEFAULT_PORT
  54 # define PINBA_DEFAULT_PORT 12345 /* FIXME */
  55 #endif
  56
  57 /*
  58  * Private data structures
  59  */
  60 typedef struct _pinba_statres_ pinba_statres;
  61 struct _pinba_statres_ {
  62   const char *name;
  63   double req_per_sec;
  64   double req_time;
  65   double ru_utime;
  66   double ru_stime;
  67   double doc_size;
  68   double mem_peak;
  69 };
  70
  71 typedef struct _pinba_socket_ pinba_socket;
  72 struct _pinba_socket_ {
  73   int listen_sock;
  74   struct event *accept_event;
  75 };
  76
  77 typedef double pinba_time;
  78 typedef uint32_t pinba_size;
  79
  80 static pinba_time now (void)
  81 {
  82   static struct timeval tv;
  83
  84   gettimeofday (&tv, /* tz = */ NULL);
  85
  86   return (double)tv.tv_sec+((double)tv.tv_usec/(double)1000000);
  87 }
  88
  89 static pthread_rwlock_t temp_lock;
  90
  91 static struct event_base *temp_base = NULL;
  92
  93 static pinba_socket *temp_sock = NULL;
  94
  95 static pthread_t temp_thrd;
  96
  97 typedef struct _pinba_statnode_ pinba_statnode;
  98 struct _pinba_statnode_{
  99   /* collector name */
 100   char* name;
 101   /* query data */
 102   char *host;
 103   char *server;
 104   char *script;
 105   /* collected data */
 106   pinba_time last_coll;
 107   pinba_size req_count;
 108   pinba_time req_time;
 109   pinba_time ru_utime;
 110   pinba_time ru_stime;
 111   pinba_size doc_size;
 112   pinba_size mem_peak;
 113 };
 114
 115 static unsigned int stat_nodes_count=0;
 116
 117 static pinba_statnode *stat_nodes = NULL;
 118
 119 char service_status=0;
 120 char *service_address = PINBA_DEFAULT_ADDRESS;
 121 unsigned int service_port=PINBA_DEFAULT_PORT;
 122
 123 static void service_statnode_reset (pinba_statnode *node) /* {{{ */
 124 {
 125   node->last_coll=now();
 126   node->req_count=0;
 127   node->req_time=0.0;
 128   node->ru_utime=0.0;
 129   node->ru_stime=0.0;
 130   node->doc_size=0;
 131   node->mem_peak=0;
 132 } /* }}} void service_statnode_reset */
 133
 134 static void strset (char **str, const char *new) /* {{{ */
 135 {
 136   char *tmp;
 137
 138   if (!str || !new)
 139     return;
 140
 141   tmp = strdup (new);
 142   if (tmp == NULL)
 143     return;
 144
 145   sfree (*str);
 146   *str = tmp;
 147 } /* }}} void strset */
 148
 149 static void service_statnode_add(const char *name, /* {{{ */
 150     const char *host,
 151     const char *server,
 152     const char *script)
 153 {
 154   pinba_statnode *node;
 155   DEBUG("adding node `%s' to collector { %s, %s, %s }", name, host?host:"", server?server:"", script?script:"");
 156
 157   stat_nodes=realloc(stat_nodes, sizeof(pinba_statnode)*(stat_nodes_count+1));
 158   if(!stat_nodes){
 159     ERROR("Realloc failed!");
 160     exit(-1);
 161   }
 162
 163   node=&stat_nodes[stat_nodes_count];
 164
 165   /* reset stat data */
 166   service_statnode_reset(node);
 167
 168   /* reset strings */
 169   node->name=NULL;
 170   node->host=NULL;
 171   node->server=NULL;
 172   node->script=NULL;
 173
 174   /* fill query data */
 175   strset(&node->name, name);
 176   strset(&node->host, host);
 177   strset(&node->server, server);
 178   strset(&node->script, script);
 179
 180   /* increment counter */
 181   stat_nodes_count++;
 182 } /* }}} void service_statnode_add */
 183
 184 static void service_statnode_free (void)
 185 {
 186   unsigned int i;
 187
 188   if(stat_nodes_count < 1)
 189     return;
 190
 191   for (i = 0; i < stat_nodes_count; i++)
 192   {
 193     sfree (stat_nodes[i].name);
 194     sfree (stat_nodes[i].host);
 195     sfree (stat_nodes[i].server);
 196     sfree (stat_nodes[i].script);
 197   }
 198
 199   sfree (stat_nodes);
 200   stat_nodes_count = 0;
 201
 202   pthread_rwlock_destroy (&temp_lock);
 203 }
 204
 205 static void service_statnode_init (void)
 206 {
 207   /* only total info collect by default */
 208   service_statnode_free();
 209
 210   DEBUG("initializing collector..");
 211   pthread_rwlock_init(&temp_lock, 0);
 212 }
 213
 214 static void service_statnode_begin (void)
 215 {
 216   service_statnode_init();
 217   pthread_rwlock_wrlock(&temp_lock);
 218
 219   service_statnode_add("total", NULL, NULL, NULL);
 220 }
 221
 222 static void service_statnode_end (void)
 223 {
 224   pthread_rwlock_unlock(&temp_lock);
 225 }
 226
 227 static unsigned int service_statnode_collect (pinba_statres *res, /* {{{ */
 228     unsigned int index)
 229 {
 230   pinba_statnode* node;
 231   pinba_time delta;
 232
 233   if (stat_nodes_count == 0)
 234     return 0;
 235
 236   /* begin collecting */
 237   if (index == 0)
 238     pthread_rwlock_wrlock (&temp_lock);
 239
 240   /* end collecting */
 241   if (index >= stat_nodes_count)
 242   {
 243     pthread_rwlock_unlock (&temp_lock);
 244     return 0;
 245   }
 246
 247   node = stat_nodes + index;
 248   delta = now() - node->last_coll;
 249
 250   res->name = node->name;
 251   res->req_per_sec = node->req_count / delta;
 252
 253   if (node->req_count == 0)
 254     node->req_count = 1;
 255
 256   res->req_time = node->req_time / node->req_count;
 257   res->ru_utime = node->ru_utime / node->req_count;
 258   res->ru_stime = node->ru_stime / node->req_count;
 259   res->ru_stime = node->ru_stime / node->req_count;
 260   res->doc_size = node->doc_size / node->req_count;
 261   res->mem_peak = node->mem_peak / node->req_count;
 262
 263   service_statnode_reset (node);
 264   return (index + 1);
 265 } /* }}} unsigned int service_statnode_collect */
 266
 267 static void service_statnode_process (pinba_statnode *node,
 268     Pinba__Request* request)
 269 {
 270   node->req_count++;
 271   node->req_time+=request->request_time;
 272   node->ru_utime+=request->ru_utime;
 273   node->ru_stime+=request->ru_stime;
 274   node->doc_size+=request->document_size;
 275   node->mem_peak+=request->memory_peak;
 276 }
 277
 278 static void service_process_request (Pinba__Request *request)
 279 {
 280   unsigned int i;
 281
 282   pthread_rwlock_wrlock (&temp_lock);
 283
 284   for (i = 0; i < stat_nodes_count; i++)
 285   {
 286     if(stat_nodes[i].host && strcmp(request->hostname, stat_nodes[i].host))
 287       continue;
 288     if(stat_nodes[i].server && strcmp(request->server_name, stat_nodes[i].server))
 289       continue;
 290     if(stat_nodes[i].script && strcmp(request->script_name, stat_nodes[i].script))
 291       continue;
 292
 293     service_statnode_process(&stat_nodes[i], request);
 294   }
 295
 296   pthread_rwlock_unlock(&temp_lock);
 297 }
 298
 299 static void *pinba_main (void *arg)
 300 {
 301   DEBUG("entering listen-loop..");
 302
 303   service_status=1;
 304   event_base_dispatch(temp_base);
 305
 306   /* unreachable */
 307   return NULL;
 308 }
 309
 310 static void pinba_socket_free (pinba_socket *socket) /* {{{ */
 311 {
 312   if (!socket)
 313     return;
 314
 315   if (socket->listen_sock >= 0)
 316   {
 317     close(socket->listen_sock);
 318     socket->listen_sock = -1;
 319   }
 320
 321   if (socket->accept_event)
 322   {
 323     event_del(socket->accept_event);
 324     free(socket->accept_event);
 325     socket->accept_event = NULL;
 326   }
 327
 328   free(socket);
 329 } /* }}} void pinba_socket_free */
 330
 331 static int pinba_process_stats_packet (const unsigned char *buf,
 332     int buf_len)
 333 {
 334   Pinba__Request *request;
 335
 336   request = pinba__request__unpack(NULL, buf_len, buf);
 337
 338   if (!request) {
 339     return P_FAILURE;
 340   } else {
 341     service_process_request(request);
 342
 343     pinba__request__free_unpacked(request, NULL);
 344
 345     return P_SUCCESS;
 346   }
 347 }
 348
 349 static void pinba_udp_read_callback_fn (int sock, short event, void *arg)
 350 {
 351   if (event & EV_READ) {
 352     int ret;
 353     unsigned char buf[PINBA_UDP_BUFFER_SIZE];
 354     struct sockaddr_in from;
 355     socklen_t fromlen = sizeof(struct sockaddr_in);
 356
 357     ret = recvfrom(sock, buf, PINBA_UDP_BUFFER_SIZE-1, MSG_DONTWAIT, (struct sockaddr *)&from, &fromlen);
 358     if (ret > 0) {
 359       if (pinba_process_stats_packet(buf, ret) != P_SUCCESS) {
 360         DEBUG("failed to parse data received from %s", inet_ntoa(from.sin_addr));
 361       }
 362     } else if (ret < 0) {
 363       if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
 364         return;
 365       }
 366       WARNING("recv() failed: %s (%d)", strerror(errno), errno);
 367     } else {
 368       WARNING("recv() returned 0");
 369     }
 370   }
 371 }
 372
 373 static pinba_socket *pinba_socket_open (const char *ip, /* {{{ */
 374     int listen_port)
 375 {
 376   struct sockaddr_in addr;
 377   pinba_socket *s;
 378   int sfd, flags, yes = 1;
 379
 380   if ((sfd = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
 381     ERROR("socket() failed: %s (%d)", strerror(errno), errno);
 382     return NULL;
 383   }
 384
 385   if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
 386     close(sfd);
 387     return NULL;
 388   }
 389
 390   if(setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) {
 391     close(sfd);
 392     return NULL;
 393   }
 394
 395   s = (pinba_socket *)calloc(1, sizeof(pinba_socket));
 396   if (!s) {
 397     return NULL;
 398   }
 399   s->listen_sock = sfd;
 400
 401   memset(&addr, 0, sizeof(addr));
 402
 403   addr.sin_family = AF_INET;
 404   addr.sin_port = htons(listen_port);
 405   addr.sin_addr.s_addr = htonl(INADDR_ANY);
 406
 407   if (ip && *ip) {
 408     struct in_addr tmp;
 409
 410     if (inet_aton(ip, &tmp)) {
 411       addr.sin_addr.s_addr = tmp.s_addr;
 412     } else {
 413       WARNING("inet_aton(%s) failed, listening on ANY IP-address", ip);
 414     }
 415   }
 416
 417   if (bind(s->listen_sock, (struct sockaddr *)&addr, sizeof(addr))) {
 418     pinba_socket_free(s);
 419     ERROR("bind() failed: %s (%d)", strerror(errno), errno);
 420     return NULL;
 421   }
 422
 423   s->accept_event = (struct event *)calloc(1, sizeof(struct event));
 424   if (!s->accept_event) {
 425     ERROR("calloc() failed: %s (%d)", strerror(errno), errno);
 426     pinba_socket_free(s);
 427     return NULL;
 428   }
 429
 430   event_set(s->accept_event, s->listen_sock, EV_READ | EV_PERSIST, pinba_udp_read_callback_fn, s);
 431   event_base_set(temp_base, s->accept_event);
 432   event_add(s->accept_event, NULL);
 433
 434   return s;
 435 } /* }}} pinba_socket_open */
 436
 437 static int service_cleanup (void)
 438 {
 439   DEBUG("closing socket..");
 440   if(temp_sock){
 441     pthread_rwlock_wrlock(&temp_lock);
 442     pinba_socket_free(temp_sock);
 443     pthread_rwlock_unlock(&temp_lock);
 444   }
 445
 446   DEBUG("shutdowning event..");
 447   event_base_free(temp_base);
 448
 449   DEBUG("shutting down..");
 450
 451   return (0);
 452 }
 453
 454 static int service_start(void)
 455 {
 456   DEBUG("starting up..");
 457
 458   DEBUG("initializing event..");
 459   temp_base = event_base_new();
 460
 461   DEBUG("opening socket..");
 462
 463   temp_sock = pinba_socket_open(service_address, service_port);
 464
 465   if (!temp_sock) {
 466     service_cleanup();
 467     return 1;
 468   }
 469
 470   if (pthread_create(&temp_thrd, NULL, pinba_main, NULL)) {
 471     service_cleanup();
 472     return 1;
 473   }
 474
 475   return 0;
 476 }
 477
 478 static int service_stop (void)
 479 {
 480   pthread_cancel(temp_thrd);
 481   pthread_join(temp_thrd, NULL);
 482   service_status=0;
 483   DEBUG("terminating listen-loop..");
 484
 485   service_cleanup();
 486
 487   return 0;
 488 }
 489
 490 static void service_config (const char *address, unsigned int port) /* {{{ */
 491 {
 492   int need_restart = 0;
 493
 494   if (address && service_address && (strcmp(service_address, address) != 0))
 495   {
 496     strset (&service_address, address);
 497     need_restart++;
 498   }
 499
 500   if ((port > 0) && (port < 65536) && (service_port != port))
 501   {
 502     service_port=port;
 503     need_restart++;
 504   }
 505
 506   if(service_status && need_restart)
 507   {
 508     service_stop();
 509     service_start();
 510   }
 511 } /* }}} void service_config */
 512
 513 /*
 514  * Plugin declaration section
 515  */
 516
 517 static int config_set (char **var, const char *value)
 518 {
 519   /* code from nginx plugin for collectd */
 520   if (*var != NULL) {
 521     free (*var);
 522     *var = NULL;
 523   }
 524
 525   if ((*var = strdup (value)) == NULL) return (1);
 526   else return (0);
 527 }
 528
 529 static int plugin_config (oconfig_item_t *ci)
 530 {
 531   unsigned int i, o;
 532   int pinba_port = 0;
 533   char *pinba_address = NULL;
 534
 535   INFO("Pinba Configure..");
 536
 537   service_statnode_begin();
 538
 539   /* Set default values */
 540   config_set(&pinba_address, PINBA_DEFAULT_ADDRESS);
 541   pinba_port = PINBA_DEFAULT_PORT;
 542
 543   for (i = 0; i < ci->children_num; i++) {
 544     oconfig_item_t *child = ci->children + i;
 545     if (strcasecmp ("Address", child->key) == 0) {
 546       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_STRING)){
 547         WARNING ("pinba plugin: `Address' needs exactly one string argument.");
 548         return (-1);
 549       }
 550       config_set(&pinba_address, child->values[0].value.string);
 551     } else if (strcasecmp ("Port", child->key) == 0) {
 552       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_NUMBER)){
 553         WARNING ("pinba plugin: `Port' needs exactly one number argument.");
 554         return (-1);
 555       }
 556       pinba_port=child->values[0].value.number;
 557     } else if (strcasecmp ("View", child->key) == 0) {
 558       const char *name=NULL, *host=NULL, *server=NULL, *script=NULL;
 559       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_STRING) || strlen(child->values[0].value.string)==0){
 560         WARNING ("pinba plugin: `View' needs exactly one non-empty string argument.");
 561         return (-1);
 562       }
 563       name = child->values[0].value.string;
 564       for(o=0; o<child->children_num; o++){
 565         oconfig_item_t *node = child->children + o;
 566         if (strcasecmp ("Host", node->key) == 0) {
 567           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 568             WARNING ("pinba plugin: `View->Host' needs exactly one non-empty string argument.");
 569             return (-1);
 570           }
 571           host = node->values[0].value.string;
 572         } else if (strcasecmp ("Server", node->key) == 0) {
 573           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 574             WARNING ("pinba plugin: `View->Server' needs exactly one non-empty string argument.");
 575             return (-1);
 576           }
 577           server = node->values[0].value.string;
 578         } else if (strcasecmp ("Script", node->key) == 0) {
 579           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 580             WARNING ("pinba plugin: `View->Script' needs exactly one non-empty string argument.");
 581             return (-1);
 582           }
 583           script = node->values[0].value.string;
 584         } else {
 585           WARNING ("pinba plugin: In `<View>' context allowed only `Host', `Server' and `Script' options but not the `%s'.", node->key);
 586           return (-1);
 587         }
 588       }
 589       /* add new statnode */
 590       service_statnode_add(name, host, server, script);
 591     } else {
 592       WARNING ("pinba plugin: In `<Plugin pinba>' context allowed only `Address', `Port' and `Observe' options but not the `%s'.", child->key);
 593       return (-1);
 594     }
 595   }
 596
 597   service_statnode_end();
 598
 599   service_config(pinba_address, pinba_port);
 600 } /* int pinba_config */
 601
 602 static int plugin_init (void)
 603 {
 604   INFO("Pinba Starting..");
 605   service_start();
 606   return 0;
 607 }
 608
 609 static int plugin_shutdown (void)
 610 {
 611   INFO("Pinba Stopping..");
 612   service_stop();
 613   service_statnode_free();
 614   return 0;
 615 }
 616
 617 static int plugin_submit (const char *plugin_instance,
 618                const char *type,
 619                const pinba_statres *res) {
 620   value_t values[6];
 621   value_list_t vl = VALUE_LIST_INIT;
 622
 623   values[0].gauge = res->req_per_sec;
 624   values[1].gauge = res->req_time;
 625   values[2].gauge = res->ru_utime;
 626   values[3].gauge = res->ru_stime;
 627   values[4].gauge = res->doc_size;
 628   values[5].gauge = res->mem_peak;
 629
 630   vl.values = values;
 631   vl.values_len = 6;
 632   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 633   sstrncpy (vl.plugin, "pinba", sizeof (vl.plugin));
 634   sstrncpy (vl.plugin_instance, plugin_instance,
 635             sizeof(vl.plugin_instance));
 636   sstrncpy (vl.type, type, sizeof (vl.type));
 637   INFO("Pinba Dispatch");
 638   plugin_dispatch_values (&vl);
 639
 640   return (0);
 641 }
 642
 643 static int plugin_read (void)
 644 {
 645   unsigned int i=0;
 646   static pinba_statres res;
 647
 648   while ((i = service_statnode_collect (&res, i)) != 0)
 649   {
 650     plugin_submit(res.name, "pinba_view", &res);
 651   }
 652
 653   return 0;
 654 }
 655
 656 void module_register (void)
 657 {
 658   plugin_register_complex_config ("pinba", plugin_config);
 659   plugin_register_init ("pinba", plugin_init);
 660   plugin_register_read ("pinba", plugin_read);
 661   plugin_register_shutdown ("pinba", plugin_shutdown);
 662 } /* void module_register */
 663
 664 /* vim: set sw=2 sts=2 et fdm=marker : */