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 cc_add_daemon_config(oconfig_item_t *ci)
398 struct ceph_daemon *array, *nd, cd;
399 memset(&cd, 0, sizeof(struct ceph_daemon));
401 if ((ci->values_num != 1) || (ci->values[0].type != OCONFIG_TYPE_STRING))
403 WARNING("ceph plugin: `Daemon' blocks need exactly one string argument.");
407 ret = cc_handle_str(ci, cd.name, DATA_MAX_NAME_LEN);
411 for (i=0; i < ci->children_num; i++)
413 oconfig_item_t *child = ci->children + i;
415 if (strcasecmp("SocketPath", child->key) == 0)
417 ret = cc_handle_str(child, cd.asok_path, sizeof(cd.asok_path));
423 WARNING("ceph plugin: ignoring unknown option %s", child->key);
426 if (cd.name[0] == '\0')
428 ERROR("ceph plugin: you must configure a daemon name.\n");
431 else if (cd.asok_path[0] == '\0')
433 ERROR("ceph plugin(name=%s): you must configure an administrative "
434 "socket path.\n", cd.name);
437 else if (!((cd.asok_path[0] == '/')
438 || (cd.asok_path[0] == '.' && cd.asok_path[1] == '/')))
440 ERROR("ceph plugin(name=%s): administrative socket paths must begin with "
441 "'/' or './' Can't parse: '%s'\n", cd.name, cd.asok_path);
444 array = realloc(g_daemons,
445 sizeof(struct ceph_daemon *) * (g_num_daemons + 1));
448 /* The positive return value here indicates that this is a
449 * runtime error, not a configuration error. */
452 g_daemons = (struct ceph_daemon**) array;
453 nd = malloc(sizeof(struct ceph_daemon));
456 memcpy(nd, &cd, sizeof(struct ceph_daemon));
457 g_daemons[g_num_daemons++] = nd;
461 static int ceph_config(oconfig_item_t *ci)
465 for (i = 0; i < ci->children_num; ++i)
467 oconfig_item_t *child = ci->children + i;
468 if (strcasecmp("Daemon", child->key) == 0)
470 ret = cc_add_daemon_config(child);
476 WARNING("ceph plugin: ignoring unknown option %s", child->key);
482 /******* JSON parsing *******/
483 typedef int (*node_handler_t)(void*, json_object*, const char*);
485 /** Perform a depth-first traversal of the JSON parse tree,
486 * calling node_handler at each node.*/
487 static int traverse_json_impl(json_object *jo, char *key, int max_key,
488 node_handler_t handler, void *handler_arg)
490 struct json_object_iter iter;
493 if (json_object_get_type(jo) != json_type_object)
496 json_object_object_foreachC(jo, iter)
498 klen = strlen(iter.key);
499 if (plen + klen + 2 > max_key)
500 return -ENAMETOOLONG;
502 strncat(key, ".", max_key); /* really should be strcat */
503 strncat(key, iter.key, max_key);
505 ret = handler(handler_arg, iter.val, key);
508 ret = traverse_json_impl(iter.val, key, max_key, handler,
521 static int traverse_json(const char *json, node_handler_t handler,
527 root = json_tokener_parse(json);
530 int result = traverse_json_impl(root, buf, sizeof(buf), handler, handler_arg);
531 json_object_put(root);
535 static int node_handler_define_schema(void *arg, json_object *jo,
538 struct ceph_daemon *d = (struct ceph_daemon *) arg;
540 if (json_object_get_type(jo) == json_type_object)
542 else if (json_object_get_type(jo) != json_type_int)
544 pc_type = json_object_get_int(jo);
545 DEBUG("\nceph_daemon_add_ds_entry(d=%s,key=%s,pc_type=%04x)",
546 d->name, key, pc_type);
547 return ceph_daemon_add_ds_entry(d, key, pc_type);
555 /** A set of values_t data that we build up in memory while parsing the JSON. */
558 struct ceph_daemon *d;
560 struct values_holder vh[0];
563 static int node_handler_fetch_data(void *arg, json_object *jo, const char *key)
565 int dset_idx, ds_idx;
567 char dset_name[DATA_MAX_NAME_LEN];
568 char ds_name[MAX_RRD_DS_NAME_LEN];
569 struct values_tmp *vtmp = (struct values_tmp*) arg;
570 memset(dset_name, 0, sizeof(dset_name));
571 memset(ds_name, 0, sizeof(ds_name));
572 if (parse_keys(key, dset_name, ds_name))
573 return 1;DEBUG("enter node_handler_fetch_data");
574 dset_idx = get_matching_dset(vtmp->d, dset_name);
577 ds_idx = get_matching_value(&vtmp->d->dset[dset_idx], ds_name,
578 vtmp->d->dset[dset_idx].ds_num);
580 return 1;DEBUG("DSet:%s, DS:%s, DSet idx:%d, DS idx:%d",
581 dset_name,ds_name,dset_idx,ds_idx);
582 uv = &(vtmp->vh[dset_idx].values[ds_idx]);
583 if (vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_LONGRUNAVG)
585 json_object *avgcount, *sum;
588 if (json_object_get_type(jo) != json_type_object)
590 avgcount = json_object_object_get(jo, "avgcount");
591 sum = json_object_object_get(jo, "sum");
592 if ((!avgcount) || (!sum))
594 avgcounti = json_object_get_int(avgcount);
595 DEBUG("avgcounti:%ld",avgcounti);
598 sumd = json_object_get_int(sum);
599 DEBUG("sumd:%lf",sumd);
600 uv->gauge = sumd / avgcounti;
601 DEBUG("uv->gauge = sumd / avgcounti = :%lf",uv->gauge);
603 else if (vtmp->d->pc_types[dset_idx][ds_idx] & PERFCOUNTER_COUNTER)
605 /* We use json_object_get_double here because anything > 32
606 * bits may get truncated by json_object_get_int */
607 uv->counter = json_object_get_double(jo);
608 DEBUG("uv->counter %ld",(long)uv->counter);
612 uv->gauge = json_object_get_double(jo);
613 DEBUG("uv->gauge %lf",uv->gauge);
618 /******* network I/O *******/
621 CSTATE_UNCONNECTED = 0,
622 CSTATE_WRITE_REQUEST,
630 ASOK_REQ_VERSION = 0,
633 ASOK_REQ_NONE = 1000,
638 /** The Ceph daemon that we're talking to */
639 struct ceph_daemon *d;
642 uint32_t request_type;
644 /** The connection state */
647 /** The socket we use to talk to this daemon */
650 /** The amount of data remaining to read / write. */
653 /** Length of the JSON to read */
656 /** Buffer containing JSON data */
660 static int cconn_connect(struct cconn *io)
662 struct sockaddr_un address;
664 if (io->state != CSTATE_UNCONNECTED)
666 ERROR("cconn_connect: io->state != CSTATE_UNCONNECTED");
669 fd = socket(PF_UNIX, SOCK_STREAM, 0);
673 ERROR("cconn_connect: socket(PF_UNIX, SOCK_STREAM, 0) failed: "
677 memset(&address, 0, sizeof(struct sockaddr_un));
678 address.sun_family = AF_UNIX;
679 snprintf(address.sun_path, sizeof(address.sun_path), "%s",
682 connect(fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)));
685 ERROR("cconn_connect: connect(%d) failed: error %d", fd, err);
689 flags = fcntl(fd, F_GETFL, 0);
690 if (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0)
693 ERROR("cconn_connect: fcntl(%d, O_NONBLOCK) error %d", fd, err);
697 io->state = CSTATE_WRITE_REQUEST;
704 static void cconn_close(struct cconn *io)
706 io->state = CSTATE_UNCONNECTED;
710 RETRY_ON_EINTR(res, close(io->asok));
719 /* Process incoming JSON counter data */
720 /*static int cconn_process_data(struct cconn *io)
723 value_list_t vl = VALUE_LIST_INIT;
724 struct values_tmp *vtmp = calloc(1, sizeof(struct values_tmp) +
725 (sizeof(value_t) * io->d->dset.ds_num));
729 vtmp->values_len = io->d->dset.ds_num;
730 ret = traverse_json(io->json, node_handler_fetch_data, vtmp);
733 sstrncpy(vl.host, hostname_g, sizeof(vl.host));
734 sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
735 sstrncpy(vl.type, io->d->dset.type, sizeof(vl.type));
736 vl.values = vtmp->values;
737 vl.values_len = vtmp->values_len;
738 DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
739 io->d->dset.type, vl.values_len, io->json);
740 ret = plugin_dispatch_values(&vl);
745 static int cconn_process_data(struct cconn *io)
748 struct values_tmp *vtmp = calloc(1,
749 sizeof(struct values_tmp)
750 + (sizeof(struct values_holder)) * io->d->dset_num);
753 for (i = 0; i < io->d->dset_num; i++)
755 value_t *val = calloc(1, (sizeof(value_t) * io->d->dset[i].ds_num));
756 vtmp->vh[i].values = val;
757 vtmp->vh[i].values_len = io->d->dset[i].ds_num;
760 vtmp->holder_num = io->d->dset_num;
761 ret = traverse_json(io->json, node_handler_fetch_data, vtmp);
764 for (i = 0; i < vtmp->holder_num; i++)
766 value_list_t vl = VALUE_LIST_INIT;
767 sstrncpy(vl.host, hostname_g, sizeof(vl.host));
768 sstrncpy(vl.plugin, "ceph", sizeof(vl.plugin));
769 strncpy(vl.plugin_instance, io->d->name, sizeof(vl.plugin_instance));
770 sstrncpy(vl.type, io->d->dset[i].type, sizeof(vl.type));
771 vl.values = vtmp->vh[i].values;
772 vl.values_len = vtmp->vh[i].values_len;
773 DEBUG("cconn_process_data(io=%s): vl.values_len=%d, json=\"%s\"",
774 io->d->name, vl.values_len, io->json);
775 ret = plugin_dispatch_values(&vl);
780 done: for (i = 0; i < vtmp->holder_num; i++)
782 sfree(vtmp->vh[i].values);
788 static int cconn_process_json(struct cconn *io)
790 switch (io->request_type)
793 return cconn_process_data(io);
794 case ASOK_REQ_SCHEMA:
795 return traverse_json(io->json, node_handler_define_schema, io->d);
801 static int cconn_validate_revents(struct cconn *io, int revents)
803 if (revents & POLLERR)
805 ERROR("cconn_validate_revents(name=%s): got POLLERR", io->d->name);
810 case CSTATE_WRITE_REQUEST:
811 return (revents & POLLOUT) ? 0 : -EINVAL;
812 case CSTATE_READ_VERSION:
813 case CSTATE_READ_AMT:
814 case CSTATE_READ_JSON:
815 return (revents & POLLIN) ? 0 : -EINVAL;
816 return (revents & POLLIN) ? 0 : -EINVAL;
818 ERROR("cconn_validate_revents(name=%s) got to illegal state on line %d",
819 io->d->name, __LINE__);
824 /** Handle a network event for a connection */
825 static int cconn_handle_event(struct cconn *io)
830 case CSTATE_UNCONNECTED:
831 ERROR("cconn_handle_event(name=%s) got to illegal state on line %d",
832 io->d->name, __LINE__);
835 case CSTATE_WRITE_REQUEST:
838 /*snprintf(cmd, sizeof(cmd), "%s%d%s", "{\"prefix\":\"", io->request_type,
840 char req_type_str[2];
841 snprintf(req_type_str, sizeof(req_type_str), "%1.1d", io->request_type);
842 json_object *cmd_object = json_object_new_object();
843 json_object_object_add(cmd_object, "prefix",
844 json_object_new_string(req_type_str));
845 const char *cmd_json = json_object_to_json_string(cmd_object);
846 /** we should send '\n' to server **/
847 snprintf(cmd, sizeof(cmd), "%s\n", cmd_json);
848 size_t cmd_len = strlen(cmd);
850 write(io->asok, ((char*)&cmd) + io->amt, cmd_len - io->amt));
851 DEBUG("cconn_handle_event(name=%s,state=%d,amt=%d,ret=%d)",
852 io->d->name, io->state, io->amt, ret);
856 if (io->amt >= cmd_len)
859 switch (io->request_type)
861 case ASOK_REQ_VERSION:
862 io->state = CSTATE_READ_VERSION;
865 io->state = CSTATE_READ_AMT;
869 json_object_put(cmd_object);
872 case CSTATE_READ_VERSION:
875 read(io->asok, ((char*)(&io->d->version)) + io->amt,
876 sizeof(io->d->version) - 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->d->version))
884 io->d->version = ntohl(io->d->version);
885 if (io->d->version != 1)
887 ERROR("cconn_handle_event(name=%s) not "
888 "expecting version %d!", io->d->name, io->d->version);
890 }DEBUG("cconn_handle_event(name=%s): identified as "
891 "version %d", io->d->name, io->d->version);
894 io->request_type = ASOK_REQ_SCHEMA;
898 case CSTATE_READ_AMT:
901 read(io->asok, ((char*)(&io->json_len)) + io->amt,
902 sizeof(io->json_len) - io->amt));
903 DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
904 io->d->name, io->state, ret);
908 if (io->amt >= sizeof(io->json_len))
910 io->json_len = ntohl(io->json_len);
912 io->state = CSTATE_READ_JSON;
913 io->json = calloc(1, io->json_len + 1);
919 case CSTATE_READ_JSON:
922 read(io->asok, io->json + io->amt, io->json_len - io->amt));
923 DEBUG("cconn_handle_event(name=%s,state=%d,ret=%d)",
924 io->d->name, io->state, ret);
928 if (io->amt >= io->json_len)
930 ret = cconn_process_json(io);
934 io->request_type = ASOK_REQ_NONE;
939 ERROR("cconn_handle_event(name=%s) got to illegal state on "
940 "line %d", io->d->name, __LINE__);
945 static int cconn_prepare(struct cconn *io, struct pollfd* fds)
948 if (io->request_type == ASOK_REQ_NONE)
950 /* The request has already been serviced. */
953 else if ((io->request_type == ASOK_REQ_DATA) && (io->d->dset_num == 0))
955 /* If there are no counters to report on, don't bother
962 case CSTATE_UNCONNECTED:
963 ret = cconn_connect(io);
969 fds->events = POLLOUT;
971 case CSTATE_WRITE_REQUEST:
973 fds->events = POLLOUT;
975 case CSTATE_READ_VERSION:
976 case CSTATE_READ_AMT:
977 case CSTATE_READ_JSON:
979 fds->events = POLLIN;
982 ERROR("cconn_prepare(name=%s) got to illegal state on line %d",
983 io->d->name, __LINE__);
988 /** Returns the difference between two struct timevals in milliseconds.
989 * On overflow, we return max/min int.
991 static int milli_diff(const struct timeval *t1, const struct timeval *t2)
994 int sec_diff = t1->tv_sec - t2->tv_sec;
995 int usec_diff = t1->tv_usec - t2->tv_usec;
996 ret = usec_diff / 1000;
997 ret += (sec_diff * 1000);
1000 else if (ret < INT_MIN)
1005 /** This handles the actual network I/O to talk to the Ceph daemons.
1007 static int cconn_main_loop(uint32_t request_type)
1009 int i, ret, some_unreachable = 0;
1010 struct timeval end_tv;
1011 struct cconn io_array[g_num_daemons];
1013 DEBUG("entering cconn_main_loop(request_type = %d)", request_type);
1015 /* create cconn array */
1016 memset(io_array, 0, sizeof(io_array));
1017 for (i = 0; i < g_num_daemons; ++i)
1019 io_array[i].d = g_daemons[i];
1020 io_array[i].request_type = request_type;
1021 io_array[i].state = CSTATE_UNCONNECTED;
1024 /** Calculate the time at which we should give up */
1025 gettimeofday(&end_tv, NULL);
1026 end_tv.tv_sec += CEPH_TIMEOUT_INTERVAL;
1032 struct cconn *polled_io_array[g_num_daemons];
1033 struct pollfd fds[g_num_daemons];
1034 memset(fds, 0, sizeof(fds));
1036 for (i = 0; i < g_num_daemons; ++i)
1038 struct cconn *io = io_array + i;
1039 ret = cconn_prepare(io, fds + nfds);
1042 WARNING("ERROR: cconn_prepare(name=%s,i=%d,st=%d)=%d",
1043 io->d->name, i, io->state, ret);
1045 io->request_type = ASOK_REQ_NONE;
1046 some_unreachable = 1;
1050 DEBUG("did cconn_prepare(name=%s,i=%d,st=%d)",
1051 io->d->name, i, io->state);
1052 polled_io_array[nfds++] = io_array + i;
1059 DEBUG("cconn_main_loop: no more cconn to manage.");
1062 gettimeofday(&tv, NULL);
1063 diff = milli_diff(&end_tv, &tv);
1068 WARNING("ERROR: cconn_main_loop: timed out.\n");
1071 RETRY_ON_EINTR(ret, poll(fds, nfds, diff));
1074 ERROR("poll(2) error: %d", ret);
1077 for (i = 0; i < nfds; ++i)
1079 struct cconn *io = polled_io_array[i];
1080 int revents = fds[i].revents;
1085 else if (cconn_validate_revents(io, revents))
1087 WARNING("ERROR: cconn(name=%s,i=%d,st=%d): "
1088 "revents validation error: "
1089 "revents=0x%08x", io->d->name, i, io->state, revents);
1091 io->request_type = ASOK_REQ_NONE;
1092 some_unreachable = 1;
1096 int ret = cconn_handle_event(io);
1099 WARNING("ERROR: cconn_handle_event(name=%s,"
1100 "i=%d,st=%d): error %d", io->d->name, i, io->state, ret);
1102 io->request_type = ASOK_REQ_NONE;
1103 some_unreachable = 1;
1108 done: for (i = 0; i < g_num_daemons; ++i)
1110 cconn_close(io_array + i);
1112 if (some_unreachable)
1114 DEBUG("cconn_main_loop: some Ceph daemons were unreachable.");
1118 DEBUG("cconn_main_loop: reached all Ceph daemons :)");
1123 static int ceph_read(void)
1125 return cconn_main_loop(ASOK_REQ_DATA);
1128 /******* lifecycle *******/
1129 static int ceph_init(void)
1133 ceph_daemons_print();
1135 ret = cconn_main_loop(ASOK_REQ_VERSION);
1138 for (i = 0; i < g_num_daemons; ++i)
1140 struct ceph_daemon *d = g_daemons[i];
1141 for (j = 0; j < d->dset_num; j++)
1143 ret = plugin_register_data_set(d->dset + j);
1146 ERROR("plugin_register_data_set(%s) failed!", d->name);
1150 DEBUG("plugin_register_data_set(%s): "
1151 "(d->dset)[%d]->ds_num=%d",
1152 d->name, j, d->dset[j].ds_num);
1159 static int ceph_shutdown(void)
1162 for (i = 0; i < g_num_daemons; ++i)
1164 ceph_daemon_free(g_daemons[i]);
1169 DEBUG("finished ceph_shutdown");
1173 void module_register(void)
1175 plugin_register_complex_config("ceph", ceph_config);
1176 plugin_register_init("ceph", ceph_init);
1177 plugin_register_read("ceph", ceph_read);
1178 plugin_register_shutdown("ceph", ceph_shutdown);