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 uint8_t *buffer, /* {{{ */
 332     size_t buffer_size)
 333 {
 334   Pinba__Request *request;
 335
 336   request = pinba__request__unpack (NULL, buffer_size, buffer);
 337
 338   if (!request)
 339     return (-1);
 340
 341   service_process_request(request);
 342   pinba__request__free_unpacked (request, NULL);
 343
 344   return (0);
 345 } /* }}} int pinba_process_stats_packet */
 346
 347 static void pinba_udp_read_callback_fn (int sock, short event, void *arg) /* {{{ */
 348 {
 349   uint8_t buffer[PINBA_UDP_BUFFER_SIZE];
 350   size_t buffer_size;
 351   int status;
 352
 353   if ((event & EV_READ) == 0)
 354     return;
 355
 356   while (42)
 357   {
 358     buffer_size = sizeof (buffer);
 359     status = recvfrom (sock, buffer, buffer_size - 1, MSG_DONTWAIT, /* from = */ NULL, /* from len = */ 0);
 360     if (status < 0)
 361     {
 362       char errbuf[1024];
 363
 364       if ((errno == EINTR)
 365 #ifdef EWOULDBLOCK
 366           || (errno == EWOULDBLOCK)
 367 #endif
 368           || (errno == EAGAIN))
 369       {
 370         continue;
 371       }
 372
 373       WARNING("pinba plugin: recvfrom(2) failed: %s",
 374           sstrerror (errno, errbuf, sizeof (errbuf)));
 375       return;
 376     }
 377     else if (status == 0)
 378     {
 379       DEBUG ("pinba plugin: recvfrom(2) returned unexpected status zero.");
 380       return;
 381     }
 382     else /* if (status > 0) */
 383     {
 384       assert (((size_t) status) < buffer_size);
 385       buffer_size = (size_t) status;
 386       buffer[buffer_size] = 0;
 387
 388       status = pinba_process_stats_packet (buffer, buffer_size);
 389       if (status != 0)
 390         DEBUG("pinba plugin: Parsing packet failed.");
 391       return;
 392     }
 393
 394     /* not reached */
 395     assert (23 == 42);
 396   } /* while (42) */
 397 } /* }}} void pinba_udp_read_callback_fn */
 398
 399 static pinba_socket *pinba_socket_open (const char *ip, /* {{{ */
 400     int listen_port)
 401 {
 402   struct sockaddr_in addr;
 403   pinba_socket *s;
 404   int sfd, flags, yes = 1;
 405
 406   if ((sfd = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
 407     ERROR("socket() failed: %s (%d)", strerror(errno), errno);
 408     return NULL;
 409   }
 410
 411   if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
 412     close(sfd);
 413     return NULL;
 414   }
 415
 416   if(setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) {
 417     close(sfd);
 418     return NULL;
 419   }
 420
 421   s = (pinba_socket *)calloc(1, sizeof(pinba_socket));
 422   if (!s) {
 423     return NULL;
 424   }
 425   s->listen_sock = sfd;
 426
 427   memset(&addr, 0, sizeof(addr));
 428
 429   addr.sin_family = AF_INET;
 430   addr.sin_port = htons(listen_port);
 431   addr.sin_addr.s_addr = htonl(INADDR_ANY);
 432
 433   if (ip && *ip) {
 434     struct in_addr tmp;
 435
 436     if (inet_aton(ip, &tmp)) {
 437       addr.sin_addr.s_addr = tmp.s_addr;
 438     } else {
 439       WARNING("inet_aton(%s) failed, listening on ANY IP-address", ip);
 440     }
 441   }
 442
 443   if (bind(s->listen_sock, (struct sockaddr *)&addr, sizeof(addr))) {
 444     pinba_socket_free(s);
 445     ERROR("bind() failed: %s (%d)", strerror(errno), errno);
 446     return NULL;
 447   }
 448
 449   s->accept_event = (struct event *)calloc(1, sizeof(struct event));
 450   if (!s->accept_event) {
 451     ERROR("calloc() failed: %s (%d)", strerror(errno), errno);
 452     pinba_socket_free(s);
 453     return NULL;
 454   }
 455
 456   event_set(s->accept_event, s->listen_sock, EV_READ | EV_PERSIST, pinba_udp_read_callback_fn, s);
 457   event_base_set(temp_base, s->accept_event);
 458   event_add(s->accept_event, NULL);
 459
 460   return s;
 461 } /* }}} pinba_socket_open */
 462
 463 static int service_cleanup (void)
 464 {
 465   DEBUG("closing socket..");
 466   if(temp_sock){
 467     pthread_rwlock_wrlock(&temp_lock);
 468     pinba_socket_free(temp_sock);
 469     pthread_rwlock_unlock(&temp_lock);
 470   }
 471
 472   DEBUG("shutdowning event..");
 473   event_base_free(temp_base);
 474
 475   DEBUG("shutting down..");
 476
 477   return (0);
 478 }
 479
 480 static int service_start(void)
 481 {
 482   DEBUG("starting up..");
 483
 484   DEBUG("initializing event..");
 485   temp_base = event_base_new();
 486
 487   DEBUG("opening socket..");
 488
 489   temp_sock = pinba_socket_open(service_address, service_port);
 490
 491   if (!temp_sock) {
 492     service_cleanup();
 493     return 1;
 494   }
 495
 496   if (pthread_create(&temp_thrd, NULL, pinba_main, NULL)) {
 497     service_cleanup();
 498     return 1;
 499   }
 500
 501   return 0;
 502 }
 503
 504 static int service_stop (void)
 505 {
 506   pthread_cancel(temp_thrd);
 507   pthread_join(temp_thrd, NULL);
 508   service_status=0;
 509   DEBUG("terminating listen-loop..");
 510
 511   service_cleanup();
 512
 513   return 0;
 514 }
 515
 516 static void service_config (const char *address, unsigned int port) /* {{{ */
 517 {
 518   int need_restart = 0;
 519
 520   if (address && service_address && (strcmp(service_address, address) != 0))
 521   {
 522     strset (&service_address, address);
 523     need_restart++;
 524   }
 525
 526   if ((port > 0) && (port < 65536) && (service_port != port))
 527   {
 528     service_port=port;
 529     need_restart++;
 530   }
 531
 532   if(service_status && need_restart)
 533   {
 534     service_stop();
 535     service_start();
 536   }
 537 } /* }}} void service_config */
 538
 539 /*
 540  * Plugin declaration section
 541  */
 542
 543 static int config_set (char **var, const char *value)
 544 {
 545   /* code from nginx plugin for collectd */
 546   if (*var != NULL) {
 547     free (*var);
 548     *var = NULL;
 549   }
 550
 551   if ((*var = strdup (value)) == NULL) return (1);
 552   else return (0);
 553 }
 554
 555 static int plugin_config (oconfig_item_t *ci)
 556 {
 557   unsigned int i, o;
 558   int pinba_port = 0;
 559   char *pinba_address = NULL;
 560
 561   INFO("Pinba Configure..");
 562
 563   service_statnode_begin();
 564
 565   /* Set default values */
 566   config_set(&pinba_address, PINBA_DEFAULT_ADDRESS);
 567   pinba_port = PINBA_DEFAULT_PORT;
 568
 569   for (i = 0; i < ci->children_num; i++) {
 570     oconfig_item_t *child = ci->children + i;
 571     if (strcasecmp ("Address", child->key) == 0) {
 572       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_STRING)){
 573         WARNING ("pinba plugin: `Address' needs exactly one string argument.");
 574         return (-1);
 575       }
 576       config_set(&pinba_address, child->values[0].value.string);
 577     } else if (strcasecmp ("Port", child->key) == 0) {
 578       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_NUMBER)){
 579         WARNING ("pinba plugin: `Port' needs exactly one number argument.");
 580         return (-1);
 581       }
 582       pinba_port=child->values[0].value.number;
 583     } else if (strcasecmp ("View", child->key) == 0) {
 584       const char *name=NULL, *host=NULL, *server=NULL, *script=NULL;
 585       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_STRING) || strlen(child->values[0].value.string)==0){
 586         WARNING ("pinba plugin: `View' needs exactly one non-empty string argument.");
 587         return (-1);
 588       }
 589       name = child->values[0].value.string;
 590       for(o=0; o<child->children_num; o++){
 591         oconfig_item_t *node = child->children + o;
 592         if (strcasecmp ("Host", node->key) == 0) {
 593           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 594             WARNING ("pinba plugin: `View->Host' needs exactly one non-empty string argument.");
 595             return (-1);
 596           }
 597           host = node->values[0].value.string;
 598         } else if (strcasecmp ("Server", node->key) == 0) {
 599           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 600             WARNING ("pinba plugin: `View->Server' needs exactly one non-empty string argument.");
 601             return (-1);
 602           }
 603           server = node->values[0].value.string;
 604         } else if (strcasecmp ("Script", node->key) == 0) {
 605           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 606             WARNING ("pinba plugin: `View->Script' needs exactly one non-empty string argument.");
 607             return (-1);
 608           }
 609           script = node->values[0].value.string;
 610         } else {
 611           WARNING ("pinba plugin: In `<View>' context allowed only `Host', `Server' and `Script' options but not the `%s'.", node->key);
 612           return (-1);
 613         }
 614       }
 615       /* add new statnode */
 616       service_statnode_add(name, host, server, script);
 617     } else {
 618       WARNING ("pinba plugin: In `<Plugin pinba>' context allowed only `Address', `Port' and `Observe' options but not the `%s'.", child->key);
 619       return (-1);
 620     }
 621   }
 622
 623   service_statnode_end();
 624
 625   service_config(pinba_address, pinba_port);
 626 } /* int pinba_config */
 627
 628 static int plugin_init (void)
 629 {
 630   INFO("Pinba Starting..");
 631   service_start();
 632   return 0;
 633 }
 634
 635 static int plugin_shutdown (void)
 636 {
 637   INFO("Pinba Stopping..");
 638   service_stop();
 639   service_statnode_free();
 640   return 0;
 641 }
 642
 643 static int plugin_submit (const char *plugin_instance,
 644                const char *type,
 645                const pinba_statres *res) {
 646   value_t values[6];
 647   value_list_t vl = VALUE_LIST_INIT;
 648
 649   values[0].gauge = res->req_per_sec;
 650   values[1].gauge = res->req_time;
 651   values[2].gauge = res->ru_utime;
 652   values[3].gauge = res->ru_stime;
 653   values[4].gauge = res->doc_size;
 654   values[5].gauge = res->mem_peak;
 655
 656   vl.values = values;
 657   vl.values_len = 6;
 658   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 659   sstrncpy (vl.plugin, "pinba", sizeof (vl.plugin));
 660   sstrncpy (vl.plugin_instance, plugin_instance,
 661             sizeof(vl.plugin_instance));
 662   sstrncpy (vl.type, type, sizeof (vl.type));
 663   INFO("Pinba Dispatch");
 664   plugin_dispatch_values (&vl);
 665
 666   return (0);
 667 }
 668
 669 static int plugin_read (void)
 670 {
 671   unsigned int i=0;
 672   static pinba_statres res;
 673
 674   while ((i = service_statnode_collect (&res, i)) != 0)
 675   {
 676     plugin_submit(res.name, "pinba_view", &res);
 677   }
 678
 679   return 0;
 680 }
 681
 682 void module_register (void)
 683 {
 684   plugin_register_complex_config ("pinba", plugin_config);
 685   plugin_register_init ("pinba", plugin_init);
 686   plugin_register_read ("pinba", plugin_read);
 687   plugin_register_shutdown ("pinba", plugin_shutdown);
 688 } /* void module_register */
 689
 690 /* vim: set sw=2 sts=2 et fdm=marker : */