+/**
+ * A set of data that we build up in memory while parsing the JSON.
+ */
+struct values_tmp
+{
+ /** ceph daemon we are processing data for*/
+ struct ceph_daemon *d;
+ /** track avgcount across counters for avgcount/sum latency pairs */
+ uint64_t avgcount;
+ /** current index of counters - used to get type of counter */
+ int index;
+ /** do we already have an avgcount for latency pair */
+ int avgcount_exists;
+ /**
+ * similar to index, but current index of latency type counters -
+ * used to get last poll data of counter
+ */
+ int latency_index;
+ /**
+ * values list - maintain across counters since
+ * host/plugin/plugin instance are always the same
+ */
+ value_list_t vlist;
+};
+
+/**
+ * A set of count/sum pairs to keep track of latency types and get difference
+ * between this poll data and last poll data.
+ */
+struct last_data
+{
+ char ds_name[DATA_MAX_NAME_LEN];
+ double last_sum;
+ uint64_t last_count;
+};
+
+/******* network I/O *******/
+enum cstate_t
+{
+ CSTATE_UNCONNECTED = 0,
+ CSTATE_WRITE_REQUEST,
+ CSTATE_READ_VERSION,
+ CSTATE_READ_AMT,
+ CSTATE_READ_JSON,
+};
+
+enum request_type_t
+{
+ ASOK_REQ_VERSION = 0,
+ ASOK_REQ_DATA = 1,
+ ASOK_REQ_SCHEMA = 2,
+ ASOK_REQ_NONE = 1000,
+};
+
+struct cconn
+{
+ /** The Ceph daemon that we're talking to */
+ struct ceph_daemon *d;
+
+ /** Request type */
+ uint32_t request_type;
+
+ /** The connection state */
+ enum cstate_t state;
+
+ /** The socket we use to talk to this daemon */
+ int asok;
+
+ /** The amount of data remaining to read / write. */
+ uint32_t amt;
+
+ /** Length of the JSON to read */
+ uint32_t json_len;
+
+ /** Buffer containing JSON data */
+ unsigned char *json;
+
+ /** Keep data important to yajl processing */
+ struct yajl_struct yajl;
+};
+
+static int ceph_cb_null(void *ctx)
+{
+ return CEPH_CB_CONTINUE;
+}
+
+static int ceph_cb_boolean(void *ctx, int bool_val)
+{
+ return CEPH_CB_CONTINUE;
+}
+
+static int
+ceph_cb_number(void *ctx, const char *number_val, yajl_len_t number_len)
+{
+ yajl_struct *yajl = (yajl_struct*)ctx;
+ char buffer[number_len+1];
+ int i, latency_type = 0, result;
+ char key[128];
+
+ memcpy(buffer, number_val, number_len);
+ buffer[sizeof(buffer) - 1] = 0;
+
+ ssnprintf(key, yajl->state[0].key_len, "%s", yajl->state[0].key);
+ for(i = 1; i < yajl->depth; i++)
+ {
+ if((i == yajl->depth-1) && ((strcmp(yajl->state[i].key,"avgcount") == 0)
+ || (strcmp(yajl->state[i].key,"sum") == 0)))
+ {
+ if(convert_special_metrics)
+ {
+ /**
+ * Special case for filestore:JournalWrBytes. For some reason,
+ * Ceph schema encodes this as a count/sum pair while all
+ * other "Bytes" data (excluding used/capacity bytes for OSD
+ * space) uses a single "Derive" type. To spare further
+ * confusion, keep this KPI as the same type of other "Bytes".
+ * Instead of keeping an "average" or "rate", use the "sum" in
+ * the pair and assign that to the derive value.
+ */
+ if((strcmp(yajl->state[i-1].key, "journal_wr_bytes") == 0) &&
+ (strcmp(yajl->state[i-2].key,"filestore") == 0) &&
+ (strcmp(yajl->state[i].key,"avgcount") == 0))
+ {
+ DEBUG("ceph plugin: Skipping avgcount for filestore.JournalWrBytes");
+ yajl->depth = (yajl->depth - 1);
+ return CEPH_CB_CONTINUE;
+ }
+ }
+ //probably a avgcount/sum pair. if not - we'll try full key later
+ latency_type = 1;
+ break;
+ }
+ strncat(key, ".", 1);
+ strncat(key, yajl->state[i].key, yajl->state[i].key_len+1);
+ }
+
+ result = yajl->handler(yajl->handler_arg, buffer, key);
+
+ if((result == RETRY_AVGCOUNT) && latency_type)
+ {
+ strncat(key, ".", 1);
+ strncat(key, yajl->state[yajl->depth-1].key,
+ yajl->state[yajl->depth-1].key_len+1);
+ result = yajl->handler(yajl->handler_arg, buffer, key);
+ }
+
+ if(result == -ENOMEM)
+ {
+ ERROR("ceph plugin: memory allocation failed");
+ return CEPH_CB_ABORT;
+ }
+
+ yajl->depth = (yajl->depth - 1);
+ return CEPH_CB_CONTINUE;
+}
+
+static int ceph_cb_string(void *ctx, const unsigned char *string_val,
+ yajl_len_t string_len)
+{
+ return CEPH_CB_CONTINUE;
+}
+
+static int ceph_cb_start_map(void *ctx)
+{
+ return CEPH_CB_CONTINUE;
+}
+
+static int
+ceph_cb_map_key(void *ctx, const unsigned char *key, yajl_len_t string_len)
+{
+ yajl_struct *yajl = (yajl_struct*)ctx;
+
+ if((yajl->depth+1) >= YAJL_MAX_DEPTH)
+ {
+ ERROR("ceph plugin: depth exceeds max, aborting.");
+ return CEPH_CB_ABORT;
+ }
+
+ char buffer[string_len+1];
+
+ memcpy(buffer, key, string_len);
+ buffer[sizeof(buffer) - 1] = 0;
+
+ snprintf(yajl->state[yajl->depth].key, sizeof(buffer), "%s", buffer);
+ yajl->state[yajl->depth].key_len = sizeof(buffer);
+ yajl->depth = (yajl->depth + 1);
+
+ return CEPH_CB_CONTINUE;
+}
+
+static int ceph_cb_end_map(void *ctx)
+{
+ yajl_struct *yajl = (yajl_struct*)ctx;
+
+ yajl->depth = (yajl->depth - 1);
+ return CEPH_CB_CONTINUE;
+}
+
+static int ceph_cb_start_array(void *ctx)
+{
+ return CEPH_CB_CONTINUE;
+}
+
+static int ceph_cb_end_array(void *ctx)
+{
+ return CEPH_CB_CONTINUE;
+}
+
+static yajl_callbacks callbacks = {
+ ceph_cb_null,
+ ceph_cb_boolean,
+ NULL,
+ NULL,
+ ceph_cb_number,
+ ceph_cb_string,
+ ceph_cb_start_map,
+ ceph_cb_map_key,
+ ceph_cb_end_map,
+ ceph_cb_start_array,
+ ceph_cb_end_array
+};
+