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