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 i)
 229 {
 230   pinba_statnode* node;
 231
 232   if(stat_nodes_count==0) return 0;
 233
 234   /* begin collecting */
 235   if(i==0){
 236     pthread_rwlock_wrlock(&temp_lock);
 237   }
 238
 239   /* find non-empty node */
 240   //for(node=stat_nodes+i; node->req_count==0 && ++i<stat_nodes_count; node=stat_nodes+i);
 241
 242   /* end collecting */
 243   if(i>=stat_nodes_count){
 244     pthread_rwlock_unlock(&temp_lock);
 245     return 0;
 246   }
 247
 248   node=stat_nodes+i;
 249
 250   pinba_time delta=now()-node->last_coll;
 251
 252   res->name=node->name;
 253
 254   res->req_per_sec=node->req_count/delta;
 255
 256   if(node->req_count==0)node->req_count=1;
 257   res->req_time=node->req_time/node->req_count;
 258   res->ru_utime=node->ru_utime/node->req_count;
 259   res->ru_stime=node->ru_stime/node->req_count;
 260   res->ru_stime=node->ru_stime/node->req_count;
 261   res->doc_size=node->doc_size/node->req_count;
 262   res->mem_peak=node->mem_peak/node->req_count;
 263
 264   service_statnode_reset(node);
 265   return ++i;
 266 }
 267
 268 static void service_statnode_process (pinba_statnode *node,
 269     Pinba__Request* request)
 270 {
 271   node->req_count++;
 272   node->req_time+=request->request_time;
 273   node->ru_utime+=request->ru_utime;
 274   node->ru_stime+=request->ru_stime;
 275   node->doc_size+=request->document_size;
 276   node->mem_peak+=request->memory_peak;
 277 }
 278
 279 static void service_process_request (Pinba__Request *request)
 280 {
 281   unsigned int i;
 282
 283   pthread_rwlock_wrlock (&temp_lock);
 284
 285   for (i = 0; i < stat_nodes_count; i++)
 286   {
 287     if(stat_nodes[i].host && strcmp(request->hostname, stat_nodes[i].host))
 288       continue;
 289     if(stat_nodes[i].server && strcmp(request->server_name, stat_nodes[i].server))
 290       continue;
 291     if(stat_nodes[i].script && strcmp(request->script_name, stat_nodes[i].script))
 292       continue;
 293
 294     service_statnode_process(&stat_nodes[i], request);
 295   }
 296
 297   pthread_rwlock_unlock(&temp_lock);
 298 }
 299
 300 static void *pinba_main (void *arg)
 301 {
 302   DEBUG("entering listen-loop..");
 303
 304   service_status=1;
 305   event_base_dispatch(temp_base);
 306
 307   /* unreachable */
 308   return NULL;
 309 }
 310
 311 static void pinba_socket_free (pinba_socket *socket) /* {{{ */
 312 {
 313   if (!socket)
 314     return;
 315
 316   if (socket->listen_sock >= 0)
 317   {
 318     close(socket->listen_sock);
 319     socket->listen_sock = -1;
 320   }
 321
 322   if (socket->accept_event)
 323   {
 324     event_del(socket->accept_event);
 325     free(socket->accept_event);
 326     socket->accept_event = NULL;
 327   }
 328
 329   free(socket);
 330 } /* }}} void pinba_socket_free */
 331
 332 static int pinba_process_stats_packet (const unsigned char *buf,
 333     int buf_len)
 334 {
 335   Pinba__Request *request;
 336
 337   request = pinba__request__unpack(NULL, buf_len, buf);
 338
 339   if (!request) {
 340     return P_FAILURE;
 341   } else {
 342     service_process_request(request);
 343
 344     pinba__request__free_unpacked(request, NULL);
 345
 346     return P_SUCCESS;
 347   }
 348 }
 349
 350 static void pinba_udp_read_callback_fn (int sock, short event, void *arg)
 351 {
 352   if (event & EV_READ) {
 353     int ret;
 354     unsigned char buf[PINBA_UDP_BUFFER_SIZE];
 355     struct sockaddr_in from;
 356     socklen_t fromlen = sizeof(struct sockaddr_in);
 357
 358     ret = recvfrom(sock, buf, PINBA_UDP_BUFFER_SIZE-1, MSG_DONTWAIT, (struct sockaddr *)&from, &fromlen);
 359     if (ret > 0) {
 360       if (pinba_process_stats_packet(buf, ret) != P_SUCCESS) {
 361         DEBUG("failed to parse data received from %s", inet_ntoa(from.sin_addr));
 362       }
 363     } else if (ret < 0) {
 364       if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
 365         return;
 366       }
 367       WARNING("recv() failed: %s (%d)", strerror(errno), errno);
 368     } else {
 369       WARNING("recv() returned 0");
 370     }
 371   }
 372 }
 373
 374 static pinba_socket *pinba_socket_open (const char *ip, /* {{{ */
 375     int listen_port)
 376 {
 377   struct sockaddr_in addr;
 378   pinba_socket *s;
 379   int sfd, flags, yes = 1;
 380
 381   if ((sfd = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
 382     ERROR("socket() failed: %s (%d)", strerror(errno), errno);
 383     return NULL;
 384   }
 385
 386   if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
 387     close(sfd);
 388     return NULL;
 389   }
 390
 391   if(setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) {
 392     close(sfd);
 393     return NULL;
 394   }
 395
 396   s = (pinba_socket *)calloc(1, sizeof(pinba_socket));
 397   if (!s) {
 398     return NULL;
 399   }
 400   s->listen_sock = sfd;
 401
 402   memset(&addr, 0, sizeof(addr));
 403
 404   addr.sin_family = AF_INET;
 405   addr.sin_port = htons(listen_port);
 406   addr.sin_addr.s_addr = htonl(INADDR_ANY);
 407
 408   if (ip && *ip) {
 409     struct in_addr tmp;
 410
 411     if (inet_aton(ip, &tmp)) {
 412       addr.sin_addr.s_addr = tmp.s_addr;
 413     } else {
 414       WARNING("inet_aton(%s) failed, listening on ANY IP-address", ip);
 415     }
 416   }
 417
 418   if (bind(s->listen_sock, (struct sockaddr *)&addr, sizeof(addr))) {
 419     pinba_socket_free(s);
 420     ERROR("bind() failed: %s (%d)", strerror(errno), errno);
 421     return NULL;
 422   }
 423
 424   s->accept_event = (struct event *)calloc(1, sizeof(struct event));
 425   if (!s->accept_event) {
 426     ERROR("calloc() failed: %s (%d)", strerror(errno), errno);
 427     pinba_socket_free(s);
 428     return NULL;
 429   }
 430
 431   event_set(s->accept_event, s->listen_sock, EV_READ | EV_PERSIST, pinba_udp_read_callback_fn, s);
 432   event_base_set(temp_base, s->accept_event);
 433   event_add(s->accept_event, NULL);
 434
 435   return s;
 436 } /* }}} */
 437
 438 static int service_cleanup (void)
 439 {
 440   DEBUG("closing socket..");
 441   if(temp_sock){
 442     pthread_rwlock_wrlock(&temp_lock);
 443     pinba_socket_free(temp_sock);
 444     pthread_rwlock_unlock(&temp_lock);
 445   }
 446
 447   DEBUG("shutdowning event..");
 448   event_base_free(temp_base);
 449
 450   DEBUG("shutting down..");
 451
 452   return (0);
 453 }
 454
 455 static int service_start(void)
 456 {
 457   DEBUG("starting up..");
 458
 459   DEBUG("initializing event..");
 460   temp_base = event_base_new();
 461
 462   DEBUG("opening socket..");
 463
 464   temp_sock = pinba_socket_open(service_address, service_port);
 465
 466   if (!temp_sock) {
 467     service_cleanup();
 468     return 1;
 469   }
 470
 471   if (pthread_create(&temp_thrd, NULL, pinba_main, NULL)) {
 472     service_cleanup();
 473     return 1;
 474   }
 475
 476   return 0;
 477 }
 478
 479 static int service_stop (void)
 480 {
 481   pthread_cancel(temp_thrd);
 482   pthread_join(temp_thrd, NULL);
 483   service_status=0;
 484   DEBUG("terminating listen-loop..");
 485
 486   service_cleanup();
 487
 488   return 0;
 489 }
 490
 491 static void service_config (const char *address, unsigned int port) /* {{{ */
 492 {
 493   int need_restart = 0;
 494
 495   if (address && service_address && (strcmp(service_address, address) != 0))
 496   {
 497     strset (&service_address, address);
 498     need_restart++;
 499   }
 500
 501   if ((port > 0) && (port < 65536) && (service_port != port))
 502   {
 503     service_port=port;
 504     need_restart++;
 505   }
 506
 507   if(service_status && need_restart)
 508   {
 509     service_stop();
 510     service_start();
 511   }
 512 } /* }}} void service_config */
 513
 514 /*
 515  * Plugin declaration section
 516  */
 517
 518 static int config_set (char **var, const char *value)
 519 {
 520   /* code from nginx plugin for collectd */
 521   if (*var != NULL) {
 522     free (*var);
 523     *var = NULL;
 524   }
 525
 526   if ((*var = strdup (value)) == NULL) return (1);
 527   else return (0);
 528 }
 529
 530 static int plugin_config (oconfig_item_t *ci)
 531 {
 532   unsigned int i, o;
 533   int pinba_port = 0;
 534   char *pinba_address = NULL;
 535
 536   INFO("Pinba Configure..");
 537
 538   service_statnode_begin();
 539
 540   /* Set default values */
 541   config_set(&pinba_address, PINBA_DEFAULT_ADDRESS);
 542   pinba_port = PINBA_DEFAULT_PORT;
 543
 544   for (i = 0; i < ci->children_num; i++) {
 545     oconfig_item_t *child = ci->children + i;
 546     if (strcasecmp ("Address", child->key) == 0) {
 547       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_STRING)){
 548         WARNING ("pinba plugin: `Address' needs exactly one string argument.");
 549         return (-1);
 550       }
 551       config_set(&pinba_address, child->values[0].value.string);
 552     } else if (strcasecmp ("Port", child->key) == 0) {
 553       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_NUMBER)){
 554         WARNING ("pinba plugin: `Port' needs exactly one number argument.");
 555         return (-1);
 556       }
 557       pinba_port=child->values[0].value.number;
 558     } else if (strcasecmp ("View", child->key) == 0) {
 559       const char *name=NULL, *host=NULL, *server=NULL, *script=NULL;
 560       if ((child->values_num != 1) || (child->values[0].type != OCONFIG_TYPE_STRING) || strlen(child->values[0].value.string)==0){
 561         WARNING ("pinba plugin: `View' needs exactly one non-empty string argument.");
 562         return (-1);
 563       }
 564       name = child->values[0].value.string;
 565       for(o=0; o<child->children_num; o++){
 566         oconfig_item_t *node = child->children + o;
 567         if (strcasecmp ("Host", node->key) == 0) {
 568           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 569             WARNING ("pinba plugin: `View->Host' needs exactly one non-empty string argument.");
 570             return (-1);
 571           }
 572           host = node->values[0].value.string;
 573         } else if (strcasecmp ("Server", node->key) == 0) {
 574           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 575             WARNING ("pinba plugin: `View->Server' needs exactly one non-empty string argument.");
 576             return (-1);
 577           }
 578           server = node->values[0].value.string;
 579         } else if (strcasecmp ("Script", node->key) == 0) {
 580           if ((node->values_num != 1) || (node->values[0].type != OCONFIG_TYPE_STRING) || strlen(node->values[0].value.string)==0){
 581             WARNING ("pinba plugin: `View->Script' needs exactly one non-empty string argument.");
 582             return (-1);
 583           }
 584           script = node->values[0].value.string;
 585         } else {
 586           WARNING ("pinba plugin: In `<View>' context allowed only `Host', `Server' and `Script' options but not the `%s'.", node->key);
 587           return (-1);
 588         }
 589       }
 590       /* add new statnode */
 591       service_statnode_add(name, host, server, script);
 592     } else {
 593       WARNING ("pinba plugin: In `<Plugin pinba>' context allowed only `Address', `Port' and `Observe' options but not the `%s'.", child->key);
 594       return (-1);
 595     }
 596   }
 597
 598   service_statnode_end();
 599
 600   service_config(pinba_address, pinba_port);
 601 } /* int pinba_config */
 602
 603 static int plugin_init (void)
 604 {
 605   INFO("Pinba Starting..");
 606   service_start();
 607   return 0;
 608 }
 609
 610 static int plugin_shutdown (void)
 611 {
 612   INFO("Pinba Stopping..");
 613   service_stop();
 614   service_statnode_free();
 615   return 0;
 616 }
 617
 618 static int
 619 plugin_submit (const char *plugin_instance,
 620                const char *type,
 621                const pinba_statres *res) {
 622   value_t values[6];
 623   value_list_t vl = VALUE_LIST_INIT;
 624
 625   values[0].gauge = res->req_per_sec;
 626   values[1].gauge = res->req_time;
 627   values[2].gauge = res->ru_utime;
 628   values[3].gauge = res->ru_stime;
 629   values[4].gauge = res->doc_size;
 630   values[5].gauge = res->mem_peak;
 631
 632   vl.values = values;
 633   vl.values_len = 6;
 634   sstrncpy (vl.host, hostname_g, sizeof (vl.host));
 635   sstrncpy (vl.plugin, "pinba", sizeof (vl.plugin));
 636   sstrncpy (vl.plugin_instance, plugin_instance,
 637             sizeof(vl.plugin_instance));
 638   sstrncpy (vl.type, type, sizeof (vl.type));
 639   INFO("Pinba Dispatch");
 640   plugin_dispatch_values (&vl);
 641
 642   return (0);
 643 }
 644
 645 static int plugin_read (void)
 646 {
 647   unsigned int i=0;
 648   static pinba_statres res;
 649
 650   while ((i = service_statnode_collect (&res, i)) != 0)
 651   {
 652     plugin_submit(res.name, "pinba_view", &res);
 653   }
 654
 655   return 0;
 656 }
 657
 658 void module_register (void)
 659 {
 660   plugin_register_complex_config ("pinba", plugin_config);
 661   plugin_register_init ("pinba", plugin_init);
 662   plugin_register_read ("pinba", plugin_read);
 663   plugin_register_shutdown ("pinba", plugin_shutdown);
 664 } /* void module_register */
 665
 666 /* vim: set sw=2 sts=2 et : */