src/ceph.c

   1 /**
   2  * collectd - src/ceph.c
   3  * Copyright (C) 2011  New Dream Network
   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  *   Colin McCabe <cmccabe@alumni.cmu.edu>
  20  *   Dennis Zou <yunzou@cisco.com>
  21  *   Dan Ryder <daryder@cisco.com>
  22  **/
  23
  24 #define _BSD_SOURCE
  25
  26 #include "collectd.h"
  27 #include "common.h"
  28 #include "plugin.h"
  29
  30 #include <arpa/inet.h>
  31 #include <errno.h>
  32 #include <fcntl.h>
  33 #include <yajl/yajl_parse.h>
  34 #if HAVE_YAJL_YAJL_VERSION_H
  35 #include <yajl/yajl_version.h>
  36 #endif
  37
  38 #include <limits.h>
  39 #include <poll.h>
  40 #include <stdint.h>
  41 #include <stdio.h>
  42 #include <stdlib.h>
  43 #include <string.h>
  44 #include <strings.h>
  45 #include <sys/socket.h>
  46 #include <sys/time.h>
  47 #include <sys/types.h>
  48 #include <sys/un.h>
  49 #include <unistd.h>
  50 #include <math.h>
  51 #include <inttypes.h>
  52
  53 #define MAX_RRD_DS_NAME_LEN 20
  54
  55 #define RETRY_AVGCOUNT -1
  56
  57 #if defined(YAJL_MAJOR) && (YAJL_MAJOR > 1)
  58 # define HAVE_YAJL_V2 1
  59 #endif
  60
  61 #define RETRY_ON_EINTR(ret, expr) \
  62     while(1) { \
  63         ret = expr; \
  64         if(ret >= 0) \
  65             break; \
  66         ret = -errno; \
  67         if(ret != -EINTR) \
  68             break; \
  69     }
  70
  71 /** Timeout interval in seconds */
  72 #define CEPH_TIMEOUT_INTERVAL 1
  73
  74 /** Maximum path length for a UNIX domain socket on this system */
  75 #define UNIX_DOMAIN_SOCK_PATH_MAX (sizeof(((struct sockaddr_un*)0)->sun_path))
  76
  77 /** Yajl callback returns */
  78 #define CEPH_CB_CONTINUE 1
  79 #define CEPH_CB_ABORT 0
  80
  81 #if HAVE_YAJL_V2
  82 typedef size_t yajl_len_t;
  83 #else
  84 typedef unsigned int yajl_len_t;
  85 #endif
  86
  87 /******* ceph_daemon *******/
  88 struct ceph_daemon
  89 {
  90     /** Version of the admin_socket interface */
  91     uint32_t version;
  92     /** daemon name **/
  93     char name[DATA_MAX_NAME_LEN];
  94
  95     int dset_num;
  96
  97     /** Path to the socket that we use to talk to the ceph daemon */
  98     char asok_path[UNIX_DOMAIN_SOCK_PATH_MAX];
  99
 100     /** The set of  key/value pairs that this daemon reports
 101      * dset.type        The daemon name
 102      * dset.ds_num      Number of data sources (key/value pairs)
 103      * dset.ds      Dynamically allocated array of key/value pairs
 104      */
 105     /** Dynamically allocated array **/
 106     struct data_set_s *dset;
 107     int **pc_types;
 108 };
 109
 110 /******* JSON parsing *******/
 111 typedef int (*node_handler_t)(void *, const char*, const char*);
 112
 113 /** Track state and handler while parsing JSON */
 114 struct yajl_struct
 115 {
 116     node_handler_t handler;
 117     void * handler_arg;
 118     struct {
 119       char key[DATA_MAX_NAME_LEN];
 120       int key_len;
 121     } state[YAJL_MAX_DEPTH];
 122     int depth;
 123 };
 124 typedef struct yajl_struct yajl_struct;
 125
 126 /**
 127  * Keep track of last data for latency values so we can calculate rate
 128  * since last poll.
 129  */
 130 struct last_data **last_poll_data = NULL;
 131 int last_idx = 0;
 132
 133 enum perfcounter_type_d
 134 {
 135     PERFCOUNTER_LATENCY = 0x4, PERFCOUNTER_DERIVE = 0x8,
 136 };
 137
 138 /** Give user option to use default (long run = since daemon started) avg */
 139 static int long_run_latency_avg = 0;
 140
 141 /**
 142  * Give user option to use default type for special cases -
 143  * filestore.journal_wr_bytes is currently only metric here. Ceph reports the
 144  * type as a sum/count pair and will calculate it the same as a latency value.
 145  * All other "bytes" metrics (excluding the used/capacity bytes for the OSD)
 146  * use the DERIVE type. Unless user specifies to use given type, convert this
 147  * metric to use DERIVE.
 148  */
 149 static int convert_special_metrics = 1;
 150
 151 /** Array of daemons to monitor */
 152 static struct ceph_daemon **g_daemons = NULL;
 153
 154 /** Number of elements in g_daemons */
 155 static int g_num_daemons = 0;
 156
 157 struct values_holder
 158 {
 159     int values_len;
 160     value_t *values;
 161 };
 162
 163 /**
 164  * A set of values_t data that we build up in memory while parsing the JSON.
 165  */
 166 struct values_tmp
 167 {
 168     struct ceph_daemon *d;
 169     int holder_num;
 170     struct values_holder vh[0];
 171     uint64_t avgcount;
 172 };
 173
 174 /**
 175  * A set of count/sum pairs to keep track of latency types and get difference
 176  * between this poll data and last poll data.
 177  */
 178 struct last_data
 179 {
 180     char dset_name[DATA_MAX_NAME_LEN];
 181     char ds_name[MAX_RRD_DS_NAME_LEN];
 182     double last_sum;
 183     uint64_t last_count;
 184 };
 185
 186
 187 /******* network I/O *******/
 188 enum cstate_t
 189 {
 190     CSTATE_UNCONNECTED = 0,
 191     CSTATE_WRITE_REQUEST,
 192     CSTATE_READ_VERSION,
 193     CSTATE_READ_AMT,
 194     CSTATE_READ_JSON,
 195 };
 196
 197 enum request_type_t
 198 {
 199     ASOK_REQ_VERSION = 0,
 200     ASOK_REQ_DATA = 1,
 201     ASOK_REQ_SCHEMA = 2,
 202     ASOK_REQ_NONE = 1000,
 203 };
 204
 205 struct cconn
 206 {
 207     /** The Ceph daemon that we're talking to */
 208     struct ceph_daemon *d;
 209
 210     /** Request type */
 211     uint32_t request_type;
 212
 213     /** The connection state */
 214     enum cstate_t state;
 215
 216     /** The socket we use to talk to this daemon */
 217     int asok;
 218
 219     /** The amount of data remaining to read / write. */
 220     uint32_t amt;
 221
 222     /** Length of the JSON to read */
 223     uint32_t json_len;
 224
 225     /** Buffer containing JSON data */
 226     unsigned char *json;
 227
 228     /** Keep data important to yajl processing */
 229     struct yajl_struct yajl;
 230 };
 231
 232 static int ceph_cb_null(void *ctx)
 233 {
 234     return CEPH_CB_CONTINUE;
 235 }
 236
 237 static int ceph_cb_boolean(void *ctx, int bool_val)
 238 {
 239     return CEPH_CB_CONTINUE;
 240 }
 241
 242 static int
 243 ceph_cb_number(void *ctx, const char *number_val, yajl_len_t number_len)
 244 {
 245     yajl_struct *yajl = (yajl_struct*)ctx;
 246     char buffer[number_len+1];
 247     int i, latency_type = 0, result;
 248     char key[128];
 249
 250     memcpy(buffer, number_val, number_len);
 251     buffer[sizeof(buffer) - 1] = 0;
 252
 253     ssnprintf(key, yajl->state[0].key_len, "%s", yajl->state[0].key);
 254     for(i = 1; i < yajl->depth; i++)
 255     {
 256         if((i == yajl->depth-1) && ((strcmp(yajl->state[i].key,"avgcount") == 0)
 257                 || (strcmp(yajl->state[i].key,"sum") == 0)))
 258         {
 259             if(convert_special_metrics)
 260             {
 261                 /**
 262                  * Special case for filestore:JournalWrBytes. For some reason,
 263                  * Ceph schema encodes this as a count/sum pair while all
 264                  * other "Bytes" data (excluding used/capacity bytes for OSD
 265                  * space) uses a single "Derive" type. To spare further
 266                  * confusion, keep this KPI as the same type of other "Bytes".
 267                  * Instead of keeping an "average" or "rate", use the "sum" in
 268                  * the pair and assign that to the derive value.
 269                  */
 270                 if((strcmp(yajl->state[i-1].key, "journal_wr_bytes") == 0) &&
 271                         (strcmp(yajl->state[i-2].key,"filestore") == 0) &&
 272                         (strcmp(yajl->state[i].key,"avgcount") == 0))
 273                 {
 274                     DEBUG("Skipping avgcount for filestore.JournalWrBytes");
 275                     yajl->depth = (yajl->depth - 1);
 276                     return CEPH_CB_CONTINUE;
 277                 }
 278             }
 279             //probably a avgcount/sum pair. if not - we'll try full key later
 280             latency_type = 1;
 281             break;
 282         }
 283         strncat(key, ".", 1);
 284         strncat(key, yajl->state[i].key, yajl->state[i].key_len+1);
 285     }
 286
 287     result = yajl->handler(yajl->handler_arg, buffer, key);
 288
 289     if((result == RETRY_AVGCOUNT) && latency_type)
 290     {
 291         strncat(key, ".", 1);
 292         strncat(key, yajl->state[yajl->depth-1].key,
 293                 yajl->state[yajl->depth-1].key_len+1);
 294         result = yajl->handler(yajl->handler_arg, buffer, key);
 295     }
 296
 297     if(result == -ENOMEM)
 298     {
 299         ERROR("ceph plugin: memory allocation failed");
 300         return CEPH_CB_ABORT;
 301     }
 302
 303     yajl->depth = (yajl->depth - 1);
 304     return CEPH_CB_CONTINUE;
 305 }
 306
 307 static int ceph_cb_string(void *ctx, const unsigned char *string_val,
 308         yajl_len_t string_len)
 309 {
 310     return CEPH_CB_CONTINUE;
 311 }
 312
 313 static int ceph_cb_start_map(void *ctx)
 314 {
 315     return CEPH_CB_CONTINUE;
 316 }
 317
 318 static int
 319 ceph_cb_map_key(void *ctx, const unsigned char *key, yajl_len_t string_len)
 320 {
 321     yajl_struct *yajl = (yajl_struct*)ctx;
 322
 323     if((yajl->depth+1)  >= YAJL_MAX_DEPTH)
 324     {
 325         ERROR("ceph plugin: depth exceeds max, aborting.");
 326         return CEPH_CB_ABORT;
 327     }
 328
 329     char buffer[string_len+1];
 330
 331     memcpy(buffer, key, string_len);
 332     buffer[sizeof(buffer) - 1] = 0;
 333
 334     snprintf(yajl->state[yajl->depth].key, sizeof(buffer), "%s", buffer);
 335     yajl->state[yajl->depth].key_len = sizeof(buffer);
 336     yajl->depth = (yajl->depth + 1);
 337
 338     return CEPH_CB_CONTINUE;
 339 }
 340
 341 static int ceph_cb_end_map(void *ctx)
 342 {
 343     yajl_struct *yajl = (yajl_struct*)ctx;
 344
 345     yajl->depth = (yajl->depth - 1);
 346     return CEPH_CB_CONTINUE;
 347 }
 348
 349 static int ceph_cb_start_array(void *ctx)
 350 {
 351     return CEPH_CB_CONTINUE;
 352 }
 353
 354 static int ceph_cb_end_array(void *ctx)
 355 {
 356     return CEPH_CB_CONTINUE;
 357 }
 358
 359 static yajl_callbacks callbacks = {
 360         ceph_cb_null,
 361         ceph_cb_boolean,
 362         NULL,
 363         NULL,
 364         ceph_cb_number,
 365         ceph_cb_string,
 366         ceph_cb_start_map,
 367         ceph_cb_map_key,
 368         ceph_cb_end_map,
 369         ceph_cb_start_array,
 370         ceph_cb_end_array
 371 };
 372
 373 static void ceph_daemon_print(const struct ceph_daemon *d)
 374 {
 375     DEBUG("name=%s, asok_path=%s", d->name, d->asok_path);
 376 }
 377
 378 static void ceph_daemons_print(void)
 379 {
 380     int i;
 381     for(i = 0; i < g_num_daemons; ++i)
 382     {
 383         ceph_daemon_print(g_daemons[i]);
 384     }
 385 }
 386
 387 static void ceph_daemon_free(struct ceph_daemon *d)
 388 {
 389     int i = 0;
 390     for(; i < d->dset_num; i++)
 391     {
 392         plugin_unregister_data_set((d->dset + i)->type);
 393         sfree(d->dset->ds);
 394         sfree(d->pc_types[i]);
 395     }
 396     sfree(d->dset);
 397     sfree(d->pc_types);
 398     sfree(d);
 399 }
 400
 401 static void compact_ds_name(char *source, char *dest)
 402 {
 403     int keys_num = 0, i;
 404     char *save_ptr = NULL, *tmp_ptr = source;
 405     char *keys[16];
 406     char len_str[3];
 407     char tmp[DATA_MAX_NAME_LEN];
 408     int reserved = 0;
 409     int offset = 0;
 410     memset(tmp, 0, sizeof(tmp));
 411     if(source == NULL || dest == NULL || source[0] == '\0' || dest[0] != '\0')
 412     {
 413         return;
 414     }
 415     size_t src_len = strlen(source);
 416     snprintf(len_str, sizeof(len_str), "%zu", src_len);
 417     unsigned char append_status = 0x0;
 418     append_status |= (source[src_len - 1] == '-') ? 0x1 : 0x0;
 419     append_status |= (source[src_len - 1] == '+') ? 0x2 : 0x0;
 420     while ((keys[keys_num] = strtok_r(tmp_ptr, ":_-+", &save_ptr)) != NULL)
 421     {
 422         tmp_ptr = NULL;
 423         /** capitalize 1st char **/
 424         keys[keys_num][0] = toupper(keys[keys_num][0]);
 425         keys_num++;
 426         if(keys_num >= 16)
 427         {
 428             break;
 429         }
 430     }
 431     /** concatenate each part of source string **/
 432     for(i = 0; i < keys_num; i++)
 433     {
 434         strcat(tmp, keys[i]);
 435     }
 436     tmp[DATA_MAX_NAME_LEN - 1] = '\0';
 437     /** to coordinate limitation of length of ds name from RRD
 438      *  we will truncate ds_name
 439      *  when the its length is more than
 440      *  MAX_RRD_DS_NAME_LEN
 441      */
 442     if(strlen(tmp) > MAX_RRD_DS_NAME_LEN - 1)
 443     {
 444         append_status |= 0x4;
 445         /** we should reserve space for
 446          * len_str
 447          */
 448         reserved += 2;
 449     }
 450     if(append_status & 0x1)
 451     {
 452         /** we should reserve space for
 453          * "Minus"
 454          */
 455         reserved += 5;
 456     }
 457     if(append_status & 0x2)
 458     {
 459         /** we should reserve space for
 460          * "Plus"
 461          */
 462         reserved += 4;
 463     }
 464     snprintf(dest, MAX_RRD_DS_NAME_LEN - reserved, "%s", tmp);
 465     offset = strlen(dest);
 466     switch (append_status)
 467     {
 468         case 0x1:
 469             memcpy(dest + offset, "Minus", 5);
 470             break;
 471         case 0x2:
 472             memcpy(dest + offset, "Plus", 5);
 473             break;
 474         case 0x4:
 475             memcpy(dest + offset, len_str, 2);
 476             break;
 477         case 0x5:
 478             memcpy(dest + offset, "Minus", 5);
 479             memcpy(dest + offset + 5, len_str, 2);
 480             break;
 481         case 0x6:
 482             memcpy(dest + offset, "Plus", 4);
 483             memcpy(dest + offset + 4, len_str, 2);
 484             break;
 485         default:
 486             break;
 487     }
 488 }
 489 static int parse_keys(const char *key_str, char *dset_name, char *ds_name)
 490 {
 491     char *ptr, *rptr;
 492     size_t dset_name_len = 0;
 493     size_t ds_name_len = 0;
 494     char tmp_ds_name[DATA_MAX_NAME_LEN];
 495     memset(tmp_ds_name, 0, sizeof(tmp_ds_name));
 496     if(dset_name == NULL || ds_name == NULL || key_str == NULL ||
 497             key_str[0] == '\0' || dset_name[0] != '\0' || ds_name[0] != '\0')
 498     {
 499         return -1;
 500     }
 501     if((ptr = strchr(key_str, '.')) == NULL
 502             || (rptr = strrchr(key_str, '.')) == NULL)
 503     {
 504         strncpy(dset_name, key_str, DATA_MAX_NAME_LEN - 1);
 505         strncpy(tmp_ds_name, key_str, DATA_MAX_NAME_LEN - 1);
 506         goto compact;
 507     }
 508     dset_name_len =
 509             (ptr - key_str) > (DATA_MAX_NAME_LEN - 1) ?
 510                     (DATA_MAX_NAME_LEN - 1) : (ptr - key_str);
 511     memcpy(dset_name, key_str, dset_name_len);
 512     ds_name_len =
 513            (rptr - ptr) > DATA_MAX_NAME_LEN ? DATA_MAX_NAME_LEN : (rptr - ptr);
 514     if(ds_name_len == 0)
 515     { /** only have two keys **/
 516         if(!strncmp(rptr + 1, "type", 4))
 517         {/** if last key is "type",ignore **/
 518             strncpy(tmp_ds_name, dset_name, DATA_MAX_NAME_LEN - 1);
 519         }
 520         else
 521         {/** if last key isn't "type", copy last key **/
 522             strncpy(tmp_ds_name, rptr + 1, DATA_MAX_NAME_LEN - 1);
 523         }
 524     }
 525     else if(!strncmp(rptr + 1, "type", 4))
 526     {/** more than two keys **/
 527         memcpy(tmp_ds_name, ptr + 1, ds_name_len - 1);
 528     }
 529     else
 530     {/** copy whole keys **/
 531         strncpy(tmp_ds_name, ptr + 1, DATA_MAX_NAME_LEN - 1);
 532     }
 533     compact: compact_ds_name(tmp_ds_name, ds_name);
 534     return 0;
 535 }
 536
 537 static int get_matching_dset(const struct ceph_daemon *d, const char *name)
 538 {
 539     int idx;
 540     for(idx = 0; idx < d->dset_num; ++idx)
 541     {
 542         if(strcmp(d->dset[idx].type, name) == 0)
 543         {
 544             return idx;
 545         }
 546     }
 547     return -1;
 548 }
 549
 550 static int get_matching_value(const struct data_set_s *dset, const char *name,
 551         int num_values)
 552 {
 553     int idx;
 554     for(idx = 0; idx < num_values; ++idx)
 555     {
 556         if(strcmp(dset->ds[idx].name, name) == 0)
 557         {
 558             return idx;
 559         }
 560     }
 561     return -1;
 562 }
 563
 564 static int ceph_daemon_add_ds_entry(struct ceph_daemon *d, const char *name,
 565         int pc_type)
 566 {
 567     struct data_source_s *ds;
 568     struct data_set_s *dset;
 569     struct data_set_s *dset_array;
 570     int **pc_types_array = NULL;
 571     int *pc_types;
 572     int *pc_types_new;
 573     int idx = 0;
 574     if(strlen(name) + 1 > DATA_MAX_NAME_LEN)
 575     {
 576         return -ENAMETOOLONG;
 577     }
 578     char dset_name[DATA_MAX_NAME_LEN];
 579     char ds_name[MAX_RRD_DS_NAME_LEN];
 580     memset(dset_name, 0, sizeof(dset_name));
 581     memset(ds_name, 0, sizeof(ds_name));
 582     if(parse_keys(name, dset_name, ds_name))
 583     {
 584         return 1;
 585     }
 586     idx = get_matching_dset(d, dset_name);
 587     if(idx == -1)
 588     {/* need to add a dset **/
 589         dset_array = realloc(d->dset,
 590                 sizeof(struct data_set_s) * (d->dset_num + 1));
 591         if(!dset_array)
 592         {
 593             return -ENOMEM;
 594         }
 595         pc_types_array = realloc(d->pc_types,
 596                 sizeof(int *) * (d->dset_num + 1));
 597         if(!pc_types_array)
 598         {
 599             return -ENOMEM;
 600         }
 601         dset = &dset_array[d->dset_num];
 602         /** this step is very important, otherwise,
 603          *  realloc for dset->ds will tricky because of
 604          *  a random addr in dset->ds
 605          */
 606         memset(dset, 0, sizeof(struct data_set_s));
 607         dset->ds_num = 0;
 608         snprintf(dset->type, DATA_MAX_NAME_LEN, "%s", dset_name);
 609         pc_types = pc_types_array[d->dset_num] = NULL;
 610         d->dset = dset_array;
 611     }
 612     else
 613     {
 614         dset = &d->dset[idx];
 615         pc_types = d->pc_types[idx];
 616     }
 617     struct data_source_s *ds_array = realloc(dset->ds,
 618             sizeof(struct data_source_s) * (dset->ds_num + 1));
 619     if(!ds_array)
 620     {
 621         return -ENOMEM;
 622     }
 623     pc_types_new = realloc(pc_types, sizeof(int) * (dset->ds_num + 1));
 624     if(!pc_types_new)
 625     {
 626         return -ENOMEM;
 627     }
 628     dset->ds = ds_array;
 629
 630     if(convert_special_metrics)
 631     {
 632         /**
 633          * Special case for filestore:JournalWrBytes. For some reason, Ceph
 634          * schema encodes this as a count/sum pair while all other "Bytes" data
 635          * (excluding used/capacity bytes for OSD space) uses a single "Derive"
 636          * type. To spare further confusion, keep this KPI as the same type of
 637          * other "Bytes". Instead of keeping an "average" or "rate", use the
 638          * "sum" in the pair and assign that to the derive value.
 639          */
 640         if((strcmp(dset_name,"filestore") == 0) &&
 641                                         strcmp(ds_name, "JournalWrBytes") == 0)
 642         {
 643             pc_type = 10;
 644         }
 645     }
 646
 647     if(idx == -1)
 648     {
 649         pc_types_array[d->dset_num] = pc_types_new;
 650         d->pc_types = pc_types_array;
 651         d->pc_types[d->dset_num][dset->ds_num] = pc_type;
 652         d->dset_num++;
 653     }
 654     else
 655     {
 656         d->pc_types[idx] = pc_types_new;
 657         d->pc_types[idx][dset->ds_num] = pc_type;
 658     }
 659     ds = &ds_array[dset->ds_num++];
 660     snprintf(ds->name, MAX_RRD_DS_NAME_LEN, "%s", ds_name);
 661     ds->type = (pc_type & PERFCOUNTER_DERIVE) ? DS_TYPE_DERIVE : DS_TYPE_GAUGE;
 662
 663     /**
 664      * Use min of 0 for DERIVE types so we don't get negative values on Ceph
 665      * service restart
 666      */
 667     ds->min = (ds->type == DS_TYPE_DERIVE) ? 0 : NAN;
 668     ds->max = NAN;
 669     return 0;
 670 }
 671
 672 /******* ceph_config *******/
 673 static int cc_handle_str(struct oconfig_item_s *item, char *dest, int dest_len)
 674 {
 675     const char *val;
 676     if(item->values_num != 1)
 677     {
 678         return -ENOTSUP;
 679     }
 680     if(item->values[0].type != OCONFIG_TYPE_STRING)
 681     {
 682         return -ENOTSUP;
 683     }
 684     val = item->values[0].value.string;
 685     if(snprintf(dest, dest_len, "%s", val) > (dest_len - 1))
 686     {
 687         ERROR("ceph plugin: configuration parameter '%s' is too long.\n",
 688                 item->key);
 689         return -ENAMETOOLONG;
 690     }
 691     return 0;
 692 }
 693
 694 static int cc_handle_bool(struct oconfig_item_s *item, int *dest)
 695 {
 696     if(item->values_num != 1)
 697     {
 698         return -ENOTSUP;
 699     }
 700
 701     if(item->values[0].type != OCONFIG_TYPE_BOOLEAN)
 702     {
 703         return -ENOTSUP;
 704     }
 705
 706     *dest = (item->values[0].value.boolean) ? 1 : 0;
 707     return 0;
 708 }
 709
 710 static int cc_add_daemon_config(oconfig_item_t *ci)
 711 {
 712     int ret, i;
 713     struct ceph_daemon *array, *nd, cd;
 714     memset(&cd, 0, sizeof(struct ceph_daemon));
 715
 716     if((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING))
 717     {
 718         WARNING("ceph plugin: `Daemon' blocks need exactly one string "
 719                 "argument.");
 720         return (-1);
 721     }
 722
 723     ret = cc_handle_str(ci, cd.name, DATA_MAX_NAME_LEN);
 724     if(ret)
 725     {
 726         return ret;
 727     }
 728
 729     for(i=0; i < ci->children_num; i++)
 730     {
 731         oconfig_item_t *child = ci->children + i;
 732
 733         if(strcasecmp("SocketPath", child->key) == 0)
 734         {
 735             ret = cc_handle_str(child, cd.asok_path, sizeof(cd.asok_path));
 736             if(ret)
 737             {
 738                 return ret;
 739             }
 740         }
 741         else
 742         {
 743             WARNING("ceph plugin: ignoring unknown option %s", child->key);
 744         }
 745     }
 746     if(cd.name[0] == '\0')
 747     {
 748         ERROR("ceph plugin: you must configure a daemon name.\n");
 749         return -EINVAL;
 750     }
 751     else if(cd.asok_path[0] == '\0')
 752     {
 753         ERROR("ceph plugin(name=%s): you must configure an administrative "
 754         "socket path.\n", cd.name);
 755         return -EINVAL;
 756     }
 757     else if(!((cd.asok_path[0] == '/') ||
 758             (cd.asok_path[0] == '.' && cd.asok_path[1] == '/')))
 759     {
 760         ERROR("ceph plugin(name=%s): administrative socket paths must begin "
 761                 "with '/' or './' Can't parse: '%s'\n", cd.name, cd.asok_path);
 762         return -EINVAL;
 763     }
 764     array = realloc(g_daemons,
 765                     sizeof(struct ceph_daemon *) * (g_num_daemons + 1));
 766     if(array == NULL)
 767     {
 768         /* The positive return value here indicates that this is a
 769          * runtime error, not a configuration error.  */
 770         return ENOMEM;
 771     }
 772     g_daemons = (struct ceph_daemon**) array;
 773     nd = malloc(sizeof(struct ceph_daemon));
 774     if(!nd)
 775     {
 776         return ENOMEM;
 777     }
 778     memcpy(nd, &cd, sizeof(struct ceph_daemon));
 779     g_daemons[g_num_daemons++] = nd;
 780     return 0;
 781 }
 782
 783 static int ceph_config(oconfig_item_t *ci)
 784 {
 785     int ret, i;
 786
 787     for(i = 0; i < ci->children_num; ++i)
 788     {
 789         oconfig_item_t *child = ci->children + i;
 790         if(strcasecmp("Daemon", child->key) == 0)
 791         {
 792             ret = cc_add_daemon_config(child);
 793             if(ret)
 794             {
 795                 return ret;
 796             }
 797         }
 798         else if(strcasecmp("LongRunAvgLatency", child->key) == 0)
 799         {
 800             ret = cc_handle_bool(child, &long_run_latency_avg);
 801             if(ret)
 802             {
 803                 return ret;
 804             }
 805         }
 806         else if(strcasecmp("ConvertSpecialMetricTypes", child->key) == 0)
 807         {
 808             ret = cc_handle_bool(child, &convert_special_metrics);
 809             if(ret)
 810             {
 811                 return ret;
 812             }
 813         }
 814         else
 815         {
 816             WARNING("ceph plugin: ignoring unknown option %s", child->key);
 817         }
 818     }
 819     return 0;
 820 }
 821
 822 static int
 823 traverse_json(const unsigned char *json, uint32_t json_len, yajl_handle hand)
 824 {
 825     yajl_status status = yajl_parse(hand, json, json_len);
 826     unsigned char *msg;
 827
 828     switch(status)
 829     {
 830         case yajl_status_error:
 831             msg = yajl_get_error(hand, /* verbose = */ 1,
 832                                        /* jsonText = */ (unsigned char *) json,
 833                                                       (unsigned int) json_len);
 834             ERROR ("ceph plugin: yajl_parse failed: %s", msg);
 835             yajl_free_error(hand, msg);
 836             return 1;
 837         case yajl_status_client_canceled:
 838             return 1;
 839         default:
 840             return 0;
 841     }
 842 }
 843
 844 static int
 845 node_handler_define_schema(void *arg, const char *val, const char *key)
 846 {
 847     struct ceph_daemon *d = (struct ceph_daemon *) arg;
 848     int pc_type;
 849     pc_type = atoi(val);
 850     DEBUG("\nceph_daemon_add_ds_entry(d=%s,key=%s,pc_type=%04x)",
 851             d->name, key, pc_type);
 852     return ceph_daemon_add_ds_entry(d, key, pc_type);
 853 }
 854
 855 static int add_last(const char *dset_n, const char *ds_n, double cur_sum,
 856         uint64_t cur_count)
 857 {
 858     last_poll_data[last_idx] = malloc(1 * sizeof(struct last_data));
 859     if(!last_poll_data[last_idx])
 860     {
 861         return -ENOMEM;
 862     }
 863     sstrncpy(last_poll_data[last_idx]->dset_name,dset_n,
 864             sizeof(last_poll_data[last_idx]->dset_name));
 865     sstrncpy(last_poll_data[last_idx]->ds_name,ds_n,
 866             sizeof(last_poll_data[last_idx]->ds_name));
 867     last_poll_data[last_idx]->last_sum = cur_sum;
 868     last_poll_data[last_idx]->last_count = cur_count;
 869     last_idx++;
 870     return 0;
 871 }
 872
 873 static int update_last(const char *dset_n, const char *ds_n, double cur_sum,
 874         uint64_t cur_count)
 875 {
 876     int i;
 877     for(i = 0; i < last_idx; i++)
 878     {
 879         if(strcmp(last_poll_data[i]->dset_name,dset_n) == 0 &&
 880                 (strcmp(last_poll_data[i]->ds_name,ds_n) == 0))
 881         {
 882             last_poll_data[i]->last_sum = cur_sum;
 883             last_poll_data[i]->last_count = cur_count;
 884             return 0;
 885         }
 886     }
 887
 888     if(!last_poll_data)
 889     {
 890         last_poll_data = malloc(1 * sizeof(struct last_data *));
 891         if(!last_poll_data)
 892         {
 893             return -ENOMEM;
 894         }
 895     }
 896     else
 897     {
 898         struct last_data **tmp_last = realloc(last_poll_data,
 899                 ((last_idx+1) * sizeof(struct last_data *)));
 900         if(!tmp_last)
 901         {
 902             return -ENOMEM;
 903         }
 904         last_poll_data = tmp_last;
 905     }
 906     return add_last(dset_n,ds_n,cur_sum,cur_count);
 907 }
 908
 909 static double get_last_avg(const char *dset_n, const char *ds_n,
 910         double cur_sum, uint64_t cur_count)
 911 {
 912     int i;
 913     double result = -1.1, sum_delt = 0.0;
 914     uint64_t count_delt = 0;
 915     for(i = 0; i < last_idx; i++)
 916     {
 917         if((strcmp(last_poll_data[i]->dset_name,dset_n) == 0) &&
 918                 (strcmp(last_poll_data[i]->ds_name,ds_n) == 0))
 919         {
 920             if(cur_count < last_poll_data[i]->last_count)
 921             {
 922                 break;
 923             }
 924             sum_delt = (cur_sum - last_poll_data[i]->last_sum);
 925             count_delt = (cur_count - last_poll_data[i]->last_count);
 926             result = (sum_delt / count_delt);
 927             break;
 928         }
 929     }
 930
 931     if(result == -1.1)
 932     {
 933         result = NAN;
 934     }
 935     if(update_last(dset_n,ds_n,cur_sum,cur_count) == -ENOMEM)
 936     {
 937         return -ENOMEM;
 938     }
 939     return result;
 940 }
 941
 942 static int node_handler_fetch_data(void *arg, const char *val, const char *key)
 943 {
 944     int dset_idx, ds_idx;
 945     value_t *uv;
 946     char dset_name[DATA_MAX_NAME_LEN];
 947     char ds_name[MAX_RRD_DS_NAME_LEN];
 948     struct values_tmp *vtmp = (struct values_tmp*) arg;
 949     memset(dset_name, 0, sizeof(dset_name));
 950     memset(ds_name, 0, sizeof(ds_name));
 951     if(parse_keys(key, dset_name, ds_name))
 952     {
 953         return 1;DEBUG("enter node_handler_fetch_data");
 954     }
 955     dset_idx = get_matching_dset(vtmp->d, dset_name);
 956     if(dset_idx == -1)
 957     {
 958         return 1;
 959     }
 960     ds_idx = get_matching_value(&vtmp->d->dset[dset_idx], ds_name,
 961             vtmp->d->dset[dset_idx].ds_num);
 962     if(ds_idx == -1)
 963     {
 964         return RETRY_AVGCOUNT;DEBUG("DSet:%s, DS:%s, DSet idx:%d, DS idx:%d",
 965             dset_name,ds_name,dset_idx,ds_idx);
 966     }
 967     uv = &(vtmp->vh[dset_idx].values[ds_idx]);
 968
 969     if(vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_LATENCY)
 970     {
 971         if(vtmp->avgcount == -1)
 972         {
 973             sscanf(val, "%" PRIu64, &vtmp->avgcount);
 974         }
 975         else
 976         {
 977             double sum, result;
 978             sscanf(val, "%lf", &sum);
 979             DEBUG("avgcount:%ld",vtmp->avgcount);
 980             DEBUG("sum:%lf",sum);
 981
 982             if(vtmp->avgcount == 0)
 983             {
 984                 vtmp->avgcount = 1;
 985             }
 986
 987             /** User wants latency values as long run avg */
 988             if(long_run_latency_avg)
 989             {
 990                 result = (sum / vtmp->avgcount);
 991                 DEBUG("uv->gauge = sumd / avgcounti = :%lf", result);
 992             }
 993             else
 994             {
 995                 result = get_last_avg(dset_name, ds_name, sum, vtmp->avgcount);
 996                 if(result == -ENOMEM)
 997                 {
 998                     return -ENOMEM;
 999                 }
1000                 DEBUG("uv->gauge = (sumd_now - sumd_last) / "
1001                         "(avgcounti_now - avgcounti_last) = :%lf", result);
1002             }
1003
1004             uv->gauge = result;
1005             vtmp->avgcount = -1;
1006         }
1007     }
1008     else if(vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_DERIVE)
1009     {
1010         uint64_t derive_val;
1011         sscanf(val, "%" PRIu64, &derive_val);
1012         uv->derive = derive_val;
1013         DEBUG("uv->derive %" PRIu64 "",(uint64_t)uv->derive);
1014     }
1015     else
1016     {
1017         double other_val;
1018         sscanf(val, "%lf", &other_val);
1019         uv->gauge = other_val;
1020         DEBUG("uv->gauge %lf",uv->gauge);
1021     }
1022     return 0;
1023 }
1024
1025 static int cconn_connect(struct cconn *io)
1026 {
1027     struct sockaddr_un address;
1028     int flags, fd, err;
1029     if(io->state != CSTATE_UNCONNECTED)
1030     {
1031         ERROR("cconn_connect: io->state != CSTATE_UNCONNECTED");
1032         return -EDOM;
1033     }
1034     fd = socket(PF_UNIX, SOCK_STREAM, 0);
1035     if(fd < 0)
1036     {
1037         int err = -errno;
1038         ERROR("cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) failed: "
1039         "error %d", err);
1040         return err;
1041     }
1042     memset(&address, 0, sizeof(struct sockaddr_un));
1043     address.sun_family = AF_UNIX;
1044     snprintf(address.sun_path, sizeof(address.sun_path), "%s",
1045             io->d->asok_path);
1046     RETRY_ON_EINTR(err,
1047         connect(fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)));
1048     if(err < 0)
1049     {
1050         ERROR("cconn_connect: connect(%d) failed: error %d", fd, err);
1051         return err;
1052     }
1053
1054     flags = fcntl(fd, F_GETFL, 0);
1055     if(fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
1056     {
1057         err = -errno;
1058         ERROR("cconn_connect: fcntl(%d, O_NONBLOCK) error %d", fd, err);
1059         return err;
1060     }
1061     io->asok = fd;
1062     io->state = CSTATE_WRITE_REQUEST;
1063     io->amt = 0;
1064     io->json_len = 0;
1065     io->json = NULL;
1066     return 0;
1067 }
1068
1069 static void cconn_close(struct cconn *io)
1070 {
1071     io->state = CSTATE_UNCONNECTED;
1072     if(io->asok != -1)
1073     {
1074         int res;
1075         RETRY_ON_EINTR(res, close(io->asok));
1076     }
1077     io->asok = -1;
1078     io->amt = 0;
1079     io->json_len = 0;
1080     sfree(io->json);
1081     io->json = NULL;
1082 }
1083
1084 /* Process incoming JSON counter data */
1085 static int
1086 cconn_process_data(struct cconn *io, yajl_struct *yajl, yajl_handle hand)
1087 {
1088     int i, ret = 0;
1089     struct values_tmp *vtmp = calloc(1, sizeof(struct values_tmp)
1090                     + (sizeof(struct values_holder)) * io->d->dset_num);
1091     if(!vtmp)
1092     {
1093         return -ENOMEM;
1094     }
1095
1096     for(i = 0; i < io->d->dset_num; i++)
1097     {
1098         value_t *val = calloc(1, (sizeof(value_t) * io->d->dset[i].ds_num));
1099         vtmp->vh[i].values = val;
1100         vtmp->vh[i].values_len = io->d->dset[i].ds_num;
1101     }
1102     vtmp->d = io->d;
1103     vtmp->holder_num = io->d->dset_num;
1104     vtmp->avgcount = -1;
1105     yajl->handler_arg = vtmp;
1106     ret = traverse_json(io->json, io->json_len, hand);
1107     if(ret)
1108     {
1109         goto done;
1110     }
1111     for(i = 0; i < vtmp->holder_num; i++)
1112     {
1113         value_list_t vl = VALUE_LIST_INIT;
1114         sstrncpy(vl.host, hostname_g, sizeof(vl.host));
1115         sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
1116         strncpy(vl.plugin_instance, io->d->name, sizeof(vl.plugin_instance));
1117         sstrncpy(vl.type, io->d->dset[i].type, sizeof(vl.type));
1118         vl.values = vtmp->vh[i].values;
1119         vl.values_len = io->d->dset[i].ds_num;
1120         DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
1121                 io->d->name, vl.values_len, io->json);
1122         ret = plugin_dispatch_values(&vl);
1123         if(ret)
1124         {
1125             goto done;
1126         }
1127     }
1128
1129     done: for(i = 0; i < vtmp->holder_num; i++)
1130     {
1131         sfree(vtmp->vh[i].values);
1132     }
1133     sfree(vtmp);
1134     return ret;
1135 }
1136
1137 static int cconn_process_json(struct cconn *io)
1138 {
1139     if((io->request_type != ASOK_REQ_DATA) &&
1140             (io->request_type != ASOK_REQ_SCHEMA))
1141     {
1142         return -EDOM;
1143     }
1144
1145     int result = 1;
1146     yajl_handle hand;
1147     yajl_status status;
1148
1149     hand = yajl_alloc(&callbacks,
1150 #if HAVE_YAJL_V2
1151       /* alloc funcs = */ NULL,
1152 #else
1153       /* alloc funcs = */ NULL, NULL,
1154 #endif
1155       /* context = */ (void *)(&io->yajl));
1156
1157     if(!hand)
1158     {
1159         ERROR ("ceph plugin: yajl_alloc failed.");
1160         return ENOMEM;
1161     }
1162
1163     io->yajl.depth = 0;
1164
1165     switch(io->request_type)
1166     {
1167         case ASOK_REQ_DATA:
1168             io->yajl.handler = node_handler_fetch_data;
1169             result = cconn_process_data(io, &io->yajl, hand);
1170             break;
1171         case ASOK_REQ_SCHEMA:
1172             io->yajl.handler = node_handler_define_schema;
1173             io->yajl.handler_arg = io->d;
1174             result = traverse_json(io->json, io->json_len, hand);
1175             break;
1176     }
1177
1178     if(result)
1179     {
1180         goto done;
1181     }
1182
1183 #if HAVE_YAJL_V2
1184     status = yajl_complete_parse(hand);
1185 #else
1186     status = yajl_parse_complete(hand);
1187 #endif
1188
1189     if (status != yajl_status_ok)
1190     {
1191       unsigned char *errmsg = yajl_get_error (hand, /* verbose = */ 0,
1192           /* jsonText = */ NULL, /* jsonTextLen = */ 0);
1193       ERROR ("ceph plugin: yajl_parse_complete failed: %s",
1194           (char *) errmsg);
1195       yajl_free_error (hand, errmsg);
1196       yajl_free (hand);
1197       return 1;
1198     }
1199
1200     done:
1201     yajl_free (hand);
1202     return result;
1203 }
1204
1205 static int cconn_validate_revents(struct cconn *io, int revents)
1206 {
1207     if(revents & POLLERR)
1208     {
1209         ERROR("cconn_validate_revents(name=%s): got POLLERR", io->d->name);
1210         return -EIO;
1211     }
1212     switch (io->state)
1213     {
1214         case CSTATE_WRITE_REQUEST:
1215             return (revents & POLLOUT) ? 0 : -EINVAL;
1216         case CSTATE_READ_VERSION:
1217         case CSTATE_READ_AMT:
1218         case CSTATE_READ_JSON:
1219             return (revents & POLLIN) ? 0 : -EINVAL;
1220             return (revents & POLLIN) ? 0 : -EINVAL;
1221         default:
1222             ERROR("cconn_validate_revents(name=%s) got to illegal state on "
1223                     "line %d", io->d->name, __LINE__);
1224             return -EDOM;
1225     }
1226 }
1227
1228 /** Handle a network event for a connection */
1229 static int cconn_handle_event(struct cconn *io)
1230 {
1231     int ret;
1232     switch (io->state)
1233     {
1234         case CSTATE_UNCONNECTED:
1235             ERROR("cconn_handle_event(name=%s) got to illegal state on line "
1236                     "%d", io->d->name, __LINE__);
1237
1238             return -EDOM;
1239         case CSTATE_WRITE_REQUEST:
1240         {
1241             char cmd[32];
1242             snprintf(cmd, sizeof(cmd), "%s%d%s", "{ \"prefix\": \"",
1243                     io->request_type, "\" }\n");
1244             size_t cmd_len = strlen(cmd);
1245             RETRY_ON_EINTR(ret,
1246                   write(io->asok, ((char*)&cmd) + io->amt, cmd_len - io->amt));
1247             DEBUG("cconn_handle_event(name=%s,state=%d,amt=%d,ret=%d)",
1248                     io->d->name, io->state, io->amt, ret);
1249             if(ret < 0)
1250             {
1251                 return ret;
1252             }
1253             io->amt += ret;
1254             if(io->amt >= cmd_len)
1255             {
1256                 io->amt = 0;
1257                 switch (io->request_type)
1258                 {
1259                     case ASOK_REQ_VERSION:
1260                         io->state = CSTATE_READ_VERSION;
1261                         break;
1262                     default:
1263                         io->state = CSTATE_READ_AMT;
1264                         break;
1265                 }
1266             }
1267             return 0;
1268         }
1269         case CSTATE_READ_VERSION:
1270         {
1271             RETRY_ON_EINTR(ret,
1272                     read(io->asok, ((char*)(&io->d->version)) + io->amt,
1273                             sizeof(io->d->version) - io->amt));
1274             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1275                     io->d->name, io->state, ret);
1276             if(ret < 0)
1277             {
1278                 return ret;
1279             }
1280             io->amt += ret;
1281             if(io->amt >= sizeof(io->d->version))
1282             {
1283                 io->d->version = ntohl(io->d->version);
1284                 if(io->d->version != 1)
1285                 {
1286                     ERROR("cconn_handle_event(name=%s) not "
1287                     "expecting version %d!", io->d->name, io->d->version);
1288                     return -ENOTSUP;
1289                 }DEBUG("cconn_handle_event(name=%s): identified as "
1290                         "version %d", io->d->name, io->d->version);
1291                 io->amt = 0;
1292                 cconn_close(io);
1293                 io->request_type = ASOK_REQ_SCHEMA;
1294             }
1295             return 0;
1296         }
1297         case CSTATE_READ_AMT:
1298         {
1299             RETRY_ON_EINTR(ret,
1300                     read(io->asok, ((char*)(&io->json_len)) + io->amt,
1301                             sizeof(io->json_len) - io->amt));
1302             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1303                     io->d->name, io->state, ret);
1304             if(ret < 0)
1305             {
1306                 return ret;
1307             }
1308             io->amt += ret;
1309             if(io->amt >= sizeof(io->json_len))
1310             {
1311                 io->json_len = ntohl(io->json_len);
1312                 io->amt = 0;
1313                 io->state = CSTATE_READ_JSON;
1314                 io->json = calloc(1, io->json_len + 1);
1315                 if(!io->json)
1316                 {
1317                     ERROR("ERR CALLOCING IO->JSON");
1318                     return -ENOMEM;
1319                 }
1320             }
1321             return 0;
1322         }
1323         case CSTATE_READ_JSON:
1324         {
1325             RETRY_ON_EINTR(ret,
1326                    read(io->asok, io->json + io->amt, io->json_len - io->amt));
1327             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1328                     io->d->name, io->state, ret);
1329             if(ret < 0)
1330             {
1331                 return ret;
1332             }
1333             io->amt += ret;
1334             if(io->amt >= io->json_len)
1335             {
1336                 ret = cconn_process_json(io);
1337                 if(ret)
1338                 {
1339                     return ret;
1340                 }
1341                 cconn_close(io);
1342                 io->request_type = ASOK_REQ_NONE;
1343             }
1344             return 0;
1345         }
1346         default:
1347             ERROR("cconn_handle_event(name=%s) got to illegal state on "
1348             "line %d", io->d->name, __LINE__);
1349             return -EDOM;
1350     }
1351 }
1352
1353 static int cconn_prepare(struct cconn *io, struct pollfd* fds)
1354 {
1355     int ret;
1356     if(io->request_type == ASOK_REQ_NONE)
1357     {
1358         /* The request has already been serviced. */
1359         return 0;
1360     }
1361     else if((io->request_type == ASOK_REQ_DATA) && (io->d->dset_num == 0))
1362     {
1363         /* If there are no counters to report on, don't bother
1364          * connecting */
1365         return 0;
1366     }
1367
1368     switch (io->state)
1369     {
1370         case CSTATE_UNCONNECTED:
1371             ret = cconn_connect(io);
1372             if(ret > 0)
1373             {
1374                 return -ret;
1375             }
1376             else if(ret < 0)
1377             {
1378                 return ret;
1379             }
1380             fds->fd = io->asok;
1381             fds->events = POLLOUT;
1382             return 1;
1383         case CSTATE_WRITE_REQUEST:
1384             fds->fd = io->asok;
1385             fds->events = POLLOUT;
1386             return 1;
1387         case CSTATE_READ_VERSION:
1388         case CSTATE_READ_AMT:
1389         case CSTATE_READ_JSON:
1390             fds->fd = io->asok;
1391             fds->events = POLLIN;
1392             return 1;
1393         default:
1394             ERROR("cconn_prepare(name=%s) got to illegal state on line %d",
1395                     io->d->name, __LINE__);
1396             return -EDOM;
1397     }
1398 }
1399
1400 /** Returns the difference between two struct timevals in milliseconds.
1401  * On overflow, we return max/min int.
1402  */
1403 static int milli_diff(const struct timeval *t1, const struct timeval *t2)
1404 {
1405     int64_t ret;
1406     int sec_diff = t1->tv_sec - t2->tv_sec;
1407     int usec_diff = t1->tv_usec - t2->tv_usec;
1408     ret = usec_diff / 1000;
1409     ret += (sec_diff * 1000);
1410     return (ret > INT_MAX) ? INT_MAX : ((ret < INT_MIN) ? INT_MIN : (int)ret);
1411 }
1412
1413 /** This handles the actual network I/O to talk to the Ceph daemons.
1414  */
1415 static int cconn_main_loop(uint32_t request_type)
1416 {
1417     int i, ret, some_unreachable = 0;
1418     struct timeval end_tv;
1419     struct cconn io_array[g_num_daemons];
1420
1421     DEBUG("entering cconn_main_loop(request_type = %d)", request_type);
1422
1423     /* create cconn array */
1424     memset(io_array, 0, sizeof(io_array));
1425     for(i = 0; i < g_num_daemons; ++i)
1426     {
1427         io_array[i].d = g_daemons[i];
1428         io_array[i].request_type = request_type;
1429         io_array[i].state = CSTATE_UNCONNECTED;
1430     }
1431
1432     /** Calculate the time at which we should give up */
1433     gettimeofday(&end_tv, NULL);
1434     end_tv.tv_sec += CEPH_TIMEOUT_INTERVAL;
1435
1436     while (1)
1437     {
1438         int nfds, diff;
1439         struct timeval tv;
1440         struct cconn *polled_io_array[g_num_daemons];
1441         struct pollfd fds[g_num_daemons];
1442         memset(fds, 0, sizeof(fds));
1443         nfds = 0;
1444         for(i = 0; i < g_num_daemons; ++i)
1445         {
1446             struct cconn *io = io_array + i;
1447             ret = cconn_prepare(io, fds + nfds);
1448             if(ret < 0)
1449             {
1450                 WARNING("ERROR: cconn_prepare(name=%s,i=%d,st=%d)=%d",
1451                         io->d->name, i, io->state, ret);
1452                 cconn_close(io);
1453                 io->request_type = ASOK_REQ_NONE;
1454                 some_unreachable = 1;
1455             }
1456             else if(ret == 1)
1457             {
1458                 DEBUG("did cconn_prepare(name=%s,i=%d,st=%d)",
1459                         io->d->name, i, io->state);
1460                 polled_io_array[nfds++] = io_array + i;
1461             }
1462         }
1463         if(nfds == 0)
1464         {
1465             /* finished */
1466             ret = 0;
1467             DEBUG("cconn_main_loop: no more cconn to manage.");
1468             goto done;
1469         }
1470         gettimeofday(&tv, NULL);
1471         diff = milli_diff(&end_tv, &tv);
1472         if(diff <= 0)
1473         {
1474             /* Timed out */
1475             ret = -ETIMEDOUT;
1476             WARNING("ERROR: cconn_main_loop: timed out.\n");
1477             goto done;
1478         }
1479         RETRY_ON_EINTR(ret, poll(fds, nfds, diff));
1480         if(ret < 0)
1481         {
1482             ERROR("poll(2) error: %d", ret);
1483             goto done;
1484         }
1485         for(i = 0; i < nfds; ++i)
1486         {
1487             struct cconn *io = polled_io_array[i];
1488             int revents = fds[i].revents;
1489             if(revents == 0)
1490             {
1491                 /* do nothing */
1492             }
1493             else if(cconn_validate_revents(io, revents))
1494             {
1495                 WARNING("ERROR: cconn(name=%s,i=%d,st=%d): "
1496                 "revents validation error: "
1497                 "revents=0x%08x", io->d->name, i, io->state, revents);
1498                 cconn_close(io);
1499                 io->request_type = ASOK_REQ_NONE;
1500                 some_unreachable = 1;
1501             }
1502             else
1503             {
1504                 int ret = cconn_handle_event(io);
1505                 if(ret)
1506                 {
1507                     WARNING("ERROR: cconn_handle_event(name=%s,"
1508                     "i=%d,st=%d): error %d", io->d->name, i, io->state, ret);
1509                     cconn_close(io);
1510                     io->request_type = ASOK_REQ_NONE;
1511                     some_unreachable = 1;
1512                 }
1513             }
1514         }
1515     }
1516     done: for(i = 0; i < g_num_daemons; ++i)
1517     {
1518         cconn_close(io_array + i);
1519     }
1520     if(some_unreachable)
1521     {
1522         DEBUG("cconn_main_loop: some Ceph daemons were unreachable.");
1523     }
1524     else
1525     {
1526         DEBUG("cconn_main_loop: reached all Ceph daemons :)");
1527     }
1528     return ret;
1529 }
1530
1531 static int ceph_read(void)
1532 {
1533     return cconn_main_loop(ASOK_REQ_DATA);
1534 }
1535
1536 /******* lifecycle *******/
1537 static int ceph_init(void)
1538 {
1539     int i, ret, j;
1540     DEBUG("ceph_init");
1541     ceph_daemons_print();
1542
1543     ret = cconn_main_loop(ASOK_REQ_VERSION);
1544     if(ret)
1545     {
1546         return ret;
1547     }
1548     for(i = 0; i < g_num_daemons; ++i)
1549     {
1550         struct ceph_daemon *d = g_daemons[i];
1551         for(j = 0; j < d->dset_num; j++)
1552         {
1553             ret = plugin_register_data_set(d->dset + j);
1554             if(ret)
1555             {
1556                 ERROR("plugin_register_data_set(%s) failed!", d->name);
1557             }
1558             else
1559             {
1560                 DEBUG("plugin_register_data_set(%s): "
1561                         "(d->dset)[%d]->ds_num=%d",
1562                         d->name, j, d->dset[j].ds_num);
1563             }
1564         }
1565     }
1566     return 0;
1567 }
1568
1569 static int ceph_shutdown(void)
1570 {
1571     int i;
1572     for(i = 0; i < g_num_daemons; ++i)
1573     {
1574         ceph_daemon_free(g_daemons[i]);
1575     }
1576     sfree(g_daemons);
1577     g_daemons = NULL;
1578     g_num_daemons = 0;
1579     for(i = 0; i < last_idx; i++)
1580     {
1581         sfree(last_poll_data[i]);
1582     }
1583     sfree(last_poll_data);
1584     last_poll_data = NULL;
1585     last_idx = 0;
1586     DEBUG("finished ceph_shutdown");
1587     return 0;
1588 }
1589
1590 void module_register(void)
1591 {
1592     plugin_register_complex_config("ceph", ceph_config);
1593     plugin_register_init("ceph", ceph_init);
1594     plugin_register_read("ceph", ceph_read);
1595     plugin_register_shutdown("ceph", ceph_shutdown);
1596 }