src/pinba.c

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