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     size_t key_chars_remaining = (DATA_MAX_NAME_LEN-1);
 409     int reserved = 0;
 410     int offset = 0;
 411     memset(tmp, 0, sizeof(tmp));
 412     if(source == NULL || dest == NULL || source[0] == '\0' || dest[0] != '\0')
 413     {
 414         return;
 415     }
 416     size_t src_len = strlen(source);
 417     snprintf(len_str, sizeof(len_str), "%zu", src_len);
 418     unsigned char append_status = 0x0;
 419     append_status |= (source[src_len - 1] == '-') ? 0x1 : 0x0;
 420     append_status |= (source[src_len - 1] == '+') ? 0x2 : 0x0;
 421     while ((keys[keys_num] = strtok_r(tmp_ptr, ":_-+", &save_ptr)) != NULL)
 422     {
 423         tmp_ptr = NULL;
 424         /** capitalize 1st char **/
 425         keys[keys_num][0] = toupper(keys[keys_num][0]);
 426         keys_num++;
 427         if(keys_num >= 16)
 428         {
 429             break;
 430         }
 431     }
 432     /** concatenate each part of source string **/
 433     for(i = 0; i < keys_num; i++)
 434     {
 435         strncat(tmp, keys[i], key_chars_remaining);
 436         key_chars_remaining -= strlen(keys[i]);
 437     }
 438     /** to coordinate limitation of length of ds name from RRD
 439      *  we will truncate ds_name
 440      *  when the its length is more than
 441      *  MAX_RRD_DS_NAME_LEN
 442      */
 443     if(strlen(tmp) > MAX_RRD_DS_NAME_LEN - 1)
 444     {
 445         append_status |= 0x4;
 446         /** we should reserve space for
 447          * len_str
 448          */
 449         reserved += 2;
 450     }
 451     if(append_status & 0x1)
 452     {
 453         /** we should reserve space for
 454          * "Minus"
 455          */
 456         reserved += 5;
 457     }
 458     if(append_status & 0x2)
 459     {
 460         /** we should reserve space for
 461          * "Plus"
 462          */
 463         reserved += 4;
 464     }
 465     snprintf(dest, MAX_RRD_DS_NAME_LEN - reserved, "%s", tmp);
 466     offset = strlen(dest);
 467     switch (append_status)
 468     {
 469         case 0x1:
 470             memcpy(dest + offset, "Minus", 5);
 471             break;
 472         case 0x2:
 473             memcpy(dest + offset, "Plus", 5);
 474             break;
 475         case 0x4:
 476             memcpy(dest + offset, len_str, 2);
 477             break;
 478         case 0x5:
 479             memcpy(dest + offset, "Minus", 5);
 480             memcpy(dest + offset + 5, len_str, 2);
 481             break;
 482         case 0x6:
 483             memcpy(dest + offset, "Plus", 4);
 484             memcpy(dest + offset + 4, len_str, 2);
 485             break;
 486         default:
 487             break;
 488     }
 489 }
 490 static int parse_keys(const char *key_str, char *dset_name, char *ds_name)
 491 {
 492     char *ptr, *rptr;
 493     size_t dset_name_len = 0;
 494     size_t ds_name_len = 0;
 495     char tmp_ds_name[DATA_MAX_NAME_LEN];
 496     memset(tmp_ds_name, 0, sizeof(tmp_ds_name));
 497     if(dset_name == NULL || ds_name == NULL || key_str == NULL ||
 498             key_str[0] == '\0' || dset_name[0] != '\0' || ds_name[0] != '\0')
 499     {
 500         return -1;
 501     }
 502     if((ptr = strchr(key_str, '.')) == NULL
 503             || (rptr = strrchr(key_str, '.')) == NULL)
 504     {
 505         strncpy(dset_name, key_str, DATA_MAX_NAME_LEN - 1);
 506         strncpy(tmp_ds_name, key_str, DATA_MAX_NAME_LEN - 1);
 507         goto compact;
 508     }
 509     dset_name_len =
 510             (ptr - key_str) > (DATA_MAX_NAME_LEN - 1) ?
 511                     (DATA_MAX_NAME_LEN - 1) : (ptr - key_str);
 512     memcpy(dset_name, key_str, dset_name_len);
 513     ds_name_len =
 514            (rptr - ptr) > DATA_MAX_NAME_LEN ? DATA_MAX_NAME_LEN : (rptr - ptr);
 515     if(ds_name_len == 0)
 516     { /** only have two keys **/
 517         if(!strncmp(rptr + 1, "type", 4))
 518         {/** if last key is "type",ignore **/
 519             strncpy(tmp_ds_name, dset_name, DATA_MAX_NAME_LEN - 1);
 520         }
 521         else
 522         {/** if last key isn't "type", copy last key **/
 523             strncpy(tmp_ds_name, rptr + 1, DATA_MAX_NAME_LEN - 1);
 524         }
 525     }
 526     else if(!strncmp(rptr + 1, "type", 4))
 527     {/** more than two keys **/
 528         memcpy(tmp_ds_name, ptr + 1, ds_name_len - 1);
 529     }
 530     else
 531     {/** copy whole keys **/
 532         strncpy(tmp_ds_name, ptr + 1, DATA_MAX_NAME_LEN - 1);
 533     }
 534     compact: compact_ds_name(tmp_ds_name, ds_name);
 535     return 0;
 536 }
 537
 538 static int get_matching_dset(const struct ceph_daemon *d, const char *name)
 539 {
 540     int idx;
 541     for(idx = 0; idx < d->dset_num; ++idx)
 542     {
 543         if(strcmp(d->dset[idx].type, name) == 0)
 544         {
 545             return idx;
 546         }
 547     }
 548     return -1;
 549 }
 550
 551 static int get_matching_value(const struct data_set_s *dset, const char *name,
 552         int num_values)
 553 {
 554     int idx;
 555     for(idx = 0; idx < num_values; ++idx)
 556     {
 557         if(strcmp(dset->ds[idx].name, name) == 0)
 558         {
 559             return idx;
 560         }
 561     }
 562     return -1;
 563 }
 564
 565 static int ceph_daemon_add_ds_entry(struct ceph_daemon *d, const char *name,
 566         int pc_type)
 567 {
 568     struct data_source_s *ds;
 569     struct data_set_s *dset;
 570     struct data_set_s *dset_array;
 571     int **pc_types_array = NULL;
 572     int *pc_types;
 573     int *pc_types_new;
 574     int idx = 0;
 575     if(strlen(name) + 1 > DATA_MAX_NAME_LEN)
 576     {
 577         return -ENAMETOOLONG;
 578     }
 579     char dset_name[DATA_MAX_NAME_LEN];
 580     char ds_name[MAX_RRD_DS_NAME_LEN];
 581     memset(dset_name, 0, sizeof(dset_name));
 582     memset(ds_name, 0, sizeof(ds_name));
 583     if(parse_keys(name, dset_name, ds_name))
 584     {
 585         return 1;
 586     }
 587     idx = get_matching_dset(d, dset_name);
 588     if(idx == -1)
 589     {/* need to add a dset **/
 590         dset_array = realloc(d->dset,
 591                 sizeof(struct data_set_s) * (d->dset_num + 1));
 592         if(!dset_array)
 593         {
 594             return -ENOMEM;
 595         }
 596         pc_types_array = realloc(d->pc_types,
 597                 sizeof(int *) * (d->dset_num + 1));
 598         if(!pc_types_array)
 599         {
 600             return -ENOMEM;
 601         }
 602         dset = &dset_array[d->dset_num];
 603         /** this step is very important, otherwise,
 604          *  realloc for dset->ds will tricky because of
 605          *  a random addr in dset->ds
 606          */
 607         memset(dset, 0, sizeof(struct data_set_s));
 608         dset->ds_num = 0;
 609         snprintf(dset->type, DATA_MAX_NAME_LEN, "%s", dset_name);
 610         pc_types = pc_types_array[d->dset_num] = NULL;
 611         d->dset = dset_array;
 612     }
 613     else
 614     {
 615         dset = &d->dset[idx];
 616         pc_types = d->pc_types[idx];
 617     }
 618     struct data_source_s *ds_array = realloc(dset->ds,
 619             sizeof(struct data_source_s) * (dset->ds_num + 1));
 620     if(!ds_array)
 621     {
 622         return -ENOMEM;
 623     }
 624     pc_types_new = realloc(pc_types, sizeof(int) * (dset->ds_num + 1));
 625     if(!pc_types_new)
 626     {
 627         return -ENOMEM;
 628     }
 629     dset->ds = ds_array;
 630
 631     if(convert_special_metrics)
 632     {
 633         /**
 634          * Special case for filestore:JournalWrBytes. For some reason, Ceph
 635          * schema encodes this as a count/sum pair while all other "Bytes" data
 636          * (excluding used/capacity bytes for OSD space) uses a single "Derive"
 637          * type. To spare further confusion, keep this KPI as the same type of
 638          * other "Bytes". Instead of keeping an "average" or "rate", use the
 639          * "sum" in the pair and assign that to the derive value.
 640          */
 641         if((strcmp(dset_name,"filestore") == 0) &&
 642                                         strcmp(ds_name, "JournalWrBytes") == 0)
 643         {
 644             pc_type = 10;
 645         }
 646     }
 647
 648     if(idx == -1)
 649     {
 650         pc_types_array[d->dset_num] = pc_types_new;
 651         d->pc_types = pc_types_array;
 652         d->pc_types[d->dset_num][dset->ds_num] = pc_type;
 653         d->dset_num++;
 654     }
 655     else
 656     {
 657         d->pc_types[idx] = pc_types_new;
 658         d->pc_types[idx][dset->ds_num] = pc_type;
 659     }
 660     ds = &ds_array[dset->ds_num++];
 661     snprintf(ds->name, MAX_RRD_DS_NAME_LEN, "%s", ds_name);
 662     ds->type = (pc_type & PERFCOUNTER_DERIVE) ? DS_TYPE_DERIVE : DS_TYPE_GAUGE;
 663
 664     /**
 665      * Use min of 0 for DERIVE types so we don't get negative values on Ceph
 666      * service restart
 667      */
 668     ds->min = (ds->type == DS_TYPE_DERIVE) ? 0 : NAN;
 669     ds->max = NAN;
 670     return 0;
 671 }
 672
 673 /******* ceph_config *******/
 674 static int cc_handle_str(struct oconfig_item_s *item, char *dest, int dest_len)
 675 {
 676     const char *val;
 677     if(item->values_num != 1)
 678     {
 679         return -ENOTSUP;
 680     }
 681     if(item->values[0].type != OCONFIG_TYPE_STRING)
 682     {
 683         return -ENOTSUP;
 684     }
 685     val = item->values[0].value.string;
 686     if(snprintf(dest, dest_len, "%s", val) > (dest_len - 1))
 687     {
 688         ERROR("ceph plugin: configuration parameter '%s' is too long.\n",
 689                 item->key);
 690         return -ENAMETOOLONG;
 691     }
 692     return 0;
 693 }
 694
 695 static int cc_handle_bool(struct oconfig_item_s *item, int *dest)
 696 {
 697     if(item->values_num != 1)
 698     {
 699         return -ENOTSUP;
 700     }
 701
 702     if(item->values[0].type != OCONFIG_TYPE_BOOLEAN)
 703     {
 704         return -ENOTSUP;
 705     }
 706
 707     *dest = (item->values[0].value.boolean) ? 1 : 0;
 708     return 0;
 709 }
 710
 711 static int cc_add_daemon_config(oconfig_item_t *ci)
 712 {
 713     int ret, i;
 714     struct ceph_daemon *array, *nd, cd;
 715     memset(&cd, 0, sizeof(struct ceph_daemon));
 716
 717     if((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING))
 718     {
 719         WARNING("ceph plugin: `Daemon' blocks need exactly one string "
 720                 "argument.");
 721         return (-1);
 722     }
 723
 724     ret = cc_handle_str(ci, cd.name, DATA_MAX_NAME_LEN);
 725     if(ret)
 726     {
 727         return ret;
 728     }
 729
 730     for(i=0; i < ci->children_num; i++)
 731     {
 732         oconfig_item_t *child = ci->children + i;
 733
 734         if(strcasecmp("SocketPath", child->key) == 0)
 735         {
 736             ret = cc_handle_str(child, cd.asok_path, sizeof(cd.asok_path));
 737             if(ret)
 738             {
 739                 return ret;
 740             }
 741         }
 742         else
 743         {
 744             WARNING("ceph plugin: ignoring unknown option %s", child->key);
 745         }
 746     }
 747     if(cd.name[0] == '\0')
 748     {
 749         ERROR("ceph plugin: you must configure a daemon name.\n");
 750         return -EINVAL;
 751     }
 752     else if(cd.asok_path[0] == '\0')
 753     {
 754         ERROR("ceph plugin(name=%s): you must configure an administrative "
 755         "socket path.\n", cd.name);
 756         return -EINVAL;
 757     }
 758     else if(!((cd.asok_path[0] == '/') ||
 759             (cd.asok_path[0] == '.' && cd.asok_path[1] == '/')))
 760     {
 761         ERROR("ceph plugin(name=%s): administrative socket paths must begin "
 762                 "with '/' or './' Can't parse: '%s'\n", cd.name, cd.asok_path);
 763         return -EINVAL;
 764     }
 765     array = realloc(g_daemons,
 766                     sizeof(struct ceph_daemon *) * (g_num_daemons + 1));
 767     if(array == NULL)
 768     {
 769         /* The positive return value here indicates that this is a
 770          * runtime error, not a configuration error.  */
 771         return ENOMEM;
 772     }
 773     g_daemons = (struct ceph_daemon**) array;
 774     nd = malloc(sizeof(struct ceph_daemon));
 775     if(!nd)
 776     {
 777         return ENOMEM;
 778     }
 779     memcpy(nd, &cd, sizeof(struct ceph_daemon));
 780     g_daemons[g_num_daemons++] = nd;
 781     return 0;
 782 }
 783
 784 static int ceph_config(oconfig_item_t *ci)
 785 {
 786     int ret, i;
 787
 788     for(i = 0; i < ci->children_num; ++i)
 789     {
 790         oconfig_item_t *child = ci->children + i;
 791         if(strcasecmp("Daemon", child->key) == 0)
 792         {
 793             ret = cc_add_daemon_config(child);
 794             if(ret == ENOMEM)
 795             {
 796                 ERROR("ceph plugin: Couldn't allocate memory");
 797                 return ret;
 798             }
 799             else if(ret)
 800             {
 801                 //process other daemons and ignore this one
 802                 continue;
 803             }
 804         }
 805         else if(strcasecmp("LongRunAvgLatency", child->key) == 0)
 806         {
 807             ret = cc_handle_bool(child, &long_run_latency_avg);
 808             if(ret)
 809             {
 810                 return ret;
 811             }
 812         }
 813         else if(strcasecmp("ConvertSpecialMetricTypes", child->key) == 0)
 814         {
 815             ret = cc_handle_bool(child, &convert_special_metrics);
 816             if(ret)
 817             {
 818                 return ret;
 819             }
 820         }
 821         else
 822         {
 823             WARNING("ceph plugin: ignoring unknown option %s", child->key);
 824         }
 825     }
 826     return 0;
 827 }
 828
 829 static int
 830 traverse_json(const unsigned char *json, uint32_t json_len, yajl_handle hand)
 831 {
 832     yajl_status status = yajl_parse(hand, json, json_len);
 833     unsigned char *msg;
 834
 835     switch(status)
 836     {
 837         case yajl_status_error:
 838             msg = yajl_get_error(hand, /* verbose = */ 1,
 839                                        /* jsonText = */ (unsigned char *) json,
 840                                                       (unsigned int) json_len);
 841             ERROR ("ceph plugin: yajl_parse failed: %s", msg);
 842             yajl_free_error(hand, msg);
 843             return 1;
 844         case yajl_status_client_canceled:
 845             return 1;
 846         default:
 847             return 0;
 848     }
 849 }
 850
 851 static int
 852 node_handler_define_schema(void *arg, const char *val, const char *key)
 853 {
 854     struct ceph_daemon *d = (struct ceph_daemon *) arg;
 855     int pc_type;
 856     pc_type = atoi(val);
 857     DEBUG("\nceph_daemon_add_ds_entry(d=%s,key=%s,pc_type=%04x)",
 858             d->name, key, pc_type);
 859     return ceph_daemon_add_ds_entry(d, key, pc_type);
 860 }
 861
 862 static int add_last(const char *dset_n, const char *ds_n, double cur_sum,
 863         uint64_t cur_count)
 864 {
 865     last_poll_data[last_idx] = malloc(1 * sizeof(struct last_data));
 866     if(!last_poll_data[last_idx])
 867     {
 868         return -ENOMEM;
 869     }
 870     sstrncpy(last_poll_data[last_idx]->dset_name,dset_n,
 871             sizeof(last_poll_data[last_idx]->dset_name));
 872     sstrncpy(last_poll_data[last_idx]->ds_name,ds_n,
 873             sizeof(last_poll_data[last_idx]->ds_name));
 874     last_poll_data[last_idx]->last_sum = cur_sum;
 875     last_poll_data[last_idx]->last_count = cur_count;
 876     last_idx++;
 877     return 0;
 878 }
 879
 880 static int update_last(const char *dset_n, const char *ds_n, double cur_sum,
 881         uint64_t cur_count)
 882 {
 883     int i;
 884     for(i = 0; i < last_idx; i++)
 885     {
 886         if(strcmp(last_poll_data[i]->dset_name,dset_n) == 0 &&
 887                 (strcmp(last_poll_data[i]->ds_name,ds_n) == 0))
 888         {
 889             last_poll_data[i]->last_sum = cur_sum;
 890             last_poll_data[i]->last_count = cur_count;
 891             return 0;
 892         }
 893     }
 894
 895     if(!last_poll_data)
 896     {
 897         last_poll_data = malloc(1 * sizeof(struct last_data *));
 898         if(!last_poll_data)
 899         {
 900             return -ENOMEM;
 901         }
 902     }
 903     else
 904     {
 905         struct last_data **tmp_last = realloc(last_poll_data,
 906                 ((last_idx+1) * sizeof(struct last_data *)));
 907         if(!tmp_last)
 908         {
 909             return -ENOMEM;
 910         }
 911         last_poll_data = tmp_last;
 912     }
 913     return add_last(dset_n,ds_n,cur_sum,cur_count);
 914 }
 915
 916 static double get_last_avg(const char *dset_n, const char *ds_n,
 917         double cur_sum, uint64_t cur_count)
 918 {
 919     int i;
 920     double result = -1.1, sum_delt = 0.0;
 921     uint64_t count_delt = 0;
 922     for(i = 0; i < last_idx; i++)
 923     {
 924         if((strcmp(last_poll_data[i]->dset_name,dset_n) == 0) &&
 925                 (strcmp(last_poll_data[i]->ds_name,ds_n) == 0))
 926         {
 927             if(cur_count < last_poll_data[i]->last_count)
 928             {
 929                 break;
 930             }
 931             sum_delt = (cur_sum - last_poll_data[i]->last_sum);
 932             count_delt = (cur_count - last_poll_data[i]->last_count);
 933             result = (sum_delt / count_delt);
 934             break;
 935         }
 936     }
 937
 938     if(result == -1.1)
 939     {
 940         result = NAN;
 941     }
 942     if(update_last(dset_n,ds_n,cur_sum,cur_count) == -ENOMEM)
 943     {
 944         return -ENOMEM;
 945     }
 946     return result;
 947 }
 948
 949 static int node_handler_fetch_data(void *arg, const char *val, const char *key)
 950 {
 951     int dset_idx, ds_idx;
 952     value_t *uv;
 953     char dset_name[DATA_MAX_NAME_LEN];
 954     char ds_name[MAX_RRD_DS_NAME_LEN];
 955     struct values_tmp *vtmp = (struct values_tmp*) arg;
 956     memset(dset_name, 0, sizeof(dset_name));
 957     memset(ds_name, 0, sizeof(ds_name));
 958     if(parse_keys(key, dset_name, ds_name))
 959     {
 960         DEBUG("enter node_handler_fetch_data");
 961         return 1;
 962     }
 963     dset_idx = get_matching_dset(vtmp->d, dset_name);
 964     if(dset_idx == -1)
 965     {
 966         return 1;
 967     }
 968     ds_idx = get_matching_value(&vtmp->d->dset[dset_idx], ds_name,
 969             vtmp->d->dset[dset_idx].ds_num);
 970     if(ds_idx == -1)
 971     {
 972         DEBUG("DSet:%s, DS:%s, DSet idx:%d, DS idx:%d",
 973             dset_name,ds_name,dset_idx,ds_idx);
 974         return RETRY_AVGCOUNT;
 975     }
 976     uv = &(vtmp->vh[dset_idx].values[ds_idx]);
 977
 978     if(vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_LATENCY)
 979     {
 980         if(vtmp->avgcount == -1)
 981         {
 982             sscanf(val, "%" PRIu64, &vtmp->avgcount);
 983         }
 984         else
 985         {
 986             double sum, result;
 987             sscanf(val, "%lf", &sum);
 988             DEBUG("avgcount:%" PRIu64 "",vtmp->avgcount);
 989             DEBUG("sum:%lf",sum);
 990
 991             if(vtmp->avgcount == 0)
 992             {
 993                 vtmp->avgcount = 1;
 994             }
 995
 996             /** User wants latency values as long run avg */
 997             if(long_run_latency_avg)
 998             {
 999                 result = (sum / vtmp->avgcount);
1000                 DEBUG("uv->gauge = sumd / avgcounti = :%lf", result);
1001             }
1002             else
1003             {
1004                 result = get_last_avg(dset_name, ds_name, sum, vtmp->avgcount);
1005                 if(result == -ENOMEM)
1006                 {
1007                     return -ENOMEM;
1008                 }
1009                 DEBUG("uv->gauge = (sumd_now - sumd_last) / "
1010                         "(avgcounti_now - avgcounti_last) = :%lf", result);
1011             }
1012
1013             uv->gauge = result;
1014             vtmp->avgcount = -1;
1015         }
1016     }
1017     else if(vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_DERIVE)
1018     {
1019         uint64_t derive_val;
1020         sscanf(val, "%" PRIu64, &derive_val);
1021         uv->derive = derive_val;
1022         DEBUG("uv->derive %" PRIu64 "",(uint64_t)uv->derive);
1023     }
1024     else
1025     {
1026         double other_val;
1027         sscanf(val, "%lf", &other_val);
1028         uv->gauge = other_val;
1029         DEBUG("uv->gauge %lf",uv->gauge);
1030     }
1031     return 0;
1032 }
1033
1034 static int cconn_connect(struct cconn *io)
1035 {
1036     struct sockaddr_un address;
1037     int flags, fd, err;
1038     if(io->state != CSTATE_UNCONNECTED)
1039     {
1040         ERROR("ceph plugin: cconn_connect: io->state != CSTATE_UNCONNECTED");
1041         return -EDOM;
1042     }
1043     fd = socket(PF_UNIX, SOCK_STREAM, 0);
1044     if(fd < 0)
1045     {
1046         int err = -errno;
1047         ERROR("ceph plugin: cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) "
1048             "failed: error %d", err);
1049         return err;
1050     }
1051     memset(&address, 0, sizeof(struct sockaddr_un));
1052     address.sun_family = AF_UNIX;
1053     snprintf(address.sun_path, sizeof(address.sun_path), "%s",
1054             io->d->asok_path);
1055     RETRY_ON_EINTR(err,
1056         connect(fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)));
1057     if(err < 0)
1058     {
1059         ERROR("ceph plugin: cconn_connect: connect(%d) failed: error %d",
1060             fd, err);
1061         return err;
1062     }
1063
1064     flags = fcntl(fd, F_GETFL, 0);
1065     if(fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
1066     {
1067         err = -errno;
1068         ERROR("ceph plugin: cconn_connect: fcntl(%d, O_NONBLOCK) error %d",
1069             fd, err);
1070         return err;
1071     }
1072     io->asok = fd;
1073     io->state = CSTATE_WRITE_REQUEST;
1074     io->amt = 0;
1075     io->json_len = 0;
1076     io->json = NULL;
1077     return 0;
1078 }
1079
1080 static void cconn_close(struct cconn *io)
1081 {
1082     io->state = CSTATE_UNCONNECTED;
1083     if(io->asok != -1)
1084     {
1085         int res;
1086         RETRY_ON_EINTR(res, close(io->asok));
1087     }
1088     io->asok = -1;
1089     io->amt = 0;
1090     io->json_len = 0;
1091     sfree(io->json);
1092     io->json = NULL;
1093 }
1094
1095 /* Process incoming JSON counter data */
1096 static int
1097 cconn_process_data(struct cconn *io, yajl_struct *yajl, yajl_handle hand)
1098 {
1099     int i, ret = 0;
1100     struct values_tmp *vtmp = calloc(1, sizeof(struct values_tmp)
1101                     + (sizeof(struct values_holder)) * io->d->dset_num);
1102     if(!vtmp)
1103     {
1104         return -ENOMEM;
1105     }
1106
1107     for(i = 0; i < io->d->dset_num; i++)
1108     {
1109         value_t *val = calloc(1, (sizeof(value_t) * io->d->dset[i].ds_num));
1110         vtmp->vh[i].values = val;
1111         vtmp->vh[i].values_len = io->d->dset[i].ds_num;
1112     }
1113     vtmp->d = io->d;
1114     vtmp->holder_num = io->d->dset_num;
1115     vtmp->avgcount = -1;
1116     yajl->handler_arg = vtmp;
1117     ret = traverse_json(io->json, io->json_len, hand);
1118     if(ret)
1119     {
1120         goto done;
1121     }
1122     for(i = 0; i < vtmp->holder_num; i++)
1123     {
1124         value_list_t vl = VALUE_LIST_INIT;
1125         sstrncpy(vl.host, hostname_g, sizeof(vl.host));
1126         sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
1127         strncpy(vl.plugin_instance, io->d->name, sizeof(vl.plugin_instance));
1128         sstrncpy(vl.type, io->d->dset[i].type, sizeof(vl.type));
1129         vl.values = vtmp->vh[i].values;
1130         vl.values_len = io->d->dset[i].ds_num;
1131         DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
1132                 io->d->name, vl.values_len, io->json);
1133         ret = plugin_dispatch_values(&vl);
1134         if(ret)
1135         {
1136             goto done;
1137         }
1138     }
1139
1140     done: for(i = 0; i < vtmp->holder_num; i++)
1141     {
1142         sfree(vtmp->vh[i].values);
1143     }
1144     sfree(vtmp);
1145     return ret;
1146 }
1147
1148 static int cconn_process_json(struct cconn *io)
1149 {
1150     if((io->request_type != ASOK_REQ_DATA) &&
1151             (io->request_type != ASOK_REQ_SCHEMA))
1152     {
1153         return -EDOM;
1154     }
1155
1156     int result = 1;
1157     yajl_handle hand;
1158     yajl_status status;
1159
1160     hand = yajl_alloc(&callbacks,
1161 #if HAVE_YAJL_V2
1162       /* alloc funcs = */ NULL,
1163 #else
1164       /* alloc funcs = */ NULL, NULL,
1165 #endif
1166       /* context = */ (void *)(&io->yajl));
1167
1168     if(!hand)
1169     {
1170         ERROR ("ceph plugin: yajl_alloc failed.");
1171         return ENOMEM;
1172     }
1173
1174     io->yajl.depth = 0;
1175
1176     switch(io->request_type)
1177     {
1178         case ASOK_REQ_DATA:
1179             io->yajl.handler = node_handler_fetch_data;
1180             result = cconn_process_data(io, &io->yajl, hand);
1181             break;
1182         case ASOK_REQ_SCHEMA:
1183             io->yajl.handler = node_handler_define_schema;
1184             io->yajl.handler_arg = io->d;
1185             result = traverse_json(io->json, io->json_len, hand);
1186             break;
1187     }
1188
1189     if(result)
1190     {
1191         goto done;
1192     }
1193
1194 #if HAVE_YAJL_V2
1195     status = yajl_complete_parse(hand);
1196 #else
1197     status = yajl_parse_complete(hand);
1198 #endif
1199
1200     if (status != yajl_status_ok)
1201     {
1202       unsigned char *errmsg = yajl_get_error (hand, /* verbose = */ 0,
1203           /* jsonText = */ NULL, /* jsonTextLen = */ 0);
1204       ERROR ("ceph plugin: yajl_parse_complete failed: %s",
1205           (char *) errmsg);
1206       yajl_free_error (hand, errmsg);
1207       yajl_free (hand);
1208       return 1;
1209     }
1210
1211     done:
1212     yajl_free (hand);
1213     return result;
1214 }
1215
1216 static int cconn_validate_revents(struct cconn *io, int revents)
1217 {
1218     if(revents & POLLERR)
1219     {
1220         ERROR("ceph plugin: cconn_validate_revents(name=%s): got POLLERR",
1221             io->d->name);
1222         return -EIO;
1223     }
1224     switch (io->state)
1225     {
1226         case CSTATE_WRITE_REQUEST:
1227             return (revents & POLLOUT) ? 0 : -EINVAL;
1228         case CSTATE_READ_VERSION:
1229         case CSTATE_READ_AMT:
1230         case CSTATE_READ_JSON:
1231             return (revents & POLLIN) ? 0 : -EINVAL;
1232             return (revents & POLLIN) ? 0 : -EINVAL;
1233         default:
1234             ERROR("ceph plugin: cconn_validate_revents(name=%s) got to "
1235                 "illegal state on line %d", io->d->name, __LINE__);
1236             return -EDOM;
1237     }
1238 }
1239
1240 /** Handle a network event for a connection */
1241 static int cconn_handle_event(struct cconn *io)
1242 {
1243     int ret;
1244     switch (io->state)
1245     {
1246         case CSTATE_UNCONNECTED:
1247             ERROR("ceph plugin: cconn_handle_event(name=%s) got to illegal "
1248                 "state on line %d", io->d->name, __LINE__);
1249
1250             return -EDOM;
1251         case CSTATE_WRITE_REQUEST:
1252         {
1253             char cmd[32];
1254             snprintf(cmd, sizeof(cmd), "%s%d%s", "{ \"prefix\": \"",
1255                     io->request_type, "\" }\n");
1256             size_t cmd_len = strlen(cmd);
1257             RETRY_ON_EINTR(ret,
1258                   write(io->asok, ((char*)&cmd) + io->amt, cmd_len - io->amt));
1259             DEBUG("cconn_handle_event(name=%s,state=%d,amt=%d,ret=%d)",
1260                     io->d->name, io->state, io->amt, ret);
1261             if(ret < 0)
1262             {
1263                 return ret;
1264             }
1265             io->amt += ret;
1266             if(io->amt >= cmd_len)
1267             {
1268                 io->amt = 0;
1269                 switch (io->request_type)
1270                 {
1271                     case ASOK_REQ_VERSION:
1272                         io->state = CSTATE_READ_VERSION;
1273                         break;
1274                     default:
1275                         io->state = CSTATE_READ_AMT;
1276                         break;
1277                 }
1278             }
1279             return 0;
1280         }
1281         case CSTATE_READ_VERSION:
1282         {
1283             RETRY_ON_EINTR(ret,
1284                     read(io->asok, ((char*)(&io->d->version)) + io->amt,
1285                             sizeof(io->d->version) - io->amt));
1286             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1287                     io->d->name, io->state, ret);
1288             if(ret < 0)
1289             {
1290                 return ret;
1291             }
1292             io->amt += ret;
1293             if(io->amt >= sizeof(io->d->version))
1294             {
1295                 io->d->version = ntohl(io->d->version);
1296                 if(io->d->version != 1)
1297                 {
1298                     ERROR("ceph plugin: cconn_handle_event(name=%s) not "
1299                         "expecting version %d!", io->d->name, io->d->version);
1300                     return -ENOTSUP;
1301                 }
1302                 DEBUG("cconn_handle_event(name=%s): identified as "
1303                         "version %d", io->d->name, io->d->version);
1304                 io->amt = 0;
1305                 cconn_close(io);
1306                 io->request_type = ASOK_REQ_SCHEMA;
1307             }
1308             return 0;
1309         }
1310         case CSTATE_READ_AMT:
1311         {
1312             RETRY_ON_EINTR(ret,
1313                     read(io->asok, ((char*)(&io->json_len)) + io->amt,
1314                             sizeof(io->json_len) - io->amt));
1315             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1316                     io->d->name, io->state, ret);
1317             if(ret < 0)
1318             {
1319                 return ret;
1320             }
1321             io->amt += ret;
1322             if(io->amt >= sizeof(io->json_len))
1323             {
1324                 io->json_len = ntohl(io->json_len);
1325                 io->amt = 0;
1326                 io->state = CSTATE_READ_JSON;
1327                 io->json = calloc(1, io->json_len + 1);
1328                 if(!io->json)
1329                 {
1330                     ERROR("ceph plugin: error callocing io->json");
1331                     return -ENOMEM;
1332                 }
1333             }
1334             return 0;
1335         }
1336         case CSTATE_READ_JSON:
1337         {
1338             RETRY_ON_EINTR(ret,
1339                    read(io->asok, io->json + io->amt, io->json_len - io->amt));
1340             DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
1341                     io->d->name, io->state, ret);
1342             if(ret < 0)
1343             {
1344                 return ret;
1345             }
1346             io->amt += ret;
1347             if(io->amt >= io->json_len)
1348             {
1349                 ret = cconn_process_json(io);
1350                 if(ret)
1351                 {
1352                     return ret;
1353                 }
1354                 cconn_close(io);
1355                 io->request_type = ASOK_REQ_NONE;
1356             }
1357             return 0;
1358         }
1359         default:
1360             ERROR("ceph plugin: cconn_handle_event(name=%s) got to illegal "
1361                 "state on line %d", io->d->name, __LINE__);
1362             return -EDOM;
1363     }
1364 }
1365
1366 static int cconn_prepare(struct cconn *io, struct pollfd* fds)
1367 {
1368     int ret;
1369     if(io->request_type == ASOK_REQ_NONE)
1370     {
1371         /* The request has already been serviced. */
1372         return 0;
1373     }
1374     else if((io->request_type == ASOK_REQ_DATA) && (io->d->dset_num == 0))
1375     {
1376         /* If there are no counters to report on, don't bother
1377          * connecting */
1378         return 0;
1379     }
1380
1381     switch (io->state)
1382     {
1383         case CSTATE_UNCONNECTED:
1384             ret = cconn_connect(io);
1385             if(ret > 0)
1386             {
1387                 return -ret;
1388             }
1389             else if(ret < 0)
1390             {
1391                 return ret;
1392             }
1393             fds->fd = io->asok;
1394             fds->events = POLLOUT;
1395             return 1;
1396         case CSTATE_WRITE_REQUEST:
1397             fds->fd = io->asok;
1398             fds->events = POLLOUT;
1399             return 1;
1400         case CSTATE_READ_VERSION:
1401         case CSTATE_READ_AMT:
1402         case CSTATE_READ_JSON:
1403             fds->fd = io->asok;
1404             fds->events = POLLIN;
1405             return 1;
1406         default:
1407             ERROR("ceph plugin: cconn_prepare(name=%s) got to illegal state "
1408                 "on line %d", io->d->name, __LINE__);
1409             return -EDOM;
1410     }
1411 }
1412
1413 /** Returns the difference between two struct timevals in milliseconds.
1414  * On overflow, we return max/min int.
1415  */
1416 static int milli_diff(const struct timeval *t1, const struct timeval *t2)
1417 {
1418     int64_t ret;
1419     int sec_diff = t1->tv_sec - t2->tv_sec;
1420     int usec_diff = t1->tv_usec - t2->tv_usec;
1421     ret = usec_diff / 1000;
1422     ret += (sec_diff * 1000);
1423     return (ret > INT_MAX) ? INT_MAX : ((ret < INT_MIN) ? INT_MIN : (int)ret);
1424 }
1425
1426 /** This handles the actual network I/O to talk to the Ceph daemons.
1427  */
1428 static int cconn_main_loop(uint32_t request_type)
1429 {
1430     int i, ret, some_unreachable = 0;
1431     struct timeval end_tv;
1432     struct cconn io_array[g_num_daemons];
1433
1434     DEBUG("entering cconn_main_loop(request_type = %d)", request_type);
1435
1436     /* create cconn array */
1437     memset(io_array, 0, sizeof(io_array));
1438     for(i = 0; i < g_num_daemons; ++i)
1439     {
1440         io_array[i].d = g_daemons[i];
1441         io_array[i].request_type = request_type;
1442         io_array[i].state = CSTATE_UNCONNECTED;
1443     }
1444
1445     /** Calculate the time at which we should give up */
1446     gettimeofday(&end_tv, NULL);
1447     end_tv.tv_sec += CEPH_TIMEOUT_INTERVAL;
1448
1449     while (1)
1450     {
1451         int nfds, diff;
1452         struct timeval tv;
1453         struct cconn *polled_io_array[g_num_daemons];
1454         struct pollfd fds[g_num_daemons];
1455         memset(fds, 0, sizeof(fds));
1456         nfds = 0;
1457         for(i = 0; i < g_num_daemons; ++i)
1458         {
1459             struct cconn *io = io_array + i;
1460             ret = cconn_prepare(io, fds + nfds);
1461             if(ret < 0)
1462             {
1463                 WARNING("ceph plugin: cconn_prepare(name=%s,i=%d,st=%d)=%d",
1464                         io->d->name, i, io->state, ret);
1465                 cconn_close(io);
1466                 io->request_type = ASOK_REQ_NONE;
1467                 some_unreachable = 1;
1468             }
1469             else if(ret == 1)
1470             {
1471                 DEBUG("did cconn_prepare(name=%s,i=%d,st=%d)",
1472                         io->d->name, i, io->state);
1473                 polled_io_array[nfds++] = io_array + i;
1474             }
1475         }
1476         if(nfds == 0)
1477         {
1478             /* finished */
1479             ret = 0;
1480             DEBUG("cconn_main_loop: no more cconn to manage.");
1481             goto done;
1482         }
1483         gettimeofday(&tv, NULL);
1484         diff = milli_diff(&end_tv, &tv);
1485         if(diff <= 0)
1486         {
1487             /* Timed out */
1488             ret = -ETIMEDOUT;
1489             WARNING("ceph plugin: cconn_main_loop: timed out.");
1490             goto done;
1491         }
1492         RETRY_ON_EINTR(ret, poll(fds, nfds, diff));
1493         if(ret < 0)
1494         {
1495             ERROR("ceph plugin: poll(2) error: %d", ret);
1496             goto done;
1497         }
1498         for(i = 0; i < nfds; ++i)
1499         {
1500             struct cconn *io = polled_io_array[i];
1501             int revents = fds[i].revents;
1502             if(revents == 0)
1503             {
1504                 /* do nothing */
1505             }
1506             else if(cconn_validate_revents(io, revents))
1507             {
1508                 WARNING("ceph plugin: cconn(name=%s,i=%d,st=%d): "
1509                 "revents validation error: "
1510                 "revents=0x%08x", io->d->name, i, io->state, revents);
1511                 cconn_close(io);
1512                 io->request_type = ASOK_REQ_NONE;
1513                 some_unreachable = 1;
1514             }
1515             else
1516             {
1517                 int ret = cconn_handle_event(io);
1518                 if(ret)
1519                 {
1520                     WARNING("ceph plugin: cconn_handle_event(name=%s,"
1521                     "i=%d,st=%d): error %d", io->d->name, i, io->state, ret);
1522                     cconn_close(io);
1523                     io->request_type = ASOK_REQ_NONE;
1524                     some_unreachable = 1;
1525                 }
1526             }
1527         }
1528     }
1529     done: for(i = 0; i < g_num_daemons; ++i)
1530     {
1531         cconn_close(io_array + i);
1532     }
1533     if(some_unreachable)
1534     {
1535         DEBUG("cconn_main_loop: some Ceph daemons were unreachable.");
1536     }
1537     else
1538     {
1539         DEBUG("cconn_main_loop: reached all Ceph daemons :)");
1540     }
1541     return ret;
1542 }
1543
1544 static int ceph_read(void)
1545 {
1546     return cconn_main_loop(ASOK_REQ_DATA);
1547 }
1548
1549 /******* lifecycle *******/
1550 static int ceph_init(void)
1551 {
1552     int i, ret, j;
1553     DEBUG("ceph_init");
1554     ceph_daemons_print();
1555
1556     ret = cconn_main_loop(ASOK_REQ_VERSION);
1557     if(ret)
1558     {
1559         return ret;
1560     }
1561     for(i = 0; i < g_num_daemons; ++i)
1562     {
1563         struct ceph_daemon *d = g_daemons[i];
1564         for(j = 0; j < d->dset_num; j++)
1565         {
1566             ret = plugin_register_data_set(d->dset + j);
1567             if(ret)
1568             {
1569                 ERROR("ceph plugin: plugin_register_data_set(%s) failed!",
1570                     d->name);
1571             }
1572             else
1573             {
1574                 DEBUG("plugin_register_data_set(%s): "
1575                         "(d->dset)[%d]->ds_num=%d",
1576                         d->name, j, d->dset[j].ds_num);
1577             }
1578         }
1579     }
1580     return 0;
1581 }
1582
1583 static int ceph_shutdown(void)
1584 {
1585     int i;
1586     for(i = 0; i < g_num_daemons; ++i)
1587     {
1588         ceph_daemon_free(g_daemons[i]);
1589     }
1590     sfree(g_daemons);
1591     g_daemons = NULL;
1592     g_num_daemons = 0;
1593     for(i = 0; i < last_idx; i++)
1594     {
1595         sfree(last_poll_data[i]);
1596     }
1597     sfree(last_poll_data);
1598     last_poll_data = NULL;
1599     last_idx = 0;
1600     DEBUG("finished ceph_shutdown");
1601     return 0;
1602 }
1603
1604 void module_register(void)
1605 {
1606     plugin_register_complex_config("ceph", ceph_config);
1607     plugin_register_init("ceph", ceph_init);
1608     plugin_register_read("ceph", ceph_read);
1609     plugin_register_shutdown("ceph", ceph_shutdown);
1610 }