2 * collectd - src/ceph.c
3 * Copyright (C) 2011 New Dream Network
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.
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.
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
19 * Colin McCabe <cmccabe@alumni.cmu.edu>
20 * Dennis Zou <yunzou@cisco.com>
21 * Dan Ryder <daryder@cisco.com>
30 #include <arpa/inet.h>
33 #include <json/json.h>
34 #include <json/json_object_private.h> /* need for struct json_object_iter */
42 #include <sys/socket.h>
44 #include <sys/types.h>
47 #define MAX_RRD_DS_NAME_LEN 20
49 #define RETRY_ON_EINTR(ret, expr) \
59 /** Timeout interval in seconds */
60 #define CEPH_TIMEOUT_INTERVAL 1
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))
65 /******* ceph_daemon *******/
68 /** Version of the admin_socket interface */
71 char name[DATA_MAX_NAME_LEN];
75 /** Path to the socket that we use to talk to the ceph daemon */
76 char asok_path[UNIX_DOMAIN_SOCK_PATH_MAX];
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
83 //struct data_set_s dset;
84 /** Dynamically allocated array **/
85 struct data_set_s *dset;
89 enum perfcounter_type_d
91 PERFCOUNTER_LONGRUNAVG = 0x4, PERFCOUNTER_COUNTER = 0x8,
94 /** Array of daemons to monitor */
95 static struct ceph_daemon **g_daemons = NULL;
97 /** Number of elements in g_daemons */
98 static int g_num_daemons = 0;
100 static void ceph_daemon_print(const struct ceph_daemon *d)
102 DEBUG("name=%s, asok_path=%s", d->name, d->asok_path);
105 static void ceph_daemons_print(void)
108 for (i = 0; i < g_num_daemons; ++i)
110 ceph_daemon_print(g_daemons[i]);
114 /*static void ceph_daemon_free(struct ceph_daemon *d)
116 plugin_unregister_data_set(d->dset.type);
120 static void ceph_daemon_free(struct ceph_daemon *d)
123 for (; i < d->dset_num; i++)
125 plugin_unregister_data_set((d->dset + i)->type);
127 sfree(d->pc_types[i]);
134 static void compact_ds_name(char *source, char *dest)
137 char *save_ptr = NULL, *tmp_ptr = source;
140 char tmp[DATA_MAX_NAME_LEN];
143 memset(tmp, 0, sizeof(tmp));
144 if (source == NULL || dest == NULL || source[0] == '\0' || dest[0] != '\0')
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)
156 /** capitalize 1st char **/
157 keys[keys_num][0] = toupper(keys[keys_num][0]);
162 /** concatenate each part of source string **/
163 for (i = 0; i < keys_num; i++)
165 strcat(tmp, keys[i]);
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
173 if (strlen(tmp) > MAX_RRD_DS_NAME_LEN - 1)
175 append_status |= 0x4;
176 /** we should reserve space for
181 if (append_status & 0x1)
183 /** we should reserve space for
188 if (append_status & 0x2)
190 /** we should reserve space for
195 snprintf(dest, MAX_RRD_DS_NAME_LEN - reserved, "%s", tmp);
196 offset = strlen(dest);
197 switch (append_status)
200 memcpy(dest + offset, "Minus", 5);
203 memcpy(dest + offset, "Plus", 5);
206 memcpy(dest + offset, len_str, 2);
209 memcpy(dest + offset, "Minus", 5);
210 memcpy(dest + offset + 5, len_str, 2);
213 memcpy(dest + offset, "Plus", 4);
214 memcpy(dest + offset + 4, len_str, 2);
220 static int parse_keys(const char *key_str, char *dset_name, char *ds_name)
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')
232 if ((ptr = strchr(key_str, '.')) == NULL
233 || (rptr = strrchr(key_str, '.')) == NULL)
235 strncpy(dset_name, key_str, DATA_MAX_NAME_LEN - 1);
236 strncpy(tmp_ds_name, key_str, DATA_MAX_NAME_LEN - 1);
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);
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);
252 {/** if last key isn't "type", copy last key **/
253 strncpy(tmp_ds_name, rptr + 1, DATA_MAX_NAME_LEN - 1);
256 else if (!strncmp(rptr + 1, "type", 4))
257 {/** more than two keys **/
258 memcpy(tmp_ds_name, ptr + 1, ds_name_len - 1);
261 {/** copy whole keys **/
262 strncpy(tmp_ds_name, ptr + 1, DATA_MAX_NAME_LEN - 1);
264 compact: compact_ds_name(tmp_ds_name, ds_name);
268 int get_matching_dset(const struct ceph_daemon *d, const char *name)
271 for (idx = 0; idx < d->dset_num; ++idx)
273 if (strcmp(d->dset[idx].type, name) == 0)
281 int get_matching_value(const struct data_set_s *dset, const char *name,
285 for (idx = 0; idx < num_values; ++idx)
287 if (strcmp(dset->ds[idx].name, name) == 0)
295 static int ceph_daemon_add_ds_entry(struct ceph_daemon *d, const char *name,
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;
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))
313 idx = get_matching_dset(d, dset_name);
315 {/* need to add a dset **/
316 dset_array = realloc(d->dset,
317 sizeof(struct data_set_s) * (d->dset_num + 1));
320 pc_types_array = realloc(d->pc_types,
321 sizeof(int *) * (d->dset_num + 1));
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
329 memset(dset, 0, sizeof(struct data_set_s));
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;
337 dset = &d->dset[idx];
338 pc_types = d->pc_types[idx];
340 struct data_source_s *ds_array = realloc(dset->ds,
341 sizeof(struct data_source_s) * (dset->ds_num + 1));
346 pc_types_new = realloc(pc_types, sizeof(int) * (dset->ds_num + 1));
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;
361 d->pc_types[idx] = pc_types_new;
362 d->pc_types[idx][dset->ds_num] = pc_type;
364 ds = &ds_array[dset->ds_num++];
365 snprintf(ds->name, MAX_RRD_DS_NAME_LEN, "%s", ds_name);
367 (pc_type & PERFCOUNTER_COUNTER) ? DS_TYPE_COUNTER : DS_TYPE_GAUGE;
373 /******* ceph_config *******/
374 static int cc_handle_str(struct oconfig_item_s *item, char *dest, int dest_len)
377 if (item->values_num != 1)
381 if (item->values[0].type != OCONFIG_TYPE_STRING)
385 val = item->values[0].value.string;
386 if (snprintf(dest, dest_len, "%s", val) > (dest_len - 1))
388 ERROR("ceph plugin: configuration parameter '%s' is too long.\n",
390 return -ENAMETOOLONG;
395 static int ceph_config(oconfig_item_t *ci)
398 struct ceph_daemon *array, *nd, cd;
399 memset(&cd, 0, sizeof(struct ceph_daemon));
401 for (i = 0; i < ci->children_num; ++i)
403 oconfig_item_t *child = ci->children + i;
404 if (strcasecmp("Name", child->key) == 0)
406 ret = cc_handle_str(child, cd.name, DATA_MAX_NAME_LEN);
410 else if (strcasecmp("SocketPath", child->key) == 0)
412 ret = cc_handle_str(child, cd.asok_path, sizeof(cd.asok_path));
418 WARNING("ceph plugin: ignoring unknown option %s", child->key);
421 if (cd.name[0] == '\0')
423 ERROR("ceph plugin: you must configure a daemon name.\n");
426 else if (cd.asok_path[0] == '\0')
428 ERROR("ceph plugin(name=%s): you must configure an administrative "
429 "socket path.\n", cd.name);
432 else if (!((cd.asok_path[0] == '/')
433 || (cd.asok_path[0] == '.' && cd.asok_path[1] == '/')))
435 ERROR("ceph plugin(name=%s): administrative socket paths must begin with "
436 "'/' or './' Can't parse: '%s'\n", cd.name, cd.asok_path);
439 array = realloc(g_daemons,
440 sizeof(struct ceph_daemon *) * (g_num_daemons + 1));
443 /* The positive return value here indicates that this is a
444 * runtime error, not a configuration error. */
447 g_daemons = (struct ceph_daemon**) array;
448 nd = malloc(sizeof(struct ceph_daemon));
451 memcpy(nd, &cd, sizeof(struct ceph_daemon));
452 g_daemons[g_num_daemons++] = nd;
456 /******* JSON parsing *******/
457 typedef int (*node_handler_t)(void*, json_object*, const char*);
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)
464 struct json_object_iter iter;
467 if (json_object_get_type(jo) != json_type_object)
470 json_object_object_foreachC(jo, iter)
472 klen = strlen(iter.key);
473 if (plen + klen + 2 > max_key)
474 return -ENAMETOOLONG;
476 strncat(key, ".", max_key); /* really should be strcat */
477 strncat(key, iter.key, max_key);
479 ret = handler(handler_arg, iter.val, key);
482 ret = traverse_json_impl(iter.val, key, max_key, handler,
495 static int traverse_json(const char *json, node_handler_t handler,
501 root = json_tokener_parse(json);
504 int result = traverse_json_impl(root, buf, sizeof(buf), handler, handler_arg);
505 json_object_put(root);
509 static int node_handler_define_schema(void *arg, json_object *jo,
512 struct ceph_daemon *d = (struct ceph_daemon *) arg;
514 if (json_object_get_type(jo) == json_type_object)
516 else if (json_object_get_type(jo) != json_type_int)
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);
529 /** A set of values_t data that we build up in memory while parsing the JSON. */
532 struct ceph_daemon *d;
534 struct values_holder vh[0];
537 static int node_handler_fetch_data(void *arg, json_object *jo, const char *key)
539 int dset_idx, ds_idx;
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);
551 ds_idx = get_matching_value(&vtmp->d->dset[dset_idx], ds_name,
552 vtmp->d->dset[dset_idx].ds_num);
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)
559 json_object *avgcount, *sum;
562 if (json_object_get_type(jo) != json_type_object)
564 avgcount = json_object_object_get(jo, "avgcount");
565 sum = json_object_object_get(jo, "sum");
566 if ((!avgcount) || (!sum))
568 avgcounti = json_object_get_int(avgcount);
569 DEBUG("avgcounti:%ld",avgcounti);
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);
577 else if (vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_COUNTER)
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);
586 uv->gauge = json_object_get_double(jo);
587 DEBUG("uv->gauge %lf",uv->gauge);
592 /******* network I/O *******/
595 CSTATE_UNCONNECTED = 0,
596 CSTATE_WRITE_REQUEST,
604 ASOK_REQ_VERSION = 0,
607 ASOK_REQ_NONE = 1000,
612 /** The Ceph daemon that we're talking to */
613 struct ceph_daemon *d;
616 uint32_t request_type;
618 /** The connection state */
621 /** The socket we use to talk to this daemon */
624 /** The amount of data remaining to read / write. */
627 /** Length of the JSON to read */
630 /** Buffer containing JSON data */
634 static int cconn_connect(struct cconn *io)
636 struct sockaddr_un address;
638 if (io->state != CSTATE_UNCONNECTED)
640 ERROR("cconn_connect: io->state != CSTATE_UNCONNECTED");
643 fd = socket(PF_UNIX, SOCK_STREAM, 0);
647 ERROR("cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) failed: "
651 memset(&address, 0, sizeof(struct sockaddr_un));
652 address.sun_family = AF_UNIX;
653 snprintf(address.sun_path, sizeof(address.sun_path), "%s",
656 connect(fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)));
659 ERROR("cconn_connect: connect(%d) failed: error %d", fd, err);
663 flags = fcntl(fd, F_GETFL, 0);
664 if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
667 ERROR("cconn_connect: fcntl(%d, O_NONBLOCK) error %d", fd, err);
671 io->state = CSTATE_WRITE_REQUEST;
678 static void cconn_close(struct cconn *io)
680 io->state = CSTATE_UNCONNECTED;
684 RETRY_ON_EINTR(res, close(io->asok));
693 /* Process incoming JSON counter data */
694 /*static int cconn_process_data(struct cconn *io)
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));
703 vtmp->values_len = io->d->dset.ds_num;
704 ret = traverse_json(io->json, node_handler_fetch_data, vtmp);
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);
719 static int cconn_process_data(struct cconn *io)
722 struct values_tmp *vtmp = calloc(1,
723 sizeof(struct values_tmp)
724 + (sizeof(struct values_holder)) * io->d->dset_num);
727 for (i = 0; i < io->d->dset_num; i++)
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;
734 vtmp->holder_num = io->d->dset_num;
735 ret = traverse_json(io->json, node_handler_fetch_data, vtmp);
738 for (i = 0; i < vtmp->holder_num; i++)
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);
754 done: for (i = 0; i < vtmp->holder_num; i++)
756 sfree(vtmp->vh[i].values);
762 static int cconn_process_json(struct cconn *io)
764 switch (io->request_type)
767 return cconn_process_data(io);
768 case ASOK_REQ_SCHEMA:
769 return traverse_json(io->json, node_handler_define_schema, io->d);
775 static int cconn_validate_revents(struct cconn *io, int revents)
777 if (revents & POLLERR)
779 ERROR("cconn_validate_revents(name=%s): got POLLERR", io->d->name);
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;
792 ERROR("cconn_validate_revents(name=%s) got to illegal state on line %d",
793 io->d->name, __LINE__);
798 /** Handle a network event for a connection */
799 static int cconn_handle_event(struct cconn *io)
804 case CSTATE_UNCONNECTED:
805 ERROR("cconn_handle_event(name=%s) got to illegal state on line %d",
806 io->d->name, __LINE__);
809 case CSTATE_WRITE_REQUEST:
812 /*snprintf(cmd, sizeof(cmd), "%s%d%s", "{\"prefix\":\"", io->request_type,
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);
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);
830 if (io->amt >= cmd_len)
833 switch (io->request_type)
835 case ASOK_REQ_VERSION:
836 io->state = CSTATE_READ_VERSION;
839 io->state = CSTATE_READ_AMT;
843 json_object_put(cmd_object);
846 case CSTATE_READ_VERSION:
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);
856 if (io->amt >= sizeof(io->d->version))
858 io->d->version = ntohl(io->d->version);
859 if (io->d->version != 1)
861 ERROR("cconn_handle_event(name=%s) not "
862 "expecting version %d!", io->d->name, io->d->version);
864 }DEBUG("cconn_handle_event(name=%s): identified as "
865 "version %d", io->d->name, io->d->version);
868 io->request_type = ASOK_REQ_SCHEMA;
872 case CSTATE_READ_AMT:
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);
882 if (io->amt >= sizeof(io->json_len))
884 io->json_len = ntohl(io->json_len);
886 io->state = CSTATE_READ_JSON;
887 io->json = calloc(1, io->json_len + 1);
893 case CSTATE_READ_JSON:
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);
902 if (io->amt >= io->json_len)
904 ret = cconn_process_json(io);
908 io->request_type = ASOK_REQ_NONE;
913 ERROR("cconn_handle_event(name=%s) got to illegal state on "
914 "line %d", io->d->name, __LINE__);
919 static int cconn_prepare(struct cconn *io, struct pollfd* fds)
922 if (io->request_type == ASOK_REQ_NONE)
924 /* The request has already been serviced. */
927 else if ((io->request_type == ASOK_REQ_DATA) && (io->d->dset_num == 0))
929 /* If there are no counters to report on, don't bother
936 case CSTATE_UNCONNECTED:
937 ret = cconn_connect(io);
943 fds->events = POLLOUT;
945 case CSTATE_WRITE_REQUEST:
947 fds->events = POLLOUT;
949 case CSTATE_READ_VERSION:
950 case CSTATE_READ_AMT:
951 case CSTATE_READ_JSON:
953 fds->events = POLLIN;
956 ERROR("cconn_prepare(name=%s) got to illegal state on line %d",
957 io->d->name, __LINE__);
962 /** Returns the difference between two struct timevals in milliseconds.
963 * On overflow, we return max/min int.
965 static int milli_diff(const struct timeval *t1, const struct timeval *t2)
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);
974 else if (ret < INT_MIN)
979 /** This handles the actual network I/O to talk to the Ceph daemons.
981 static int cconn_main_loop(uint32_t request_type)
983 int i, ret, some_unreachable = 0;
984 struct timeval end_tv;
985 struct cconn io_array[g_num_daemons];
987 DEBUG("entering cconn_main_loop(request_type = %d)", request_type);
989 /* create cconn array */
990 memset(io_array, 0, sizeof(io_array));
991 for (i = 0; i < g_num_daemons; ++i)
993 io_array[i].d = g_daemons[i];
994 io_array[i].request_type = request_type;
995 io_array[i].state = CSTATE_UNCONNECTED;
998 /** Calculate the time at which we should give up */
999 gettimeofday(&end_tv, NULL);
1000 end_tv.tv_sec += CEPH_TIMEOUT_INTERVAL;
1006 struct cconn *polled_io_array[g_num_daemons];
1007 struct pollfd fds[g_num_daemons];
1008 memset(fds, 0, sizeof(fds));
1010 for (i = 0; i < g_num_daemons; ++i)
1012 struct cconn *io = io_array + i;
1013 ret = cconn_prepare(io, fds + nfds);
1016 WARNING("ERROR: cconn_prepare(name=%s,i=%d,st=%d)=%d",
1017 io->d->name, i, io->state, ret);
1019 io->request_type = ASOK_REQ_NONE;
1020 some_unreachable = 1;
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;
1033 DEBUG("cconn_main_loop: no more cconn to manage.");
1036 gettimeofday(&tv, NULL);
1037 diff = milli_diff(&end_tv, &tv);
1042 WARNING("ERROR: cconn_main_loop: timed out.\n");
1045 RETRY_ON_EINTR(ret, poll(fds, nfds, diff));
1048 ERROR("poll(2) error: %d", ret);
1051 for (i = 0; i < nfds; ++i)
1053 struct cconn *io = polled_io_array[i];
1054 int revents = fds[i].revents;
1059 else if (cconn_validate_revents(io, revents))
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);
1065 io->request_type = ASOK_REQ_NONE;
1066 some_unreachable = 1;
1070 int ret = cconn_handle_event(io);
1073 WARNING("ERROR: cconn_handle_event(name=%s,"
1074 "i=%d,st=%d): error %d", io->d->name, i, io->state, ret);
1076 io->request_type = ASOK_REQ_NONE;
1077 some_unreachable = 1;
1082 done: for (i = 0; i < g_num_daemons; ++i)
1084 cconn_close(io_array + i);
1086 if (some_unreachable)
1088 DEBUG("cconn_main_loop: some Ceph daemons were unreachable.");
1092 DEBUG("cconn_main_loop: reached all Ceph daemons :)");
1097 static int ceph_read(void)
1099 return cconn_main_loop(ASOK_REQ_DATA);
1102 /******* lifecycle *******/
1103 static int ceph_init(void)
1107 ceph_daemons_print();
1109 ret = cconn_main_loop(ASOK_REQ_VERSION);
1112 for (i = 0; i < g_num_daemons; ++i)
1114 struct ceph_daemon *d = g_daemons[i];
1115 for (j = 0; j < d->dset_num; j++)
1117 ret = plugin_register_data_set(d->dset + j);
1120 ERROR("plugin_register_data_set(%s) failed!", d->name);
1124 DEBUG("plugin_register_data_set(%s): "
1125 "(d->dset)[%d]->ds_num=%d",
1126 d->name, j, d->dset[j].ds_num);
1133 static int ceph_shutdown(void)
1136 for (i = 0; i < g_num_daemons; ++i)
1138 ceph_daemon_free(g_daemons[i]);
1143 DEBUG("finished ceph_shutdown");
1147 void module_register(void)
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);