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 <json/json.h>
  34 #include <json/json_object_private.h> /* need for struct json_object_iter */
  35 #include <limits.h>
  36 #include <poll.h>
  37 #include <stdint.h>
  38 #include <stdio.h>
  39 #include <stdlib.h>
  40 #include <string.h>
  41 #include <strings.h>
  42 #include <sys/socket.h>
  43 #include <sys/time.h>
  44 #include <sys/types.h>
  45 #include <sys/un.h>
  46 #include <unistd.h>
  47 #define MAX_RRD_DS_NAME_LEN 20
  48
  49 #define RETRY_ON_EINTR(ret, expr) \
  50         while(1) { \
  51                 ret = expr; \
  52                 if (ret >= 0) \
  53                         break; \
  54                 ret = -errno; \
  55                 if (ret != -EINTR) \
  56                         break; \
  57         }
  58
  59 /** Timeout interval in seconds */
  60 #define CEPH_TIMEOUT_INTERVAL 1
  61
  62 /** Maximum path length for a UNIX domain socket on this system */
  63 #define UNIX_DOMAIN_SOCK_PATH_MAX (sizeof(((struct sockaddr_un*)0)->sun_path))
  64
  65 /******* ceph_daemon *******/
  66 struct ceph_daemon
  67 {
  68         /** Version of the admin_socket interface */
  69         uint32_t version;
  70         /** daemon name **/
  71         char name[DATA_MAX_NAME_LEN];
  72
  73         int dset_num;
  74
  75         /** Path to the socket that we use to talk to the ceph daemon */
  76         char asok_path[UNIX_DOMAIN_SOCK_PATH_MAX];
  77
  78         /** The set of  key/value pairs that this daemon reports
  79          * dset.type            The daemon name
  80          * dset.ds_num          Number of data sources (key/value pairs)
  81          * dset.ds              Dynamically allocated array of key/value pairs
  82          */
  83         //struct data_set_s dset;
  84         /** Dynamically allocated array **/
  85         struct data_set_s *dset;
  86         int **pc_types;
  87 };
  88
  89 enum perfcounter_type_d
  90 {
  91         PERFCOUNTER_LONGRUNAVG = 0x4, PERFCOUNTER_COUNTER = 0x8,
  92 };
  93
  94 /** Array of daemons to monitor */
  95 static struct ceph_daemon **g_daemons = NULL;
  96
  97 /** Number of elements in g_daemons */
  98 static int g_num_daemons = 0;
  99
 100 static void ceph_daemon_print(const struct ceph_daemon *d)
 101 {
 102         DEBUG("name=%s, asok_path=%s", d->name, d->asok_path);
 103 }
 104
 105 static void ceph_daemons_print(void)
 106 {
 107         int i;
 108         for (i = 0; i < g_num_daemons; ++i)
 109         {
 110                 ceph_daemon_print(g_daemons[i]);
 111         }
 112 }
 113
 114 /*static void ceph_daemon_free(struct ceph_daemon *d)
 115  {
 116  plugin_unregister_data_set(d->dset.type);
 117  sfree(d->dset.ds);
 118  sfree(d);
 119  }*/
 120 static void ceph_daemon_free(struct ceph_daemon *d)
 121 {
 122         int i = 0;
 123         for (; i < d->dset_num; i++)
 124         {
 125                 plugin_unregister_data_set((d->dset + i)->type);
 126                 sfree(d->dset->ds);
 127                 sfree(d->pc_types[i]);
 128         }
 129         sfree(d->dset);
 130         sfree(d->pc_types);
 131         sfree(d);
 132 }
 133
 134 static void compact_ds_name(char *source, char *dest)
 135 {
 136         int keys_num = 0, i;
 137         char *save_ptr = NULL, *tmp_ptr = source;
 138         char *keys[16];
 139         char len_str[3];
 140         char tmp[DATA_MAX_NAME_LEN];
 141         int reserved = 0;
 142         int offset = 0;
 143         memset(tmp, 0, sizeof(tmp));
 144         if (source == NULL || dest == NULL || source[0] == '\0' || dest[0] != '\0')
 145         {
 146                 return;
 147         }
 148         size_t src_len = strlen(source);
 149         snprintf(len_str, sizeof(len_str), "%zu", src_len);
 150         unsigned char append_status = 0x0;
 151         append_status |= (source[src_len - 1] == '-') ? 0x1 : 0x0;
 152         append_status |= (source[src_len - 1] == '+') ? 0x2 : 0x0;
 153         while ((keys[keys_num] = strtok_r(tmp_ptr, ":_-+", &save_ptr)) != NULL)
 154         {
 155                 tmp_ptr = NULL;
 156                 /** capitalize 1st char **/
 157                 keys[keys_num][0] = toupper(keys[keys_num][0]);
 158                 keys_num++;
 159                 if (keys_num >= 16)
 160                         break;
 161         }
 162         /** concatenate each part of source string **/
 163         for (i = 0; i < keys_num; i++)
 164         {
 165                 strcat(tmp, keys[i]);
 166         }
 167         tmp[DATA_MAX_NAME_LEN - 1] = '\0';
 168         /** to coordinate limitation of length of ds name from RRD
 169          *  we will truncate ds_name
 170          *  when the its length is more than
 171          *  MAX_RRD_DS_NAME_LEN
 172          */
 173         if (strlen(tmp) > MAX_RRD_DS_NAME_LEN - 1)
 174         {
 175                 append_status |= 0x4;
 176                 /** we should reserve space for
 177                  * len_str
 178                  */
 179                 reserved += 2;
 180         }
 181         if (append_status & 0x1)
 182         {
 183                 /** we should reserve space for
 184                  * "Minus"
 185                  */
 186                 reserved += 5;
 187         }
 188         if (append_status & 0x2)
 189         {
 190                 /** we should reserve space for
 191                  * "Plus"
 192                  */
 193                 reserved += 4;
 194         }
 195         snprintf(dest, MAX_RRD_DS_NAME_LEN - reserved, "%s", tmp);
 196         offset = strlen(dest);
 197         switch (append_status)
 198         {
 199         case 0x1:
 200                 memcpy(dest + offset, "Minus", 5);
 201                 break;
 202         case 0x2:
 203                 memcpy(dest + offset, "Plus", 5);
 204                 break;
 205         case 0x4:
 206                 memcpy(dest + offset, len_str, 2);
 207                 break;
 208         case 0x5:
 209                 memcpy(dest + offset, "Minus", 5);
 210                 memcpy(dest + offset + 5, len_str, 2);
 211                 break;
 212         case 0x6:
 213                 memcpy(dest + offset, "Plus", 4);
 214                 memcpy(dest + offset + 4, len_str, 2);
 215                 break;
 216         default:
 217                 break;
 218         }
 219 }
 220 static int parse_keys(const char *key_str, char *dset_name, char *ds_name)
 221 {
 222         char *ptr, *rptr;
 223         size_t dset_name_len = 0;
 224         size_t ds_name_len = 0;
 225         char tmp_ds_name[DATA_MAX_NAME_LEN];
 226         memset(tmp_ds_name, 0, sizeof(tmp_ds_name));
 227         if (dset_name == NULL || ds_name == NULL || key_str == NULL
 228                         || key_str[0] == '\0' || dset_name[0] != '\0' || ds_name[0] != '\0')
 229         {
 230                 return -1;
 231         }
 232         if ((ptr = strchr(key_str, '.')) == NULL
 233                         || (rptr = strrchr(key_str, '.')) == NULL)
 234         {
 235                 strncpy(dset_name, key_str, DATA_MAX_NAME_LEN - 1);
 236                 strncpy(tmp_ds_name, key_str, DATA_MAX_NAME_LEN - 1);
 237                 goto compact;
 238         }
 239         dset_name_len =
 240                         (ptr - key_str) > (DATA_MAX_NAME_LEN - 1) ?
 241                                         (DATA_MAX_NAME_LEN - 1) : (ptr - key_str);
 242         memcpy(dset_name, key_str, dset_name_len);
 243         ds_name_len =
 244                         (rptr - ptr) > DATA_MAX_NAME_LEN ? DATA_MAX_NAME_LEN : (rptr - ptr);
 245         if (ds_name_len == 0)
 246         { /** only have two keys **/
 247                 if (!strncmp(rptr + 1, "type", 4))
 248                 {/** if last key is "type",ignore **/
 249                         strncpy(tmp_ds_name, dset_name, DATA_MAX_NAME_LEN - 1);
 250                 }
 251                 else
 252                 {/** if last key isn't "type", copy last key **/
 253                         strncpy(tmp_ds_name, rptr + 1, DATA_MAX_NAME_LEN - 1);
 254                 }
 255         }
 256         else if (!strncmp(rptr + 1, "type", 4))
 257         {/** more than two keys **/
 258                 memcpy(tmp_ds_name, ptr + 1, ds_name_len - 1);
 259         }
 260         else
 261         {/** copy whole keys **/
 262                 strncpy(tmp_ds_name, ptr + 1, DATA_MAX_NAME_LEN - 1);
 263         }
 264         compact: compact_ds_name(tmp_ds_name, ds_name);
 265         return 0;
 266 }
 267
 268 int get_matching_dset(const struct ceph_daemon *d, const char *name)
 269 {
 270         int idx;
 271         for (idx = 0; idx < d->dset_num; ++idx)
 272         {
 273                 if (strcmp(d->dset[idx].type, name) == 0)
 274                 {
 275                         return idx;
 276                 }
 277         }
 278         return -1;
 279 }
 280
 281 int get_matching_value(const struct data_set_s *dset, const char *name,
 282                 int num_values)
 283 {
 284         int idx;
 285         for (idx = 0; idx < num_values; ++idx)
 286         {
 287                 if (strcmp(dset->ds[idx].name, name) == 0)
 288                 {
 289                         return idx;
 290                 }
 291         }
 292         return -1;
 293 }
 294
 295 static int ceph_daemon_add_ds_entry(struct ceph_daemon *d, const char *name,
 296                 int pc_type)
 297 {
 298         struct data_source_s *ds;
 299         struct data_set_s *dset;
 300         struct data_set_s *dset_array;
 301         int **pc_types_array = NULL;
 302         int *pc_types;
 303         int *pc_types_new;
 304         int idx = 0;
 305         if (strlen(name) + 1 > DATA_MAX_NAME_LEN)
 306                 return -ENAMETOOLONG;
 307         char dset_name[DATA_MAX_NAME_LEN];
 308         char ds_name[MAX_RRD_DS_NAME_LEN];
 309         memset(dset_name, 0, sizeof(dset_name));
 310         memset(ds_name, 0, sizeof(ds_name));
 311         if (parse_keys(name, dset_name, ds_name))
 312                 return 1;
 313         idx = get_matching_dset(d, dset_name);
 314         if (idx == -1)
 315         {/* need to add a dset **/
 316                 dset_array = realloc(d->dset,
 317                                 sizeof(struct data_set_s) * (d->dset_num + 1));
 318                 if (!dset_array)
 319                         return -ENOMEM;
 320                 pc_types_array = realloc(d->pc_types,
 321                                 sizeof(int *) * (d->dset_num + 1));
 322                 if (!pc_types_array)
 323                         return -ENOMEM;
 324                 dset = &dset_array[d->dset_num];
 325                 /** this step is very important, otherwise,
 326                  *  realloc for dset->ds will tricky because of
 327                  *  a random addr in dset->ds
 328                  */
 329                 memset(dset, 0, sizeof(struct data_set_s));
 330                 dset->ds_num = 0;
 331                 snprintf(dset->type, DATA_MAX_NAME_LEN, "%s", dset_name);
 332                 pc_types = pc_types_array[d->dset_num] = NULL;
 333                 d->dset = dset_array;
 334         }
 335         else
 336         {
 337                 dset = &d->dset[idx];
 338                 pc_types = d->pc_types[idx];
 339         }
 340         struct data_source_s *ds_array = realloc(dset->ds,
 341                         sizeof(struct data_source_s) * (dset->ds_num + 1));
 342         if (!ds_array)
 343         {
 344                 return -ENOMEM;
 345         }
 346         pc_types_new = realloc(pc_types, sizeof(int) * (dset->ds_num + 1));
 347         if (!pc_types_new)
 348         {
 349                 return -ENOMEM;
 350         }
 351         dset->ds = ds_array;
 352         if (idx == -1)
 353         {
 354                 pc_types_array[d->dset_num] = pc_types_new;
 355                 d->pc_types = pc_types_array;
 356                 d->pc_types[d->dset_num][dset->ds_num] = pc_type;
 357                 d->dset_num++;
 358         }
 359         else
 360         {
 361                 d->pc_types[idx] = pc_types_new;
 362                 d->pc_types[idx][dset->ds_num] = pc_type;
 363         }
 364         ds = &ds_array[dset->ds_num++];
 365         snprintf(ds->name, MAX_RRD_DS_NAME_LEN, "%s", ds_name);
 366         ds->type =
 367                         (pc_type & PERFCOUNTER_COUNTER) ? DS_TYPE_COUNTER : DS_TYPE_GAUGE;
 368         ds->min = NAN;
 369         ds->max = NAN;
 370         return 0;
 371 }
 372
 373 /******* ceph_config *******/
 374 static int cc_handle_str(struct oconfig_item_s *item, char *dest, int dest_len)
 375 {
 376         const char *val;
 377         if (item->values_num != 1)
 378         {
 379                 return -ENOTSUP;
 380         }
 381         if (item->values[0].type != OCONFIG_TYPE_STRING)
 382         {
 383                 return -ENOTSUP;
 384         }
 385         val = item->values[0].value.string;
 386         if (snprintf(dest, dest_len, "%s", val) > (dest_len - 1))
 387         {
 388                 ERROR("ceph plugin: configuration parameter '%s' is too long.\n",
 389                                 item->key);
 390                 return -ENAMETOOLONG;
 391         }
 392         return 0;
 393 }
 394
 395 static int ceph_config(oconfig_item_t *ci)
 396 {
 397         int ret, i;
 398         struct ceph_daemon *array, *nd, cd;
 399         memset(&cd, 0, sizeof(struct ceph_daemon));
 400
 401         for (i = 0; i < ci->children_num; ++i)
 402         {
 403                 oconfig_item_t *child = ci->children + i;
 404                 if (strcasecmp("Name", child->key) == 0)
 405                 {
 406                         ret = cc_handle_str(child, cd.name, DATA_MAX_NAME_LEN);
 407                         if (ret)
 408                                 return ret;
 409                 }
 410                 else if (strcasecmp("SocketPath", child->key) == 0)
 411                 {
 412                         ret = cc_handle_str(child, cd.asok_path, sizeof(cd.asok_path));
 413                         if (ret)
 414                                 return ret;
 415                 }
 416                 else
 417                 {
 418                         WARNING("ceph plugin: ignoring unknown option %s", child->key);
 419                 }
 420         }
 421         if (cd.name[0] == '\0')
 422         {
 423                 ERROR("ceph plugin: you must configure a daemon name.\n");
 424                 return -EINVAL;
 425         }
 426         else if (cd.asok_path[0] == '\0')
 427         {
 428                 ERROR("ceph plugin(name=%s): you must configure an administrative "
 429                 "socket path.\n", cd.name);
 430                 return -EINVAL;
 431         }
 432         else if (!((cd.asok_path[0] == '/')
 433                         || (cd.asok_path[0] == '.' && cd.asok_path[1] == '/')))
 434         {
 435                 ERROR("ceph plugin(name=%s): administrative socket paths must begin with "
 436                                 "'/' or './' Can't parse: '%s'\n", cd.name, cd.asok_path);
 437                 return -EINVAL;
 438         }
 439         array = realloc(g_daemons,
 440                         sizeof(struct ceph_daemon *) * (g_num_daemons + 1));
 441         if (array == NULL)
 442         {
 443                 /* The positive return value here indicates that this is a
 444                  * runtime error, not a configuration error.  */
 445                 return ENOMEM;
 446         }
 447         g_daemons = (struct ceph_daemon**) array;
 448         nd = malloc(sizeof(struct ceph_daemon));
 449         if (!nd)
 450                 return ENOMEM;
 451         memcpy(nd, &cd, sizeof(struct ceph_daemon));
 452         g_daemons[g_num_daemons++] = nd;
 453         return 0;
 454 }
 455
 456 /******* JSON parsing *******/
 457 typedef int (*node_handler_t)(void*, json_object*, const char*);
 458
 459 /** Perform a depth-first traversal of the JSON parse tree,
 460  * calling node_handler at each node.*/
 461 static int traverse_json_impl(json_object *jo, char *key, int max_key,
 462                 node_handler_t handler, void *handler_arg)
 463 {
 464         struct json_object_iter iter;
 465         int ret, plen, klen;
 466
 467         if (json_object_get_type(jo) != json_type_object)
 468                 return 0;
 469         plen = strlen(key);
 470         json_object_object_foreachC(jo, iter)
 471         {
 472                 klen = strlen(iter.key);
 473                 if (plen + klen + 2 > max_key)
 474                         return -ENAMETOOLONG;
 475                 if (plen != 0)
 476                         strncat(key, ".", max_key); /* really should be strcat */
 477                 strncat(key, iter.key, max_key);
 478
 479                 ret = handler(handler_arg, iter.val, key);
 480                 if (ret == 1)
 481                 {
 482                         ret = traverse_json_impl(iter.val, key, max_key, handler,
 483                                         handler_arg);
 484                 }
 485                 else if (ret != 0)
 486                 {
 487                         return ret;
 488                 }
 489
 490                 key[plen] = '\0';
 491         }
 492         return 0;
 493 }
 494
 495 static int traverse_json(const char *json, node_handler_t handler,
 496                 void *handler_arg)
 497 {
 498         json_object *root;
 499         char buf[128];
 500         buf[0] = '\0';
 501         root = json_tokener_parse(json);
 502         if (!root)
 503                 return -EDOM;
 504         int result = traverse_json_impl(root, buf, sizeof(buf), handler, handler_arg);
 505         json_object_put(root);
 506         return result;
 507 }
 508
 509 static int node_handler_define_schema(void *arg, json_object *jo,
 510                 const char *key)
 511 {
 512         struct ceph_daemon *d = (struct ceph_daemon *) arg;
 513         int pc_type;
 514         if (json_object_get_type(jo) == json_type_object)
 515                 return 1;
 516         else if (json_object_get_type(jo) != json_type_int)
 517                 return -EDOM;
 518         pc_type = json_object_get_int(jo);
 519         DEBUG("\nceph_daemon_add_ds_entry(d=%s,key=%s,pc_type=%04x)",
 520                         d->name, key, pc_type);
 521         return ceph_daemon_add_ds_entry(d, key, pc_type);
 522 }
 523 struct values_holder
 524 {
 525         int values_len;
 526         value_t *values;
 527 };
 528
 529 /** A set of values_t data that we build up in memory while parsing the JSON. */
 530 struct values_tmp
 531 {
 532         struct ceph_daemon *d;
 533         int holder_num;
 534         struct values_holder vh[0];
 535 };
 536
 537 static int node_handler_fetch_data(void *arg, json_object *jo, const char *key)
 538 {
 539         int dset_idx, ds_idx;
 540         value_t *uv;
 541         char dset_name[DATA_MAX_NAME_LEN];
 542         char ds_name[MAX_RRD_DS_NAME_LEN];
 543         struct values_tmp *vtmp = (struct values_tmp*) arg;
 544         memset(dset_name, 0, sizeof(dset_name));
 545         memset(ds_name, 0, sizeof(ds_name));
 546         if (parse_keys(key, dset_name, ds_name))
 547                 return 1;DEBUG("enter node_handler_fetch_data");
 548         dset_idx = get_matching_dset(vtmp->d, dset_name);
 549         if (dset_idx == -1)
 550                 return 1;
 551         ds_idx = get_matching_value(&vtmp->d->dset[dset_idx], ds_name,
 552                         vtmp->d->dset[dset_idx].ds_num);
 553         if (ds_idx == -1)
 554                 return 1;DEBUG("DSet:%s, DS:%s, DSet idx:%d, DS idx:%d",
 555                         dset_name,ds_name,dset_idx,ds_idx);
 556         uv = &(vtmp->vh[dset_idx].values[ds_idx]);
 557         if (vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_LONGRUNAVG)
 558         {
 559                 json_object *avgcount, *sum;
 560                 uint64_t avgcounti;
 561                 double sumd;
 562                 if (json_object_get_type(jo) != json_type_object)
 563                         return -EINVAL;
 564                 avgcount = json_object_object_get(jo, "avgcount");
 565                 sum = json_object_object_get(jo, "sum");
 566                 if ((!avgcount) || (!sum))
 567                         return -EINVAL;
 568                 avgcounti = json_object_get_int(avgcount);
 569                 DEBUG("avgcounti:%ld",avgcounti);
 570                 if (avgcounti == 0)
 571                         avgcounti = 1;
 572                 sumd = json_object_get_int(sum);
 573                 DEBUG("sumd:%lf",sumd);
 574                 uv->gauge = sumd / avgcounti;
 575                 DEBUG("uv->gauge = sumd / avgcounti = :%lf",uv->gauge);
 576         }
 577         else if (vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_COUNTER)
 578         {
 579                 /* We use json_object_get_double here because anything > 32
 580                  * bits may get truncated by json_object_get_int */
 581                 uv->counter = json_object_get_double(jo);
 582                 DEBUG("uv->counter %ld",(long)uv->counter);
 583         }
 584         else
 585         {
 586                 uv->gauge = json_object_get_double(jo);
 587                 DEBUG("uv->gauge %lf",uv->gauge);
 588         }
 589         return 0;
 590 }
 591
 592 /******* network I/O *******/
 593 enum cstate_t
 594 {
 595         CSTATE_UNCONNECTED = 0,
 596         CSTATE_WRITE_REQUEST,
 597         CSTATE_READ_VERSION,
 598         CSTATE_READ_AMT,
 599         CSTATE_READ_JSON,
 600 };
 601
 602 enum request_type_t
 603 {
 604         ASOK_REQ_VERSION = 0,
 605         ASOK_REQ_DATA = 1,
 606         ASOK_REQ_SCHEMA = 2,
 607         ASOK_REQ_NONE = 1000,
 608 };
 609
 610 struct cconn
 611 {
 612         /** The Ceph daemon that we're talking to */
 613         struct ceph_daemon *d;
 614
 615         /** Request type */
 616         uint32_t request_type;
 617
 618         /** The connection state */
 619         enum cstate_t state;
 620
 621         /** The socket we use to talk to this daemon */
 622         int asok;
 623
 624         /** The amount of data remaining to read / write. */
 625         uint32_t amt;
 626
 627         /** Length of the JSON to read */
 628         uint32_t json_len;
 629
 630         /** Buffer containing JSON data */
 631         char *json;
 632 };
 633
 634 static int cconn_connect(struct cconn *io)
 635 {
 636         struct sockaddr_un address;
 637         int flags, fd, err;
 638         if (io->state != CSTATE_UNCONNECTED)
 639         {
 640                 ERROR("cconn_connect: io->state != CSTATE_UNCONNECTED");
 641                 return -EDOM;
 642         }
 643         fd = socket(PF_UNIX, SOCK_STREAM, 0);
 644         if (fd < 0)
 645         {
 646                 int err = -errno;
 647                 ERROR("cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) failed: "
 648                 "error %d", err);
 649                 return err;
 650         }
 651         memset(&address, 0, sizeof(struct sockaddr_un));
 652         address.sun_family = AF_UNIX;
 653         snprintf(address.sun_path, sizeof(address.sun_path), "%s",
 654                         io->d->asok_path);
 655         RETRY_ON_EINTR(err,
 656                 connect(fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)));
 657         if (err < 0)
 658         {
 659                 ERROR("cconn_connect: connect(%d) failed: error %d", fd, err);
 660                 return err;
 661         }
 662
 663         flags = fcntl(fd, F_GETFL, 0);
 664         if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
 665         {
 666                 err = -errno;
 667                 ERROR("cconn_connect: fcntl(%d, O_NONBLOCK) error %d", fd, err);
 668                 return err;
 669         }
 670         io->asok = fd;
 671         io->state = CSTATE_WRITE_REQUEST;
 672         io->amt = 0;
 673         io->json_len = 0;
 674         io->json = NULL;
 675         return 0;
 676 }
 677
 678 static void cconn_close(struct cconn *io)
 679 {
 680         io->state = CSTATE_UNCONNECTED;
 681         if (io->asok != -1)
 682         {
 683                 int res;
 684                 RETRY_ON_EINTR(res, close(io->asok));
 685         }
 686         io->asok = -1;
 687         io->amt = 0;
 688         io->json_len = 0;
 689         sfree(io->json);
 690         io->json = NULL;
 691 }
 692
 693 /* Process incoming JSON counter data */
 694 /*static int cconn_process_data(struct cconn *io)
 695  {
 696  int ret;
 697  value_list_t vl = VALUE_LIST_INIT;
 698  struct values_tmp *vtmp = calloc(1, sizeof(struct values_tmp) +
 699  (sizeof(value_t) * io->d->dset.ds_num));
 700  if (!vtmp)
 701  return -ENOMEM;
 702  vtmp->d = io->d;
 703  vtmp->values_len = io->d->dset.ds_num;
 704  ret = traverse_json(io->json, node_handler_fetch_data, vtmp);
 705  if (ret)
 706  goto done;
 707  sstrncpy(vl.host, hostname_g, sizeof(vl.host));
 708  sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
 709  sstrncpy(vl.type, io->d->dset.type, sizeof(vl.type));
 710  vl.values = vtmp->values;
 711  vl.values_len = vtmp->values_len;
 712  DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
 713  io->d->dset.type, vl.values_len, io->json);
 714  ret = plugin_dispatch_values(&vl);
 715  done:
 716  sfree(vtmp);
 717  return ret;
 718  }*/
 719 static int cconn_process_data(struct cconn *io)
 720 {
 721         int i, ret = 0;
 722         struct values_tmp *vtmp = calloc(1,
 723                         sizeof(struct values_tmp)
 724                                         + (sizeof(struct values_holder)) * io->d->dset_num);
 725         if (!vtmp)
 726                 return -ENOMEM;
 727         for (i = 0; i < io->d->dset_num; i++)
 728         {
 729                 value_t *val = calloc(1, (sizeof(value_t) * io->d->dset[i].ds_num));
 730                 vtmp->vh[i].values = val;
 731                 vtmp->vh[i].values_len = io->d->dset[i].ds_num;
 732         }
 733         vtmp->d = io->d;
 734         vtmp->holder_num = io->d->dset_num;
 735         ret = traverse_json(io->json, node_handler_fetch_data, vtmp);
 736         if (ret)
 737                 goto done;
 738         for (i = 0; i < vtmp->holder_num; i++)
 739         {
 740                 value_list_t vl = VALUE_LIST_INIT;
 741                 sstrncpy(vl.host, hostname_g, sizeof(vl.host));
 742                 sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
 743                 strncpy(vl.plugin_instance, io->d->name, sizeof(vl.plugin_instance));
 744                 sstrncpy(vl.type, io->d->dset[i].type, sizeof(vl.type));
 745                 vl.values = vtmp->vh[i].values;
 746                 vl.values_len = vtmp->vh[i].values_len;
 747                 DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
 748                                 io->d->name, vl.values_len, io->json);
 749                 ret = plugin_dispatch_values(&vl);
 750                 if (ret)
 751                         goto done;
 752         }
 753
 754         done: for (i = 0; i < vtmp->holder_num; i++)
 755         {
 756                 sfree(vtmp->vh[i].values);
 757         }
 758         sfree(vtmp);
 759         return ret;
 760 }
 761
 762 static int cconn_process_json(struct cconn *io)
 763 {
 764         switch (io->request_type)
 765         {
 766         case ASOK_REQ_DATA:
 767                 return cconn_process_data(io);
 768         case ASOK_REQ_SCHEMA:
 769                 return traverse_json(io->json, node_handler_define_schema, io->d);
 770         default:
 771                 return -EDOM;
 772         }
 773 }
 774
 775 static int cconn_validate_revents(struct cconn *io, int revents)
 776 {
 777         if (revents & POLLERR)
 778         {
 779                 ERROR("cconn_validate_revents(name=%s): got POLLERR", io->d->name);
 780                 return -EIO;
 781         }
 782         switch (io->state)
 783         {
 784         case CSTATE_WRITE_REQUEST:
 785                 return (revents & POLLOUT) ? 0 : -EINVAL;
 786         case CSTATE_READ_VERSION:
 787         case CSTATE_READ_AMT:
 788         case CSTATE_READ_JSON:
 789                 return (revents & POLLIN) ? 0 : -EINVAL;
 790                 return (revents & POLLIN) ? 0 : -EINVAL;
 791         default:
 792                 ERROR("cconn_validate_revents(name=%s) got to illegal state on line %d",
 793                                 io->d->name, __LINE__);
 794                 return -EDOM;
 795         }
 796 }
 797
 798 /** Handle a network event for a connection */
 799 static int cconn_handle_event(struct cconn *io)
 800 {
 801         int ret;
 802         switch (io->state)
 803         {
 804         case CSTATE_UNCONNECTED:
 805                 ERROR("cconn_handle_event(name=%s) got to illegal state on line %d",
 806                                 io->d->name, __LINE__);
 807
 808                 return -EDOM;
 809         case CSTATE_WRITE_REQUEST:
 810         {
 811                 char cmd[32];
 812                 /*snprintf(cmd, sizeof(cmd), "%s%d%s", "{\"prefix\":\"", io->request_type,
 813                  "\"}");*/
 814                 char req_type_str[2];
 815                 snprintf(req_type_str, sizeof(req_type_str), "%1.1d", io->request_type);
 816                 json_object *cmd_object = json_object_new_object();
 817                 json_object_object_add(cmd_object, "prefix",
 818                                 json_object_new_string(req_type_str));
 819                 const char *cmd_json = json_object_to_json_string(cmd_object);
 820                 /** we should send '\n' to server **/
 821                 snprintf(cmd, sizeof(cmd), "%s\n", cmd_json);
 822                 size_t cmd_len = strlen(cmd);
 823                 RETRY_ON_EINTR(ret,
 824                                 write(io->asok, ((char*)&cmd) + io->amt, cmd_len - io->amt));
 825                 DEBUG("cconn_handle_event(name=%s,state=%d,amt=%d,ret=%d)",
 826                                 io->d->name, io->state, io->amt, ret);
 827                 if (ret < 0)
 828                         return ret;
 829                 io->amt += ret;
 830                 if (io->amt >= cmd_len)
 831                 {
 832                         io->amt = 0;
 833                         switch (io->request_type)
 834                         {
 835                         case ASOK_REQ_VERSION:
 836                                 io->state = CSTATE_READ_VERSION;
 837                                 break;
 838                         default:
 839                                 io->state = CSTATE_READ_AMT;
 840                                 break;
 841                         }
 842                 }
 843                 json_object_put(cmd_object);
 844                 return 0;
 845         }
 846         case CSTATE_READ_VERSION:
 847         {
 848                 RETRY_ON_EINTR(ret,
 849                                 read(io->asok, ((char*)(&io->d->version)) + io->amt,
 850                                                 sizeof(io->d->version) - io->amt));
 851                 DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
 852                                 io->d->name, io->state, ret);
 853                 if (ret < 0)
 854                         return ret;
 855                 io->amt += ret;
 856                 if (io->amt >= sizeof(io->d->version))
 857                 {
 858                         io->d->version = ntohl(io->d->version);
 859                         if (io->d->version != 1)
 860                         {
 861                                 ERROR("cconn_handle_event(name=%s) not "
 862                                 "expecting version %d!", io->d->name, io->d->version);
 863                                 return -ENOTSUP;
 864                         }DEBUG("cconn_handle_event(name=%s): identified as "
 865                                         "version %d", io->d->name, io->d->version);
 866                         io->amt = 0;
 867                         cconn_close(io);
 868                         io->request_type = ASOK_REQ_SCHEMA;
 869                 }
 870                 return 0;
 871         }
 872         case CSTATE_READ_AMT:
 873         {
 874                 RETRY_ON_EINTR(ret,
 875                                 read(io->asok, ((char*)(&io->json_len)) + io->amt,
 876                                                 sizeof(io->json_len) - io->amt));
 877                 DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
 878                                 io->d->name, io->state, ret);
 879                 if (ret < 0)
 880                         return ret;
 881                 io->amt += ret;
 882                 if (io->amt >= sizeof(io->json_len))
 883                 {
 884                         io->json_len = ntohl(io->json_len);
 885                         io->amt = 0;
 886                         io->state = CSTATE_READ_JSON;
 887                         io->json = calloc(1, io->json_len + 1);
 888                         if (!io->json)
 889                                 return -ENOMEM;
 890                 }
 891                 return 0;
 892         }
 893         case CSTATE_READ_JSON:
 894         {
 895                 RETRY_ON_EINTR(ret,
 896                                 read(io->asok, io->json + io->amt, io->json_len - io->amt));
 897                 DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
 898                                 io->d->name, io->state, ret);
 899                 if (ret < 0)
 900                         return ret;
 901                 io->amt += ret;
 902                 if (io->amt >= io->json_len)
 903                 {
 904                         ret = cconn_process_json(io);
 905                         if (ret)
 906                                 return ret;
 907                         cconn_close(io);
 908                         io->request_type = ASOK_REQ_NONE;
 909                 }
 910                 return 0;
 911         }
 912         default:
 913                 ERROR("cconn_handle_event(name=%s) got to illegal state on "
 914                 "line %d", io->d->name, __LINE__);
 915                 return -EDOM;
 916         }
 917 }
 918
 919 static int cconn_prepare(struct cconn *io, struct pollfd* fds)
 920 {
 921         int ret;
 922         if (io->request_type == ASOK_REQ_NONE)
 923         {
 924                 /* The request has already been serviced. */
 925                 return 0;
 926         }
 927         else if ((io->request_type == ASOK_REQ_DATA) && (io->d->dset_num == 0))
 928         {
 929                 /* If there are no counters to report on, don't bother
 930                  * connecting */
 931                 return 0;
 932         }
 933
 934         switch (io->state)
 935         {
 936         case CSTATE_UNCONNECTED:
 937                 ret = cconn_connect(io);
 938                 if (ret > 0)
 939                         return -ret;
 940                 else if (ret < 0)
 941                         return ret;
 942                 fds->fd = io->asok;
 943                 fds->events = POLLOUT;
 944                 return 1;
 945         case CSTATE_WRITE_REQUEST:
 946                 fds->fd = io->asok;
 947                 fds->events = POLLOUT;
 948                 return 1;
 949         case CSTATE_READ_VERSION:
 950         case CSTATE_READ_AMT:
 951         case CSTATE_READ_JSON:
 952                 fds->fd = io->asok;
 953                 fds->events = POLLIN;
 954                 return 1;
 955         default:
 956                 ERROR("cconn_prepare(name=%s) got to illegal state on line %d",
 957                                 io->d->name, __LINE__);
 958                 return -EDOM;
 959         }
 960 }
 961
 962 /** Returns the difference between two struct timevals in milliseconds.
 963  * On overflow, we return max/min int.
 964  */
 965 static int milli_diff(const struct timeval *t1, const struct timeval *t2)
 966 {
 967         int64_t ret;
 968         int sec_diff = t1->tv_sec - t2->tv_sec;
 969         int usec_diff = t1->tv_usec - t2->tv_usec;
 970         ret = usec_diff / 1000;
 971         ret += (sec_diff * 1000);
 972         if (ret > INT_MAX)
 973                 return INT_MAX;
 974         else if (ret < INT_MIN)
 975                 return INT_MIN;
 976         return (int) ret;
 977 }
 978
 979 /** This handles the actual network I/O to talk to the Ceph daemons.
 980  */
 981 static int cconn_main_loop(uint32_t request_type)
 982 {
 983         int i, ret, some_unreachable = 0;
 984         struct timeval end_tv;
 985         struct cconn io_array[g_num_daemons];
 986
 987         DEBUG("entering cconn_main_loop(request_type = %d)", request_type);
 988
 989         /* create cconn array */
 990         memset(io_array, 0, sizeof(io_array));
 991         for (i = 0; i < g_num_daemons; ++i)
 992         {
 993                 io_array[i].d = g_daemons[i];
 994                 io_array[i].request_type = request_type;
 995                 io_array[i].state = CSTATE_UNCONNECTED;
 996         }
 997
 998         /** Calculate the time at which we should give up */
 999         gettimeofday(&end_tv, NULL);
1000         end_tv.tv_sec += CEPH_TIMEOUT_INTERVAL;
1001
1002         while (1)
1003         {
1004                 int nfds, diff;
1005                 struct timeval tv;
1006                 struct cconn *polled_io_array[g_num_daemons];
1007                 struct pollfd fds[g_num_daemons];
1008                 memset(fds, 0, sizeof(fds));
1009                 nfds = 0;
1010                 for (i = 0; i < g_num_daemons; ++i)
1011                 {
1012                         struct cconn *io = io_array + i;
1013                         ret = cconn_prepare(io, fds + nfds);
1014                         if (ret < 0)
1015                         {
1016                                 WARNING("ERROR: cconn_prepare(name=%s,i=%d,st=%d)=%d",
1017                                                 io->d->name, i, io->state, ret);
1018                                 cconn_close(io);
1019                                 io->request_type = ASOK_REQ_NONE;
1020                                 some_unreachable = 1;
1021                         }
1022                         else if (ret == 1)
1023                         {
1024                                 DEBUG("did cconn_prepare(name=%s,i=%d,st=%d)",
1025                                                 io->d->name, i, io->state);
1026                                 polled_io_array[nfds++] = io_array + i;
1027                         }
1028                 }
1029                 if (nfds == 0)
1030                 {
1031                         /* finished */
1032                         ret = 0;
1033                         DEBUG("cconn_main_loop: no more cconn to manage.");
1034                         goto done;
1035                 }
1036                 gettimeofday(&tv, NULL);
1037                 diff = milli_diff(&end_tv, &tv);
1038                 if (diff <= 0)
1039                 {
1040                         /* Timed out */
1041                         ret = -ETIMEDOUT;
1042                         WARNING("ERROR: cconn_main_loop: timed out.\n");
1043                         goto done;
1044                 }
1045                 RETRY_ON_EINTR(ret, poll(fds, nfds, diff));
1046                 if (ret < 0)
1047                 {
1048                         ERROR("poll(2) error: %d", ret);
1049                         goto done;
1050                 }
1051                 for (i = 0; i < nfds; ++i)
1052                 {
1053                         struct cconn *io = polled_io_array[i];
1054                         int revents = fds[i].revents;
1055                         if (revents == 0)
1056                         {
1057                                 /* do nothing */
1058                         }
1059                         else if (cconn_validate_revents(io, revents))
1060                         {
1061                                 WARNING("ERROR: cconn(name=%s,i=%d,st=%d): "
1062                                 "revents validation error: "
1063                                 "revents=0x%08x", io->d->name, i, io->state, revents);
1064                                 cconn_close(io);
1065                                 io->request_type = ASOK_REQ_NONE;
1066                                 some_unreachable = 1;
1067                         }
1068                         else
1069                         {
1070                                 int ret = cconn_handle_event(io);
1071                                 if (ret)
1072                                 {
1073                                         WARNING("ERROR: cconn_handle_event(name=%s,"
1074                                         "i=%d,st=%d): error %d", io->d->name, i, io->state, ret);
1075                                         cconn_close(io);
1076                                         io->request_type = ASOK_REQ_NONE;
1077                                         some_unreachable = 1;
1078                                 }
1079                         }
1080                 }
1081         }
1082         done: for (i = 0; i < g_num_daemons; ++i)
1083         {
1084                 cconn_close(io_array + i);
1085         }
1086         if (some_unreachable)
1087         {
1088                 DEBUG("cconn_main_loop: some Ceph daemons were unreachable.");
1089         }
1090         else
1091         {
1092                 DEBUG("cconn_main_loop: reached all Ceph daemons :)");
1093         }
1094         return ret;
1095 }
1096
1097 static int ceph_read(void)
1098 {
1099         return cconn_main_loop(ASOK_REQ_DATA);
1100 }
1101
1102 /******* lifecycle *******/
1103 static int ceph_init(void)
1104 {
1105         int i, ret, j;
1106         DEBUG("ceph_init");
1107         ceph_daemons_print();
1108
1109         ret = cconn_main_loop(ASOK_REQ_VERSION);
1110         if (ret)
1111                 return ret;
1112         for (i = 0; i < g_num_daemons; ++i)
1113         {
1114                 struct ceph_daemon *d = g_daemons[i];
1115                 for (j = 0; j < d->dset_num; j++)
1116                 {
1117                         ret = plugin_register_data_set(d->dset + j);
1118                         if (ret)
1119                         {
1120                                 ERROR("plugin_register_data_set(%s) failed!", d->name);
1121                         }
1122                         else
1123                         {
1124                                 DEBUG("plugin_register_data_set(%s): "
1125                                                 "(d->dset)[%d]->ds_num=%d",
1126                                                 d->name, j, d->dset[j].ds_num);
1127                         }
1128                 }
1129         }
1130         return 0;
1131 }
1132
1133 static int ceph_shutdown(void)
1134 {
1135         int i;
1136         for (i = 0; i < g_num_daemons; ++i)
1137         {
1138                 ceph_daemon_free(g_daemons[i]);
1139         }
1140         sfree(g_daemons);
1141         g_daemons = NULL;
1142         g_num_daemons = 0;
1143         DEBUG("finished ceph_shutdown");
1144         return 0;
1145 }
1146
1147 void module_register(void)
1148 {
1149         plugin_register_complex_config("ceph", ceph_config);
1150         plugin_register_init("ceph", ceph_init);
1151         plugin_register_read("ceph", ceph_read);
1152         plugin_register_shutdown("ceph", ceph_shutdown);
1153 }