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