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