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         DEBUG("enter node_handler_fetch_data");
 954         return 1;
 955     }
 956     dset_idx = get_matching_dset(vtmp->d, dset_name);
 957     if(dset_idx == -1)
 958     {
 959         return 1;
 960     }
 961     ds_idx = get_matching_value(&vtmp->d->dset[dset_idx], ds_name,
 962             vtmp->d->dset[dset_idx].ds_num);
 963     if(ds_idx == -1)
 964     {
 965         DEBUG("DSet:%s, DS:%s, DSet idx:%d, DS idx:%d",
 966             dset_name,ds_name,dset_idx,ds_idx);
 967         return RETRY_AVGCOUNT;
 968     }
 969     uv = &(vtmp->vh[dset_idx].values[ds_idx]);
 970
 971     if(vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_LATENCY)
 972     {
 973         if(vtmp->avgcount == -1)
 974         {
 975             sscanf(val, "%" PRIu64, &vtmp->avgcount);
 976         }
 977         else
 978         {
 979             double sum, result;
 980             sscanf(val, "%lf", &sum);
 981             DEBUG("avgcount:%ld",vtmp->avgcount);
 982             DEBUG("sum:%lf",sum);
 983
 984             if(vtmp->avgcount == 0)
 985             {
 986                 vtmp->avgcount = 1;
 987             }
 988
 989             /** User wants latency values as long run avg */
 990             if(long_run_latency_avg)
 991             {
 992                 result = (sum / vtmp->avgcount);
 993                 DEBUG("uv->gauge = sumd / avgcounti = :%lf", result);
 994             }
 995             else
 996             {
 997                 result = get_last_avg(dset_name, ds_name, sum, vtmp->avgcount);
 998                 if(result == -ENOMEM)
 999                 {
1000                     return -ENOMEM;
1001                 }
1002                 DEBUG("uv->gauge = (sumd_now - sumd_last) / "
1003                         "(avgcounti_now - avgcounti_last) = :%lf", result);
1004             }
1005
1006             uv->gauge = result;
1007             vtmp->avgcount = -1;
1008         }
1009     }
1010     else if(vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_DERIVE)
1011     {
1012         uint64_t derive_val;
1013         sscanf(val, "%" PRIu64, &derive_val);
1014         uv->derive = derive_val;
1015         DEBUG("uv->derive %" PRIu64 "",(uint64_t)uv->derive);
1016     }
1017     else
1018     {
1019         double other_val;
1020         sscanf(val, "%lf", &other_val);
1021         uv->gauge = other_val;
1022         DEBUG("uv->gauge %lf",uv->gauge);
1023     }
1024     return 0;
1025 }
1026
1027 static int cconn_connect(struct cconn *io)
1028 {
1029     struct sockaddr_un address;
1030     int flags, fd, err;
1031     if(io->state != CSTATE_UNCONNECTED)
1032     {
1033         ERROR("cconn_connect: io->state != CSTATE_UNCONNECTED");
1034         return -EDOM;
1035     }
1036     fd = socket(PF_UNIX, SOCK_STREAM, 0);
1037     if(fd < 0)
1038     {
1039         int err = -errno;
1040         ERROR("cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) failed: "
1041         "error %d", err);
1042         return err;
1043     }
1044     memset(&address, 0, sizeof(struct sockaddr_un));
1045     address.sun_family = AF_UNIX;
1046     snprintf(address.sun_path, sizeof(address.sun_path), "%s",
1047             io->d->asok_path);
1048     RETRY_ON_EINTR(err,
1049         connect(fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)));
1050     if(err < 0)
1051     {
1052         ERROR("cconn_connect: connect(%d) failed: error %d", fd, err);
1053         return err;
1054     }
1055
1056     flags = fcntl(fd, F_GETFL, 0);
1057     if(fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
1058     {
1059         err = -errno;
1060         ERROR("cconn_connect: fcntl(%d, O_NONBLOCK) error %d", fd, err);
1061         return err;
1062     }
1063     io->asok = fd;
1064     io->state = CSTATE_WRITE_REQUEST;
1065     io->amt = 0;
1066     io->json_len = 0;
1067     io->json = NULL;
1068     return 0;
1069 }
1070
1071 static void cconn_close(struct cconn *io)
1072 {
1073     io->state = CSTATE_UNCONNECTED;
1074     if(io->asok != -1)
1075     {
1076         int res;
1077         RETRY_ON_EINTR(res, close(io->asok));
1078     }
1079     io->asok = -1;
1080     io->amt = 0;
1081     io->json_len = 0;
1082     sfree(io->json);
1083     io->json = NULL;
1084 }
1085
1086 /* Process incoming JSON counter data */
1087 static int
1088 cconn_process_data(struct cconn *io, yajl_struct *yajl, yajl_handle hand)
1089 {
1090     int i, ret = 0;
1091     struct values_tmp *vtmp = calloc(1, sizeof(struct values_tmp)
1092                     + (sizeof(struct values_holder)) * io->d->dset_num);
1093     if(!vtmp)
1094     {
1095         return -ENOMEM;
1096     }
1097
1098     for(i = 0; i < io->d->dset_num; i++)
1099     {
1100         value_t *val = calloc(1, (sizeof(value_t) * io->d->dset[i].ds_num));
1101         vtmp->vh[i].values = val;
1102         vtmp->vh[i].values_len = io->d->dset[i].ds_num;
1103     }
1104     vtmp->d = io->d;
1105     vtmp->holder_num = io->d->dset_num;
1106     vtmp->avgcount = -1;
1107     yajl->handler_arg = vtmp;
1108     ret = traverse_json(io->json, io->json_len, hand);
1109     if(ret)
1110     {
1111         goto done;
1112     }
1113     for(i = 0; i < vtmp->holder_num; i++)
1114     {
1115         value_list_t vl = VALUE_LIST_INIT;
1116         sstrncpy(vl.host, hostname_g, sizeof(vl.host));
1117         sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
1118         strncpy(vl.plugin_instance, io->d->name, sizeof(vl.plugin_instance));
1119         sstrncpy(vl.type, io->d->dset[i].type, sizeof(vl.type));
1120         vl.values = vtmp->vh[i].values;
1121         vl.values_len = io->d->dset[i].ds_num;
1122         DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
1123                 io->d->name, vl.values_len, io->json);
1124         ret = plugin_dispatch_values(&vl);
1125         if(ret)
1126         {
1127             goto done;
1128         }
1129     }
1130
1131     done: for(i = 0; i < vtmp->holder_num; i++)
1132     {
1133         sfree(vtmp->vh[i].values);
1134     }
1135     sfree(vtmp);
1136     return ret;
1137 }
1138
1139 static int cconn_process_json(struct cconn *io)
1140 {
1141     if((io->request_type != ASOK_REQ_DATA) &&
1142             (io->request_type != ASOK_REQ_SCHEMA))
1143     {
1144         return -EDOM;
1145     }
1146
1147     int result = 1;
1148     yajl_handle hand;
1149     yajl_status status;
1150
1151     hand = yajl_alloc(&callbacks,
1152 #if HAVE_YAJL_V2
1153       /* alloc funcs = */ NULL,
1154 #else
1155       /* alloc funcs = */ NULL, NULL,
1156 #endif
1157       /* context = */ (void *)(&io->yajl));
1158
1159     if(!hand)
1160     {
1161         ERROR ("ceph plugin: yajl_alloc failed.");
1162         return ENOMEM;
1163     }
1164
1165     io->yajl.depth = 0;
1166
1167     switch(io->request_type)
1168     {
1169         case ASOK_REQ_DATA:
1170             io->yajl.handler = node_handler_fetch_data;
1171             result = cconn_process_data(io, &io->yajl, hand);
1172             break;
1173         case ASOK_REQ_SCHEMA:
1174             io->yajl.handler = node_handler_define_schema;
1175             io->yajl.handler_arg = io->d;
1176             result = traverse_json(io->json, io->json_len, hand);
1177             break;
1178     }
1179
1180     if(result)
1181     {
1182         goto done;
1183     }
1184
1185 #if HAVE_YAJL_V2
1186     status = yajl_complete_parse(hand);
1187 #else
1188     status = yajl_parse_complete(hand);
1189 #endif
1190
1191     if (status != yajl_status_ok)
1192     {
1193       unsigned char *errmsg = yajl_get_error (hand, /* verbose = */ 0,
1194           /* jsonText = */ NULL, /* jsonTextLen = */ 0);
1195       ERROR ("ceph plugin: yajl_parse_complete failed: %s",
1196           (char *) errmsg);
1197       yajl_free_error (hand, errmsg);
1198       yajl_free (hand);
1199       return 1;
1200     }
1201
1202     done:
1203     yajl_free (hand);
1204     return result;
1205 }
1206
1207 static int cconn_validate_revents(struct cconn *io, int revents)
1208 {
1209     if(revents & POLLERR)
1210     {
1211         ERROR("cconn_validate_revents(name=%s): got POLLERR", io->d->name);
1212         return -EIO;
1213     }
1214     switch (io->state)
1215     {
1216         case CSTATE_WRITE_REQUEST:
1217             return (revents & POLLOUT) ? 0 : -EINVAL;
1218         case CSTATE_READ_VERSION:
1219         case CSTATE_READ_AMT:
1220         case CSTATE_READ_JSON:
1221             return (revents & POLLIN) ? 0 : -EINVAL;
1222             return (revents & POLLIN) ? 0 : -EINVAL;
1223         default:
1224             ERROR("cconn_validate_revents(name=%s) got to illegal state on "
1225                     "line %d", io->d->name, __LINE__);
1226             return -EDOM;
1227     }
1228 }
1229
1230 /** Handle a network event for a connection */
1231 static int cconn_handle_event(struct cconn *io)
1232 {
1233     int ret;
1234     switch (io->state)
1235     {
1236         case CSTATE_UNCONNECTED:
1237             ERROR("cconn_handle_event(name=%s) got to illegal state on line "
1238                     "%d", io->d->name, __LINE__);
1239
1240             return -EDOM;
1241         case CSTATE_WRITE_REQUEST:
1242         {
1243             char cmd[32];
1244             snprintf(cmd, sizeof(cmd), "%s%d%s", "{ \"prefix\": \"",
1245                     io->request_type, "\" }\n");
1246             size_t cmd_len = strlen(cmd);
1247             RETRY_ON_EINTR(ret,
1248                   write(io->asok, ((char*)&cmd) + io->amt, cmd_len - io->amt));
1249             DEBUG("cconn_handle_event(name=%s,state=%d,amt=%d,ret=%d)",
1250                     io->d->name, io->state, io->amt, ret);
1251             if(ret < 0)
1252             {
1253                 return ret;
1254             }
1255             io->amt += ret;
1256             if(io->amt >= cmd_len)
1257             {
1258                 io->amt = 0;
1259                 switch (io->request_type)
1260                 {
1261                     case ASOK_REQ_VERSION:
1262                         io->state = CSTATE_READ_VERSION;
1263                         break;
1264                     default:
1265                         io->state = CSTATE_READ_AMT;
1266                         break;
1267                 }
1268             }
1269             return 0;
1270         }
1271         case CSTATE_READ_VERSION:
1272         {
1273             RETRY_ON_EINTR(ret,
1274                     read(io->asok, ((char*)(&io->d->version)) + io->amt,
1275                             sizeof(io->d->version) - io->amt));
1276             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1277                     io->d->name, io->state, ret);
1278             if(ret < 0)
1279             {
1280                 return ret;
1281             }
1282             io->amt += ret;
1283             if(io->amt >= sizeof(io->d->version))
1284             {
1285                 io->d->version = ntohl(io->d->version);
1286                 if(io->d->version != 1)
1287                 {
1288                     ERROR("cconn_handle_event(name=%s) not "
1289                     "expecting version %d!", io->d->name, io->d->version);
1290                     return -ENOTSUP;
1291                 }DEBUG("cconn_handle_event(name=%s): identified as "
1292                         "version %d", io->d->name, io->d->version);
1293                 io->amt = 0;
1294                 cconn_close(io);
1295                 io->request_type = ASOK_REQ_SCHEMA;
1296             }
1297             return 0;
1298         }
1299         case CSTATE_READ_AMT:
1300         {
1301             RETRY_ON_EINTR(ret,
1302                     read(io->asok, ((char*)(&io->json_len)) + io->amt,
1303                             sizeof(io->json_len) - io->amt));
1304             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1305                     io->d->name, io->state, ret);
1306             if(ret < 0)
1307             {
1308                 return ret;
1309             }
1310             io->amt += ret;
1311             if(io->amt >= sizeof(io->json_len))
1312             {
1313                 io->json_len = ntohl(io->json_len);
1314                 io->amt = 0;
1315                 io->state = CSTATE_READ_JSON;
1316                 io->json = calloc(1, io->json_len + 1);
1317                 if(!io->json)
1318                 {
1319                     ERROR("ERR CALLOCING IO->JSON");
1320                     return -ENOMEM;
1321                 }
1322             }
1323             return 0;
1324         }
1325         case CSTATE_READ_JSON:
1326         {
1327             RETRY_ON_EINTR(ret,
1328                    read(io->asok, io->json + io->amt, io->json_len - io->amt));
1329             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1330                     io->d->name, io->state, ret);
1331             if(ret < 0)
1332             {
1333                 return ret;
1334             }
1335             io->amt += ret;
1336             if(io->amt >= io->json_len)
1337             {
1338                 ret = cconn_process_json(io);
1339                 if(ret)
1340                 {
1341                     return ret;
1342                 }
1343                 cconn_close(io);
1344                 io->request_type = ASOK_REQ_NONE;
1345             }
1346             return 0;
1347         }
1348         default:
1349             ERROR("cconn_handle_event(name=%s) got to illegal state on "
1350             "line %d", io->d->name, __LINE__);
1351             return -EDOM;
1352     }
1353 }
1354
1355 static int cconn_prepare(struct cconn *io, struct pollfd* fds)
1356 {
1357     int ret;
1358     if(io->request_type == ASOK_REQ_NONE)
1359     {
1360         /* The request has already been serviced. */
1361         return 0;
1362     }
1363     else if((io->request_type == ASOK_REQ_DATA) && (io->d->dset_num == 0))
1364     {
1365         /* If there are no counters to report on, don't bother
1366          * connecting */
1367         return 0;
1368     }
1369
1370     switch (io->state)
1371     {
1372         case CSTATE_UNCONNECTED:
1373             ret = cconn_connect(io);
1374             if(ret > 0)
1375             {
1376                 return -ret;
1377             }
1378             else if(ret < 0)
1379             {
1380                 return ret;
1381             }
1382             fds->fd = io->asok;
1383             fds->events = POLLOUT;
1384             return 1;
1385         case CSTATE_WRITE_REQUEST:
1386             fds->fd = io->asok;
1387             fds->events = POLLOUT;
1388             return 1;
1389         case CSTATE_READ_VERSION:
1390         case CSTATE_READ_AMT:
1391         case CSTATE_READ_JSON:
1392             fds->fd = io->asok;
1393             fds->events = POLLIN;
1394             return 1;
1395         default:
1396             ERROR("cconn_prepare(name=%s) got to illegal state on line %d",
1397                     io->d->name, __LINE__);
1398             return -EDOM;
1399     }
1400 }
1401
1402 /** Returns the difference between two struct timevals in milliseconds.
1403  * On overflow, we return max/min int.
1404  */
1405 static int milli_diff(const struct timeval *t1, const struct timeval *t2)
1406 {
1407     int64_t ret;
1408     int sec_diff = t1->tv_sec - t2->tv_sec;
1409     int usec_diff = t1->tv_usec - t2->tv_usec;
1410     ret = usec_diff / 1000;
1411     ret += (sec_diff * 1000);
1412     return (ret > INT_MAX) ? INT_MAX : ((ret < INT_MIN) ? INT_MIN : (int)ret);
1413 }
1414
1415 /** This handles the actual network I/O to talk to the Ceph daemons.
1416  */
1417 static int cconn_main_loop(uint32_t request_type)
1418 {
1419     int i, ret, some_unreachable = 0;
1420     struct timeval end_tv;
1421     struct cconn io_array[g_num_daemons];
1422
1423     DEBUG("entering cconn_main_loop(request_type = %d)", request_type);
1424
1425     /* create cconn array */
1426     memset(io_array, 0, sizeof(io_array));
1427     for(i = 0; i < g_num_daemons; ++i)
1428     {
1429         io_array[i].d = g_daemons[i];
1430         io_array[i].request_type = request_type;
1431         io_array[i].state = CSTATE_UNCONNECTED;
1432     }
1433
1434     /** Calculate the time at which we should give up */
1435     gettimeofday(&end_tv, NULL);
1436     end_tv.tv_sec += CEPH_TIMEOUT_INTERVAL;
1437
1438     while (1)
1439     {
1440         int nfds, diff;
1441         struct timeval tv;
1442         struct cconn *polled_io_array[g_num_daemons];
1443         struct pollfd fds[g_num_daemons];
1444         memset(fds, 0, sizeof(fds));
1445         nfds = 0;
1446         for(i = 0; i < g_num_daemons; ++i)
1447         {
1448             struct cconn *io = io_array + i;
1449             ret = cconn_prepare(io, fds + nfds);
1450             if(ret < 0)
1451             {
1452                 WARNING("ERROR: cconn_prepare(name=%s,i=%d,st=%d)=%d",
1453                         io->d->name, i, io->state, ret);
1454                 cconn_close(io);
1455                 io->request_type = ASOK_REQ_NONE;
1456                 some_unreachable = 1;
1457             }
1458             else if(ret == 1)
1459             {
1460                 DEBUG("did cconn_prepare(name=%s,i=%d,st=%d)",
1461                         io->d->name, i, io->state);
1462                 polled_io_array[nfds++] = io_array + i;
1463             }
1464         }
1465         if(nfds == 0)
1466         {
1467             /* finished */
1468             ret = 0;
1469             DEBUG("cconn_main_loop: no more cconn to manage.");
1470             goto done;
1471         }
1472         gettimeofday(&tv, NULL);
1473         diff = milli_diff(&end_tv, &tv);
1474         if(diff <= 0)
1475         {
1476             /* Timed out */
1477             ret = -ETIMEDOUT;
1478             WARNING("ERROR: cconn_main_loop: timed out.\n");
1479             goto done;
1480         }
1481         RETRY_ON_EINTR(ret, poll(fds, nfds, diff));
1482         if(ret < 0)
1483         {
1484             ERROR("poll(2) error: %d", ret);
1485             goto done;
1486         }
1487         for(i = 0; i < nfds; ++i)
1488         {
1489             struct cconn *io = polled_io_array[i];
1490             int revents = fds[i].revents;
1491             if(revents == 0)
1492             {
1493                 /* do nothing */
1494             }
1495             else if(cconn_validate_revents(io, revents))
1496             {
1497                 WARNING("ERROR: cconn(name=%s,i=%d,st=%d): "
1498                 "revents validation error: "
1499                 "revents=0x%08x", io->d->name, i, io->state, revents);
1500                 cconn_close(io);
1501                 io->request_type = ASOK_REQ_NONE;
1502                 some_unreachable = 1;
1503             }
1504             else
1505             {
1506                 int ret = cconn_handle_event(io);
1507                 if(ret)
1508                 {
1509                     WARNING("ERROR: cconn_handle_event(name=%s,"
1510                     "i=%d,st=%d): error %d", io->d->name, i, io->state, ret);
1511                     cconn_close(io);
1512                     io->request_type = ASOK_REQ_NONE;
1513                     some_unreachable = 1;
1514                 }
1515             }
1516         }
1517     }
1518     done: for(i = 0; i < g_num_daemons; ++i)
1519     {
1520         cconn_close(io_array + i);
1521     }
1522     if(some_unreachable)
1523     {
1524         DEBUG("cconn_main_loop: some Ceph daemons were unreachable.");
1525     }
1526     else
1527     {
1528         DEBUG("cconn_main_loop: reached all Ceph daemons :)");
1529     }
1530     return ret;
1531 }
1532
1533 static int ceph_read(void)
1534 {
1535     return cconn_main_loop(ASOK_REQ_DATA);
1536 }
1537
1538 /******* lifecycle *******/
1539 static int ceph_init(void)
1540 {
1541     int i, ret, j;
1542     DEBUG("ceph_init");
1543     ceph_daemons_print();
1544
1545     ret = cconn_main_loop(ASOK_REQ_VERSION);
1546     if(ret)
1547     {
1548         return ret;
1549     }
1550     for(i = 0; i < g_num_daemons; ++i)
1551     {
1552         struct ceph_daemon *d = g_daemons[i];
1553         for(j = 0; j < d->dset_num; j++)
1554         {
1555             ret = plugin_register_data_set(d->dset + j);
1556             if(ret)
1557             {
1558                 ERROR("plugin_register_data_set(%s) failed!", d->name);
1559             }
1560             else
1561             {
1562                 DEBUG("plugin_register_data_set(%s): "
1563                         "(d->dset)[%d]->ds_num=%d",
1564                         d->name, j, d->dset[j].ds_num);
1565             }
1566         }
1567     }
1568     return 0;
1569 }
1570
1571 static int ceph_shutdown(void)
1572 {
1573     int i;
1574     for(i = 0; i < g_num_daemons; ++i)
1575     {
1576         ceph_daemon_free(g_daemons[i]);
1577     }
1578     sfree(g_daemons);
1579     g_daemons = NULL;
1580     g_num_daemons = 0;
1581     for(i = 0; i < last_idx; i++)
1582     {
1583         sfree(last_poll_data[i]);
1584     }
1585     sfree(last_poll_data);
1586     last_poll_data = NULL;
1587     last_idx = 0;
1588     DEBUG("finished ceph_shutdown");
1589     return 0;
1590 }
1591
1592 void module_register(void)
1593 {
1594     plugin_register_complex_config("ceph", ceph_config);
1595     plugin_register_init("ceph", ceph_init);
1596     plugin_register_read("ceph", ceph_read);
1597     plugin_register_shutdown("ceph", ceph_shutdown);
1598 }