自学教程：C++ CDTIME_T_TO_DOUBLE函数代码示例

static int lua_cb_dispatch_values(lua_State *L) /* {{{ */{  int nargs = lua_gettop(L);  if (nargs != 1)    return luaL_error(L, "Invalid number of arguments (%d != 1)", nargs);  luaL_checktype(L, 1, LUA_TTABLE);  value_list_t *vl = luaC_tovaluelist(L, -1);  if (vl == NULL)    return luaL_error(L, "%s", "luaC_tovaluelist failed");#if COLLECT_DEBUG  char identifier[6 * DATA_MAX_NAME_LEN];  FORMAT_VL(identifier, sizeof(identifier), vl);  DEBUG("Lua plugin: collectd.dispatch_values(): Received value list /"%s/", "        "time %.3f, interval %.3f.",        identifier, CDTIME_T_TO_DOUBLE(vl->time),        CDTIME_T_TO_DOUBLE(vl->interval));#endif  plugin_dispatch_values(vl);  sfree(vl->values);  sfree(vl);  return 0;} /* }}} lua_cb_dispatch_values */

void latency_counter_reset(latency_counter_t *lc) /* {{{ */{  if (lc == NULL)    return;  cdtime_t bin_width = lc->bin_width;  cdtime_t max_bin = (lc->max - 1) / lc->bin_width;/*  If max latency is REDUCE_THRESHOLD times less than histogram's range,  then cut it in half. REDUCE_THRESHOLD must be >= 2.  Value of 4 is selected to reduce frequent changes of bin width.*/#define REDUCE_THRESHOLD 4  if ((lc->num > 0) && (lc->bin_width >= HISTOGRAM_DEFAULT_BIN_WIDTH * 2) &&      (max_bin < HISTOGRAM_NUM_BINS / REDUCE_THRESHOLD)) {    /* new bin width will be the previous power of 2 */    bin_width = bin_width / 2;    DEBUG("utils_latency: latency_counter_reset: max_latency = %.3f; "          "max_bin = %" PRIu64 "; old_bin_width = %.3f; new_bin_width = %.3f;",          CDTIME_T_TO_DOUBLE(lc->max), max_bin,          CDTIME_T_TO_DOUBLE(lc->bin_width), CDTIME_T_TO_DOUBLE(bin_width));  }  memset(lc, 0, sizeof(*lc));  /* preserve bin width */  lc->bin_width = bin_width;  lc->start_time = cdtime();} /* }}} void latency_counter_reset */

int create_putval (char *ret, size_t ret_len, /* {{{ */	const data_set_t *ds, const value_list_t *vl){	char buffer_ident[6 * DATA_MAX_NAME_LEN];	char buffer_values[1024];	int status;	status = FORMAT_VL (buffer_ident, sizeof (buffer_ident), vl);	if (status != 0)		return (status);	escape_string (buffer_ident, sizeof (buffer_ident));	status = format_values (buffer_values, sizeof (buffer_values),			ds, vl, /* store rates = */ 0);	if (status != 0)		return (status);	escape_string (buffer_values, sizeof (buffer_values));	ssnprintf (ret, ret_len,			"PUTVAL %s interval=%.3f %s",			buffer_ident,			(vl->interval > 0)			? CDTIME_T_TO_DOUBLE (vl->interval)			: CDTIME_T_TO_DOUBLE (interval_g),			buffer_values);	return (0);} /* }}} int create_putval */

static void *camqp_subscribe_thread (void *user_data) /* {{{ */{    camqp_config_t *conf = user_data;    int status;    cdtime_t interval = plugin_get_interval ();    while (subscriber_threads_running)    {        amqp_frame_t frame;        status = camqp_connect (conf);        if (status != 0)        {            struct timespec ts_interval;            ERROR ("amqp plugin: camqp_connect failed. "                    "Will sleep for %.3f seconds.",                    CDTIME_T_TO_DOUBLE (interval));            CDTIME_T_TO_TIMESPEC (interval, &ts_interval);            nanosleep (&ts_interval, /* remaining = */ NULL);            continue;        }        status = amqp_simple_wait_frame (conf->connection, &frame);        if (status < 0)        {            struct timespec ts_interval;            ERROR ("amqp plugin: amqp_simple_wait_frame failed. "                    "Will sleep for %.3f seconds.",                    CDTIME_T_TO_DOUBLE (interval));            camqp_close_connection (conf);            CDTIME_T_TO_TIMESPEC (interval, &ts_interval);            nanosleep (&ts_interval, /* remaining = */ NULL);            continue;        }        if (frame.frame_type != AMQP_FRAME_METHOD)        {            DEBUG ("amqp plugin: Unexpected frame type: %#"PRIx8,                    frame.frame_type);            continue;        }        if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)        {            DEBUG ("amqp plugin: Unexpected method id: %#"PRIx32,                    frame.payload.method.id);            continue;        }        camqp_read_header (conf);        amqp_maybe_release_buffers (conf->connection);    } /* while (subscriber_threads_running) */    camqp_config_free (conf);    pthread_exit (NULL);    return (NULL);} /* }}} void *camqp_subscribe_thread */

static int csnmp_read_host (user_data_t *ud){  host_definition_t *host;  cdtime_t time_start;  cdtime_t time_end;  int status;  int success;  int i;  host = ud->data;  if (host->interval == 0)    host->interval = plugin_get_interval ();  time_start = cdtime ();  if (host->sess_handle == NULL)    csnmp_host_open_session (host);  if (host->sess_handle == NULL)    return (-1);  success = 0;  for (i = 0; i < host->data_list_len; i++)  {    data_definition_t *data = host->data_list[i];    if (data->is_table)      status = csnmp_read_table (host, data);    else      status = csnmp_read_value (host, data);    if (status == 0)      success++;  }  time_end = cdtime ();  if ((time_end - time_start) > host->interval)  {    WARNING ("snmp plugin: Host `%s' should be queried every %.3f "        "seconds, but reading all values takes %.3f seconds.",        host->name,        CDTIME_T_TO_DOUBLE (host->interval),        CDTIME_T_TO_DOUBLE (time_end - time_start));  }  if (success == 0)    return (-1);  return (0);} /* int csnmp_read_host */

cdtime_t latency_counter_get_percentile (latency_counter_t *lc, /* {{{ */    double percent){  double percent_upper;  double percent_lower;  double p;  cdtime_t latency_lower;  cdtime_t latency_interpolated;  int sum;  size_t i;  if ((lc == NULL) || (lc->num == 0) || !((percent > 0.0) && (percent < 100.0)))    return (0);  /* Find index i so that at least "percent" events are within i+1 ms. */  percent_upper = 0.0;  percent_lower = 0.0;  sum = 0;  for (i = 0; i < HISTOGRAM_NUM_BINS; i++)  {    percent_lower = percent_upper;    sum += lc->histogram[i];    if (sum == 0)      percent_upper = 0.0;    else      percent_upper = 100.0 * ((double) sum) / ((double) lc->num);    if (percent_upper >= percent)      break;  }  if (i >= HISTOGRAM_NUM_BINS)    return (0);  assert (percent_upper >= percent);  assert (percent_lower < percent);  if (i == 0)    return (lc->bin_width);  latency_lower = ((cdtime_t) i) * lc->bin_width;  p = (percent - percent_lower) / (percent_upper - percent_lower);  latency_interpolated = latency_lower    + DOUBLE_TO_CDTIME_T (p * CDTIME_T_TO_DOUBLE (lc->bin_width));  DEBUG ("latency_counter_get_percentile: latency_interpolated = %.3f",      CDTIME_T_TO_DOUBLE (latency_interpolated));  return (latency_interpolated);} /* }}} cdtime_t latency_counter_get_percentile */

static int do_loop(void) {  cdtime_t interval = cf_get_default_interval();  cdtime_t wait_until = cdtime() + interval;  while (loop == 0) {#if HAVE_LIBKSTAT    update_kstat();#endif    /* Issue all plugins */    plugin_read_all();    cdtime_t now = cdtime();    if (now >= wait_until) {      WARNING("Not sleeping because the next interval is "              "%.3f seconds in the past!",              CDTIME_T_TO_DOUBLE(now - wait_until));      wait_until = now + interval;      continue;    }    struct timespec ts_wait = CDTIME_T_TO_TIMESPEC(wait_until - now);    wait_until = wait_until + interval;    while ((loop == 0) && (nanosleep(&ts_wait, &ts_wait) != 0)) {      if (errno != EINTR) {        ERROR("nanosleep failed: %s", STRERRNO);        return -1;      }    }  } /* while (loop == 0) */  return 0;} /* int do_loop */

int tail_match_add_match(cu_tail_match_t *obj, cu_match_t *match,                         int (*submit_match)(cu_match_t *match,                                             void *user_data),                         void *user_data,                         void (*free_user_data)(void *user_data)) {  cu_tail_match_match_t *temp;  temp = realloc(obj->matches,                 sizeof(cu_tail_match_match_t) * (obj->matches_num + 1));  if (temp == NULL)    return (-1);  obj->matches = temp;  obj->matches_num++;  DEBUG("tail_match_add_match interval %lf",        CDTIME_T_TO_DOUBLE(((cu_tail_match_simple_t *)user_data)->interval));  temp = obj->matches + (obj->matches_num - 1);  temp->match = match;  temp->user_data = user_data;  temp->submit = submit_match;  temp->free = free_user_data;  return (0);} /* int tail_match_add_match */

static int we_flush_nolock (cdtime_t timeout, we_callback_t *cb) {  int status;  DEBUG("write_extremon plugin: we_flush_nolock: timeout = %.3f; "        "send_buffer_fill = %zu;", CDTIME_T_TO_DOUBLE (timeout),                                  cb->send_buffer_fill);  if(timeout>0)  {    cdtime_t now;    now = cdtime ();    if ((cb->send_buffer_init_time + timeout) > now)      return (0);  }/*  if (cb->send_buffer_fill <= 0)# {#   cb->send_buffer_init_time = cdtime ();#   return (0);# } */  status = we_send_buffer (cb);  we_reset_buffer (cb);  return (status);} 

double latency_counter_get_rate(const latency_counter_t *lc, /* {{{ */                                cdtime_t lower, cdtime_t upper,                                const cdtime_t now) {  if ((lc == NULL) || (lc->num == 0))    return NAN;  if (upper && (upper < lower))    return NAN;  if (lower == upper)    return 0;  /* Buckets have an exclusive lower bound and an inclusive upper bound. That   * means that the first bucket, index 0, represents (0-bin_width]. That means   * that latency==bin_width needs to result in bin=0, that's why we need to   * subtract one before dividing by bin_width. */  cdtime_t lower_bin = 0;  if (lower)    /* lower is *exclusive* => determine bucket for lower+1 */    lower_bin = ((lower + 1) - 1) / lc->bin_width;  /* lower is greater than the longest latency observed => rate is zero. */  if (lower_bin >= HISTOGRAM_NUM_BINS)    return 0;  cdtime_t upper_bin = HISTOGRAM_NUM_BINS - 1;  if (upper)    upper_bin = (upper - 1) / lc->bin_width;  if (upper_bin >= HISTOGRAM_NUM_BINS) {    upper_bin = HISTOGRAM_NUM_BINS - 1;    upper = 0;  }  double sum = 0;  for (size_t i = lower_bin; i <= upper_bin; i++)    sum += lc->histogram[i];  if (lower) {    /* Approximate ratio of requests in lower_bin, that fall between     * lower_bin_boundary and lower. This ratio is then subtracted from sum to     * increase accuracy. */    cdtime_t lower_bin_boundary = lower_bin * lc->bin_width;    assert(lower >= lower_bin_boundary);    double lower_ratio =        (double)(lower - lower_bin_boundary) / ((double)lc->bin_width);    sum -= lower_ratio * lc->histogram[lower_bin];  }  if (upper) {    /* As above: approximate ratio of requests in upper_bin, that fall between     * upper and upper_bin_boundary. */    cdtime_t upper_bin_boundary = (upper_bin + 1) * lc->bin_width;    assert(upper <= upper_bin_boundary);    double ratio = (double)(upper_bin_boundary - upper) / (double)lc->bin_width;    sum -= ratio * lc->histogram[upper_bin];  }  return sum / (CDTIME_T_TO_DOUBLE(now - lc->start_time));} /* }}} double latency_counter_get_rate */

static counter_t disk_calc_time_incr (counter_t delta_time, counter_t delta_ops){	double interval = CDTIME_T_TO_DOUBLE (plugin_get_interval ());	double avg_time = ((double) delta_time) / ((double) delta_ops);	double avg_time_incr = interval * avg_time;	return ((counter_t) (avg_time_incr + .5));}

cdtime_t latency_counter_get_average (latency_counter_t *lc) /* {{{ */{  double average;  if ((lc == NULL) || (lc->num == 0))    return (0);  average = CDTIME_T_TO_DOUBLE (lc->sum) / ((double) lc->num);  return (DOUBLE_TO_CDTIME_T (average));} /* }}} cdtime_t latency_counter_get_average */

static int ts_invoke_absolute(const data_set_t *ds, value_list_t *vl, /* {{{ */                              ts_data_t *data, int dsrc_index) {  uint64_t curr_absolute;  double rate;  int status;  /* Required meta data */  double int_fraction;  char key_int_fraction[128];  curr_absolute = (uint64_t)vl->values[dsrc_index].absolute;  snprintf(key_int_fraction, sizeof(key_int_fraction),           "target_scale[%p,%i]:int_fraction", (void *)data, dsrc_index);  int_fraction = 0.0;  /* Query the meta data */  status = uc_meta_data_get_double(vl, key_int_fraction, &int_fraction);  if (status != 0)    int_fraction = 0.0;  rate = ((double)curr_absolute) / CDTIME_T_TO_DOUBLE(vl->interval);  /* Modify the rate. */  if (!isnan(data->factor))    rate *= data->factor;  if (!isnan(data->offset))    rate += data->offset;  /* Calculate the new absolute. */  int_fraction += (rate * CDTIME_T_TO_DOUBLE(vl->interval));  curr_absolute = (uint64_t)int_fraction;  int_fraction -= ((double)curr_absolute);  vl->values[dsrc_index].absolute = (absolute_t)curr_absolute;  /* Update to the new absolute value */  uc_meta_data_add_double(vl, key_int_fraction, int_fraction);  return 0;} /* }}} int ts_invoke_absolute */

/** Histogram represents the distribution of data, it has a list of "bins".* Each bin represents an interval and has a count (frequency) of* number of values fall within its interval.** Histogram's range is determined by the number of bins and the bin width,* There are 1000 bins and all bins have the same width of default 1 millisecond.* When a value above this range is added, Histogram's range is increased by* increasing the bin width (note that number of bins remains always at 1000).* This operation of increasing bin width is little expensive as each bin need* to be visited to update its count. To reduce frequent change of bin width,* new bin width will be the next nearest power of 2. Example: 2, 4, 8, 16, 32,* 64, 128, 256, 512, 1024, 2048, 5086, ...** So, if the required bin width is 300, then new bin width will be 512 as it is* the next nearest power of 2.*/static void change_bin_width(latency_counter_t *lc, cdtime_t latency) /* {{{ */{  /* This function is called because the new value is above histogram's range.   * First find the required bin width:   *           requiredBinWidth = (value + 1) / numBins   * then get the next nearest power of 2   *           newBinWidth = 2^(ceil(log2(requiredBinWidth)))   */  double required_bin_width =      ((double)(latency + 1)) / ((double)HISTOGRAM_NUM_BINS);  double required_bin_width_logbase2 = log(required_bin_width) / log(2.0);  cdtime_t new_bin_width =      (cdtime_t)(pow(2.0, ceil(required_bin_width_logbase2)) + .5);  cdtime_t old_bin_width = lc->bin_width;  lc->bin_width = new_bin_width;  /* bin_width has been increased, now iterate through all bins and move the   * old bin's count to new bin. */  if (lc->num > 0) // if the histogram has data then iterate else skip  {    double width_change_ratio =        ((double)old_bin_width) / ((double)new_bin_width);    for (size_t i = 0; i < HISTOGRAM_NUM_BINS; i++) {      size_t new_bin = (size_t)(((double)i) * width_change_ratio);      if (i == new_bin)        continue;      assert(new_bin < i);      lc->histogram[new_bin] += lc->histogram[i];      lc->histogram[i] = 0;    }  }  DEBUG("utils_latency: change_bin_width: latency = %.3f; "        "old_bin_width = %.3f; new_bin_width = %.3f;",        CDTIME_T_TO_DOUBLE(latency), CDTIME_T_TO_DOUBLE(old_bin_width),        CDTIME_T_TO_DOUBLE(new_bin_width));} /* }}} void change_bin_width */

/** * Read configuration. */static int cgps_config (oconfig_item_t *ci){  int i;  for (i = 0; i < ci->children_num; i++)  {    oconfig_item_t *child = ci->children + i;    if (strcasecmp ("Host", child->key) == 0)      cf_util_get_string (child, &cgps_config_data.host);    else if (strcasecmp ("Port", child->key) == 0)      cf_util_get_service (child, &cgps_config_data.port);    else if (strcasecmp ("Timeout", child->key) == 0)      cf_util_get_cdtime (child, &cgps_config_data.timeout);    else if (strcasecmp ("PauseConnect", child->key) == 0)      cf_util_get_cdtime (child, &cgps_config_data.pause_connect);    else      WARNING ("gps plugin: Ignoring unknown config option /"%s/".", child->key);  }  // Controlling the value for timeout:  // If set too high it blocks the reading (> 5 s), too low it gets not reading (< 500 us).  // To avoid any issues we replace "out of range" value by the default value.  if (    cgps_config_data.timeout > TIME_T_TO_CDTIME_T(5)    ||    cgps_config_data.timeout < US_TO_CDTIME_T(500)  )   {    WARNING ("gps plugin: timeout set to %.6f sec. setting to default (%.6f).",       CDTIME_T_TO_DOUBLE(cgps_config_data.timeout),      CDTIME_T_TO_DOUBLE(CGPS_DEFAULT_TIMEOUT)    );    cgps_config_data.timeout = CGPS_DEFAULT_TIMEOUT;  }   return (0);}

int handle_listval (FILE *fh, char *buffer){  char *command;  char **names = NULL;  cdtime_t *times = NULL;  size_t number = 0;  size_t i;  int status;  DEBUG ("utils_cmd_listval: handle_listval (fh = %p, buffer = %s);",      (void *) fh, buffer);  command = NULL;  status = parse_string (&buffer, &command);  if (status != 0)  {    print_to_socket (fh, "-1 Cannot parse command./n");    free_everything_and_return (-1);  }  assert (command != NULL);  if (strcasecmp ("LISTVAL", command) != 0)  {    print_to_socket (fh, "-1 Unexpected command: `%s'./n", command);    free_everything_and_return (-1);  }  if (*buffer != 0)  {    print_to_socket (fh, "-1 Garbage after end of command: %s/n", buffer);    free_everything_and_return (-1);  }  status = uc_get_names (&names, &times, &number);  if (status != 0)  {    DEBUG ("command listval: uc_get_names failed with status %i", status);    print_to_socket (fh, "-1 uc_get_names failed./n");    free_everything_and_return (-1);  }  print_to_socket (fh, "%i Value%s found/n",      (int) number, (number == 1) ? "" : "s");  for (i = 0; i < number; i++)    print_to_socket (fh, "%.3f %s/n", CDTIME_T_TO_DOUBLE (times[i]),       	names[i]);  free_everything_and_return (0);} /* int handle_listval */

/** * Init. */static int cgps_init (void){  int status;  if (cgps_thread_running == CGPS_TRUE)  {    DEBUG ("gps plugin: error gps thread already running ... ");    return 0;  }  DEBUG ("gps plugin: config{host: /"%s/", port: /"%s/", timeout: %.6f sec., pause connect: %.3f sec.}",         cgps_config_data.host, cgps_config_data.port,         CDTIME_T_TO_DOUBLE (cgps_config_data.timeout),         CDTIME_T_TO_DOUBLE (cgps_config_data.pause_connect));  status = plugin_thread_create (&cgps_thread_id, NULL, cgps_thread, NULL);  if (status != 0)  {    ERROR ("gps plugin: pthread_create() failed.");    return (-1);  }  return (0);}

static void cc_submit_response_time (const web_page_t *wp, /* {{{ */    cdtime_t response_time){  value_t values[1];  value_list_t vl = VALUE_LIST_INIT;  values[0].gauge = CDTIME_T_TO_DOUBLE (response_time);  vl.values = values;  vl.values_len = 1;  sstrncpy (vl.host, hostname_g, sizeof (vl.host));  sstrncpy (vl.plugin, "curl", sizeof (vl.plugin));  sstrncpy (vl.plugin_instance, wp->instance, sizeof (vl.plugin_instance));  sstrncpy (vl.type, "response_time", sizeof (vl.type));  plugin_dispatch_values (&vl);} /* }}} void cc_submit_response_time */

static PGresult *c_psql_exec_query_params (c_psql_database_t *db,		udb_query_t *q, c_psql_user_data_t *data){	char *params[db->max_params_num];	char  interval[64];	int   i;	if ((data == NULL) || (data->params_num == 0))		return (c_psql_exec_query_noparams (db, q));	assert (db->max_params_num >= data->params_num);	for (i = 0; i < data->params_num; ++i) {		switch (data->params[i]) {			case C_PSQL_PARAM_HOST:				params[i] = C_PSQL_IS_UNIX_DOMAIN_SOCKET (db->host)					? "localhost" : db->host;				break;			case C_PSQL_PARAM_DB:				params[i] = db->database;				break;			case C_PSQL_PARAM_USER:				params[i] = db->user;				break;			case C_PSQL_PARAM_INTERVAL:				ssnprintf (interval, sizeof (interval), "%.3f",						(db->interval > 0)						? CDTIME_T_TO_DOUBLE (db->interval)						: plugin_get_interval ());				params[i] = interval;				break;			case C_PSQL_PARAM_INSTANCE:				params[i] = db->instance;				break;			default:				assert (0);		}	}	return PQexecParams (db->conn, udb_query_get_statement (q),			data->params_num, NULL,			(const char *const *) params,			NULL, NULL, /* return text data */ 0);} /* c_psql_exec_query_params */

/* wg_force_reconnect_check closes cb->sock_fd when it was open for longer * than cb->reconnect_interval. Must hold cb->send_lock when calling. */static void wg_force_reconnect_check (struct wg_callback *cb){    cdtime_t now;    if (cb->reconnect_interval == 0)        return;    /* check if address changes if addr_timeout */    now = cdtime ();    if ((now - cb->last_reconnect_time) < cb->reconnect_interval)        return;    /* here we should close connection on next */    close (cb->sock_fd);    cb->sock_fd = -1;    cb->last_reconnect_time = now;    cb->reconnect_interval_reached = 1;    INFO ("write_graphite plugin: Connection closed after %.3f seconds.",          CDTIME_T_TO_DOUBLE (now - cb->last_reconnect_time));}

static int init_global_variables(void) {  interval_g = cf_get_default_interval();  assert(interval_g > 0);  DEBUG("interval_g = %.3f;", CDTIME_T_TO_DOUBLE(interval_g));  const char *str = global_option_get("Timeout");  if (str == NULL)    str = "2";  timeout_g = atoi(str);  if (timeout_g <= 1) {    fprintf(stderr, "Cannot set the timeout to a correct value./n"                    "Please check your settings./n");    return -1;  }  DEBUG("timeout_g = %i;", timeout_g);  if (init_hostname() != 0)    return -1;  DEBUG("hostname_g = %s;", hostname_g);  return 0;} /* int init_global_variables */

static int apcups_config (oconfig_item_t *ci){	int i;	_Bool persistent_conn_set = 0;	for (i = 0; i < ci->children_num; i++)	{		oconfig_item_t *child = ci->children + i;		if (strcasecmp (child->key, "Host") == 0)			cf_util_get_string (child, &conf_node);		else if (strcasecmp (child->key, "Port") == 0)			cf_util_get_service (child, &conf_service);		else if (strcasecmp (child->key, "ReportSeconds") == 0)			cf_util_get_boolean (child, &conf_report_seconds);		else if (strcasecmp (child->key, "PersistentConnection") == 0) {			cf_util_get_boolean (child, &conf_persistent_conn);			persistent_conn_set = 1;		}		else			ERROR ("apcups plugin: Unknown config option /"%s/".", child->key);	}	if (!persistent_conn_set) {		double interval = CDTIME_T_TO_DOUBLE(plugin_get_interval());		if (interval > APCUPS_SERVER_TIMEOUT) {			NOTICE ("apcups plugin: Plugin poll interval set to %.3f seconds. "				"Apcupsd NIS socket timeout is %.3f seconds, "				"PersistentConnection disabled by default.",				interval, APCUPS_SERVER_TIMEOUT);			conf_persistent_conn = 0;		}	}	return (0);} /* int apcups_config */

static riemann_event_t *wrr_value_to_event(struct riemann_host const *host, /* {{{ */                   data_set_t const *ds, value_list_t const *vl, size_t index,                   gauge_t const *rates, int status) {  riemann_event_t *event;  char name_buffer[5 * DATA_MAX_NAME_LEN];  char service_buffer[6 * DATA_MAX_NAME_LEN];  size_t i;  event = riemann_event_new();  if (event == NULL) {    ERROR("write_riemann plugin: riemann_event_new() failed.");    return NULL;  }  format_name(name_buffer, sizeof(name_buffer),              /* host = */ "", vl->plugin, vl->plugin_instance, vl->type,              vl->type_instance);  if (host->always_append_ds || (ds->ds_num > 1)) {    if (host->event_service_prefix == NULL)      snprintf(service_buffer, sizeof(service_buffer), "%s/%s", &name_buffer[1],               ds->ds[index].name);    else      snprintf(service_buffer, sizeof(service_buffer), "%s%s/%s",               host->event_service_prefix, &name_buffer[1], ds->ds[index].name);  } else {    if (host->event_service_prefix == NULL)      sstrncpy(service_buffer, &name_buffer[1], sizeof(service_buffer));    else      snprintf(service_buffer, sizeof(service_buffer), "%s%s",               host->event_service_prefix, &name_buffer[1]);  }  riemann_event_set(      event, RIEMANN_EVENT_FIELD_HOST, vl->host, RIEMANN_EVENT_FIELD_TIME,      (int64_t)CDTIME_T_TO_TIME_T(vl->time), RIEMANN_EVENT_FIELD_TTL,      (float)CDTIME_T_TO_DOUBLE(vl->interval) * host->ttl_factor,      RIEMANN_EVENT_FIELD_STRING_ATTRIBUTES, "plugin", vl->plugin, "type",      vl->type, "ds_name", ds->ds[index].name, NULL,      RIEMANN_EVENT_FIELD_SERVICE, service_buffer, RIEMANN_EVENT_FIELD_NONE);#if RCC_VERSION_NUMBER >= 0x010A00  riemann_event_set(event, RIEMANN_EVENT_FIELD_TIME_MICROS,                    (int64_t)CDTIME_T_TO_US(vl->time));#endif  if (host->check_thresholds) {    const char *state = NULL;    switch (status) {    case STATE_OKAY:      state = "ok";      break;    case STATE_ERROR:      state = "critical";      break;    case STATE_WARNING:      state = "warning";      break;    case STATE_MISSING:      state = "unknown";      break;    }    if (state)      riemann_event_set(event, RIEMANN_EVENT_FIELD_STATE, state,                        RIEMANN_EVENT_FIELD_NONE);  }  if (vl->plugin_instance[0] != 0)    riemann_event_string_attribute_add(event, "plugin_instance",                                       vl->plugin_instance);  if (vl->type_instance[0] != 0)    riemann_event_string_attribute_add(event, "type_instance",                                       vl->type_instance);  if ((ds->ds[index].type != DS_TYPE_GAUGE) && (rates != NULL)) {    char ds_type[DATA_MAX_NAME_LEN];    snprintf(ds_type, sizeof(ds_type), "%s:rate",             DS_TYPE_TO_STRING(ds->ds[index].type));    riemann_event_string_attribute_add(event, "ds_type", ds_type);  } else {    riemann_event_string_attribute_add(event, "ds_type",                                       DS_TYPE_TO_STRING(ds->ds[index].type));  }  {    char ds_index[DATA_MAX_NAME_LEN];    snprintf(ds_index, sizeof(ds_index), "%" PRIsz, index);    riemann_event_string_attribute_add(event, "ds_index", ds_index);  }  for (i = 0; i < riemann_attrs_num; i += 2)    riemann_event_string_attribute_add(event, riemann_attrs[i],                                       riemann_attrs[i + 1]);  for (i = 0; i < riemann_tags_num; i++)    riemann_event_tag_add(event, riemann_tags[i]);//.........这里部分代码省略.........

static Event *riemann_value_to_protobuf (struct riemann_host const *host, /* {{{ */        data_set_t const *ds,        value_list_t const *vl, size_t index,        gauge_t const *rates){    Event *event;    char name_buffer[5 * DATA_MAX_NAME_LEN];    char service_buffer[6 * DATA_MAX_NAME_LEN];    double ttl;    int i;    event = malloc (sizeof (*event));    if (event == NULL)    {        ERROR ("write_riemann plugin: malloc failed.");        return (NULL);    }    memset (event, 0, sizeof (*event));    event__init (event);    event->host = strdup (vl->host);    event->time = CDTIME_T_TO_TIME_T (vl->time);    event->has_time = 1;    ttl = CDTIME_T_TO_DOUBLE (vl->interval) * host->ttl_factor;    event->ttl = (float) ttl;    event->has_ttl = 1;    riemann_event_add_attribute (event, "plugin", vl->plugin);    if (vl->plugin_instance[0] != 0)        riemann_event_add_attribute (event, "plugin_instance",                                     vl->plugin_instance);    riemann_event_add_attribute (event, "type", vl->type);    if (vl->type_instance[0] != 0)        riemann_event_add_attribute (event, "type_instance",                                     vl->type_instance);    if ((ds->ds[index].type != DS_TYPE_GAUGE) && (rates != NULL))    {        char ds_type[DATA_MAX_NAME_LEN];        ssnprintf (ds_type, sizeof (ds_type), "%s:rate",                   DS_TYPE_TO_STRING(ds->ds[index].type));        riemann_event_add_attribute (event, "ds_type", ds_type);    }    else    {        riemann_event_add_attribute (event, "ds_type",                                     DS_TYPE_TO_STRING(ds->ds[index].type));    }    riemann_event_add_attribute (event, "ds_name", ds->ds[index].name);    {        char ds_index[DATA_MAX_NAME_LEN];        ssnprintf (ds_index, sizeof (ds_index), "%zu", index);        riemann_event_add_attribute (event, "ds_index", ds_index);    }    for (i = 0; i < riemann_attrs_num; i += 2)        riemann_event_add_attribute(event,                                    riemann_attrs[i],                                    riemann_attrs[i +1]);    for (i = 0; i < riemann_tags_num; i++)        riemann_event_add_tag (event, riemann_tags[i]);    if (ds->ds[index].type == DS_TYPE_GAUGE)    {        event->has_metric_d = 1;        event->metric_d = (double) vl->values[index].gauge;    }    else if (rates != NULL)    {        event->has_metric_d = 1;        event->metric_d = (double) rates[index];    }    else    {        event->has_metric_sint64 = 1;        if (ds->ds[index].type == DS_TYPE_DERIVE)            event->metric_sint64 = (int64_t) vl->values[index].derive;        else if (ds->ds[index].type == DS_TYPE_ABSOLUTE)            event->metric_sint64 = (int64_t) vl->values[index].absolute;        else            event->metric_sint64 = (int64_t) vl->values[index].counter;    }    format_name (name_buffer, sizeof (name_buffer),                 /* host = */ "", vl->plugin, vl->plugin_instance,                 vl->type, vl->type_instance);    if (host->always_append_ds || (ds->ds_num > 1))        ssnprintf (service_buffer, sizeof (service_buffer),                   "%s/%s", &name_buffer[1], ds->ds[index].name);    else        sstrncpy (service_buffer, &name_buffer[1],                  sizeof (service_buffer));    event->service = strdup (service_buffer);//.........这里部分代码省略.........

int uc_update (const data_set_t *ds, const value_list_t *vl){    char name[6 * DATA_MAX_NAME_LEN];    cache_entry_t *ce = NULL;    int status;    size_t i;    if (FORMAT_VL (name, sizeof (name), vl) != 0)    {        ERROR ("uc_update: FORMAT_VL failed.");        return (-1);    }    pthread_mutex_lock (&cache_lock);    status = c_avl_get (cache_tree, name, (void *) &ce);    if (status != 0) /* entry does not yet exist */    {        status = uc_insert (ds, vl, name);        pthread_mutex_unlock (&cache_lock);        return (status);    }    assert (ce != NULL);    assert (ce->values_num == ds->ds_num);    if (ce->last_time >= vl->time)    {        pthread_mutex_unlock (&cache_lock);        NOTICE ("uc_update: Value too old: name = %s; value time = %.3f; "                "last cache update = %.3f;",                name,                CDTIME_T_TO_DOUBLE (vl->time),                CDTIME_T_TO_DOUBLE (ce->last_time));        return (-1);    }    for (i = 0; i < ds->ds_num; i++)    {        switch (ds->ds[i].type)        {        case DS_TYPE_COUNTER:        {            counter_t diff = counter_diff (ce->values_raw[i].counter, vl->values[i].counter);            ce->values_gauge[i] = ((double) diff)                                  / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time));            ce->values_raw[i].counter = vl->values[i].counter;        }        break;        case DS_TYPE_GAUGE:            ce->values_raw[i].gauge = vl->values[i].gauge;            ce->values_gauge[i] = vl->values[i].gauge;            break;        case DS_TYPE_DERIVE:        {            derive_t diff = vl->values[i].derive - ce->values_raw[i].derive;            ce->values_gauge[i] = ((double) diff)                                  / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time));            ce->values_raw[i].derive = vl->values[i].derive;        }        break;        case DS_TYPE_ABSOLUTE:            ce->values_gauge[i] = ((double) vl->values[i].absolute)                                  / (CDTIME_T_TO_DOUBLE (vl->time - ce->last_time));            ce->values_raw[i].absolute = vl->values[i].absolute;            break;        default:            /* This shouldn't happen. */            pthread_mutex_unlock (&cache_lock);            ERROR ("uc_update: Don't know how to handle data source type %i.",                   ds->ds[i].type);            return (-1);        } /* switch (ds->ds[i].type) */        DEBUG ("uc_update: %s: ds[%zu] = %lf", name, i, ce->values_gauge[i]);    } /* for (i) */    /* Update the history if it exists. */    if (ce->history != NULL)    {        assert (ce->history_index < ce->history_length);        for (i = 0; i < ce->values_num; i++)        {            size_t hist_idx = (ce->values_num * ce->history_index) + i;            ce->history[hist_idx] = ce->values_gauge[i];        }        assert (ce->history_length > 0);        ce->history_index = (ce->history_index + 1) % ce->history_length;    }    /* Prune invalid gauge data */    uc_check_range (ds, ce);    ce->last_time = vl->time;//.........这里部分代码省略.........

static int wh_flush_nolock (cdtime_t timeout, wh_callback_t *cb) /* {{{ */{        int status;        DEBUG ("write_http plugin: wh_flush_nolock: timeout = %.3f; "                        "send_buffer_fill = %zu;",                        CDTIME_T_TO_DOUBLE (timeout),                        cb->send_buffer_fill);        /* timeout == 0  => flush unconditionally */        if (timeout > 0)        {                cdtime_t now;                now = cdtime ();                if ((cb->send_buffer_init_time + timeout) > now)                        return (0);        }        if (cb->format == WH_FORMAT_COMMAND)        {                if (cb->send_buffer_fill == 0)                {                        cb->send_buffer_init_time = cdtime ();                        return (0);                }                status = wh_send_buffer (cb);                wh_reset_buffer (cb);        }        else if (cb->format == WH_FORMAT_JSON)        {                if (cb->send_buffer_fill <= 2)                {                        cb->send_buffer_init_time = cdtime ();                        return (0);                }                status = format_json_finalize (cb->send_buffer,                                &cb->send_buffer_fill,                                &cb->send_buffer_free);                if (status != 0)                {                        ERROR ("write_http: wh_flush_nolock: "                                        "format_json_finalize failed.");                        wh_reset_buffer (cb);                        return (status);                }                status = wh_send_buffer (cb);                wh_reset_buffer (cb);        }        else        {                ERROR ("write_http: wh_flush_nolock: "                                "Unknown format: %i",                                cb->format);                return (-1);        }        return (status);} /* }}} wh_flush_nolock */

/* Must hold metrics_lock when calling this function. */static int statsd_metric_submit_unsafe (statsd_config_t *conf, char const *name, /* {{{ */    statsd_metric_t const *metric){  value_t values[1];  value_list_t vl = VALUE_LIST_INIT;  char *global_prefix = NULL;  char *type_prefix = NULL;  char *global_postfix = NULL;  char full_name[DATA_MAX_NAME_LEN] = {0};  DEBUG("statsd plugin: submit metric");  vl.values = values;  vl.values_len = 1;  sstrncpy (vl.host, hostname_g, sizeof (vl.host));  sstrncpy (vl.plugin, "statsd", sizeof (vl.plugin));  sstrncpy (vl.plugin_instance, conf->node_name, sizeof (vl.plugin_instance));  global_prefix = (NULL == conf->global_prefix) ? "" : conf->global_prefix;  global_postfix = (NULL == conf->global_postfix) ? "" : conf->global_postfix;  switch (metric->type) {  case STATSD_GAUGE:    sstrncpy (vl.type, "gauge", sizeof (vl.type));    type_prefix = (NULL == conf->gauge_prefix) ? "" : conf->gauge_prefix;    break;  case STATSD_TIMER:    sstrncpy (vl.type, "latency", sizeof (vl.type));    type_prefix = (NULL == conf->timer_prefix) ? "" : conf->timer_prefix;    break;  case STATSD_SET:    sstrncpy (vl.type, "objects", sizeof (vl.type));    type_prefix = (NULL == conf->set_prefix) ? "" : conf->set_prefix;    break;  case STATSD_COUNTER:    sstrncpy (vl.type, "derive", sizeof (vl.type));    type_prefix = (NULL == conf->counter_prefix) ? "" : conf->counter_prefix;    break;  default:    ERROR("statsd plugin: unknow metrics type %d", metric->type);  }  ssnprintf(full_name, sizeof(full_name), "%s%s%s%s", global_prefix, type_prefix, name, global_postfix);  DEBUG("statsd plugin: metric name %s", full_name);  sstrncpy (vl.type_instance, full_name, sizeof (vl.type_instance));  if (metric->type == STATSD_GAUGE)    values[0].gauge = (gauge_t) metric->value;  else if (metric->type == STATSD_TIMER)  {    size_t i;    _Bool have_events = (metric->updates_num > 0);    /* Make sure all timer metrics share the *same* timestamp. */    vl.time = cdtime ();    if (!conf->leave_metrics_name_asis)      ssnprintf (vl.type_instance, sizeof (vl.type_instance),                 "%s-average", full_name);    values[0].gauge = have_events      ? CDTIME_T_TO_DOUBLE (latency_counter_get_average (metric->latency))      : NAN;    plugin_dispatch_values (&vl);    if (conf->timer_lower) {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-lower", full_name);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_min (metric->latency))        : NAN;      plugin_dispatch_values (&vl);    }    if (conf->timer_upper) {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-upper", full_name);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_max (metric->latency))        : NAN;      plugin_dispatch_values (&vl);    }    if (conf->timer_sum) {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-sum", full_name);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_sum (metric->latency))        : NAN;      plugin_dispatch_values (&vl);    }    for (i = 0; i < conf->timer_percentile_num; i++)    {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-percentile-%.0f", full_name, conf->timer_percentile[i]);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_percentile (metric->latency, conf->timer_percentile[i]))        : NAN;      plugin_dispatch_values (&vl);//.........这里部分代码省略.........

/* Must hold metrics_lock when calling this function. */static int statsd_metric_submit_unsafe (char const *name, /* {{{ */    statsd_metric_t const *metric){  value_t values[1];  value_list_t vl = VALUE_LIST_INIT;  vl.values = values;  vl.values_len = 1;  sstrncpy (vl.host, hostname_g, sizeof (vl.host));  sstrncpy (vl.plugin, "statsd", sizeof (vl.plugin));  if (metric->type == STATSD_GAUGE)    sstrncpy (vl.type, "gauge", sizeof (vl.type));  else if (metric->type == STATSD_TIMER)    sstrncpy (vl.type, "latency", sizeof (vl.type));  else if (metric->type == STATSD_SET)    sstrncpy (vl.type, "objects", sizeof (vl.type));  else /* if (metric->type == STATSD_COUNTER) */    sstrncpy (vl.type, "derive", sizeof (vl.type));  sstrncpy (vl.type_instance, name, sizeof (vl.type_instance));  if (metric->type == STATSD_GAUGE)    values[0].gauge = (gauge_t) metric->value;  else if (metric->type == STATSD_TIMER)  {    size_t i;    _Bool have_events = (metric->updates_num > 0);    /* Make sure all timer metrics share the *same* timestamp. */    vl.time = cdtime ();    ssnprintf (vl.type_instance, sizeof (vl.type_instance),        "%s-average", name);    values[0].gauge = have_events      ? CDTIME_T_TO_DOUBLE (latency_counter_get_average (metric->latency))      : NAN;    plugin_dispatch_values (&vl);    if (conf_timer_lower) {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-lower", name);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_min (metric->latency))        : NAN;      plugin_dispatch_values (&vl);    }    if (conf_timer_upper) {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-upper", name);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_max (metric->latency))        : NAN;      plugin_dispatch_values (&vl);    }    if (conf_timer_sum) {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-sum", name);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_sum (metric->latency))        : NAN;      plugin_dispatch_values (&vl);    }    for (i = 0; i < conf_timer_percentile_num; i++)    {      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-percentile-%.0f", name, conf_timer_percentile[i]);      values[0].gauge = have_events        ? CDTIME_T_TO_DOUBLE (latency_counter_get_percentile (metric->latency, conf_timer_percentile[i]))        : NAN;      plugin_dispatch_values (&vl);    }    /* Keep this at the end, since vl.type is set to "gauge" here. The     * vl.type's above are implicitly set to "latency". */    if (conf_timer_count) {      sstrncpy (vl.type, "gauge", sizeof (vl.type));      ssnprintf (vl.type_instance, sizeof (vl.type_instance),          "%s-count", name);      values[0].gauge = latency_counter_get_num (metric->latency);      plugin_dispatch_values (&vl);    }    latency_counter_reset (metric->latency);    return (0);  }  else if (metric->type == STATSD_SET)  {    if (metric->set == NULL)      values[0].gauge = 0.0;    else      values[0].gauge = (gauge_t) c_avl_size (metric->set);  }  else    values[0].derive = (derive_t) metric->value;//.........这里部分代码省略.........

/* * Functions */static int wr_write (const data_set_t *ds, /* {{{ */    const value_list_t *vl,    user_data_t *ud){  wr_node_t *node = ud->data;  char ident[512];  char key[512];  char value[512];  char time[24];  size_t value_size;  char *value_ptr;  int status;  redisReply   *rr;  status = FORMAT_VL (ident, sizeof (ident), vl);  if (status != 0)    return (status);  ssnprintf (key, sizeof (key), "%s%s",      (node->prefix != NULL) ? node->prefix : REDIS_DEFAULT_PREFIX,      ident);  ssnprintf (time, sizeof (time), "%.9f", CDTIME_T_TO_DOUBLE(vl->time));  memset (value, 0, sizeof (value));  value_size = sizeof (value);  value_ptr = &value[0];  status = format_values (value_ptr, value_size, ds, vl, node->store_rates);  if (status != 0)    return (status);  pthread_mutex_lock (&node->lock);  if (node->conn == NULL)  {    node->conn = redisConnectWithTimeout ((char *)node->host, node->port, node->timeout);    if (node->conn == NULL)    {      ERROR ("write_redis plugin: Connecting to host /"%s/" (port %i) failed: Unkown reason",          (node->host != NULL) ? node->host : "localhost",          (node->port != 0) ? node->port : 6379);      pthread_mutex_unlock (&node->lock);      return (-1);    }    else if (node->conn->err)    {      ERROR ("write_redis plugin: Connecting to host /"%s/" (port %i) failed: %s",          (node->host != NULL) ? node->host : "localhost",          (node->port != 0) ? node->port : 6379,          node->conn->errstr);      pthread_mutex_unlock (&node->lock);      return (-1);    }    rr = redisCommand(node->conn, "SELECT %d", node->database);    if (rr == NULL)      WARNING("SELECT command error. database:%d message:%s", node->database, node->conn->errstr);    else      freeReplyObject (rr);  }  rr = redisCommand (node->conn, "ZADD %s %s %s", key, time, value);  if (rr == NULL)    WARNING("ZADD command error. key:%s message:%s", key, node->conn->errstr);  else    freeReplyObject (rr);  if (node->max_set_size >= 0)  {    rr = redisCommand (node->conn, "ZREMRANGEBYRANK %s %d %d", key, 0, (-1 * node->max_set_size) - 1);    if (rr == NULL)      WARNING("ZREMRANGEBYRANK command error. key:%s message:%s", key, node->conn->errstr);    else      freeReplyObject (rr);  }  /* TODO(octo): This is more overhead than necessary. Use the cache and   * metadata to determine if it is a new metric and call SADD only once for   * each metric. */  rr = redisCommand (node->conn, "SADD %svalues %s",      (node->prefix != NULL) ? node->prefix : REDIS_DEFAULT_PREFIX,      ident);  if (rr==NULL)    WARNING("SADD command error. ident:%s message:%s", ident, node->conn->errstr);  else    freeReplyObject (rr);  pthread_mutex_unlock (&node->lock);  return (0);} /* }}} int wr_write */

static int uc_insert (const data_set_t *ds, const value_list_t *vl,                      const char *key){    char *key_copy;    cache_entry_t *ce;    size_t i;    /* `cache_lock' has been locked by `uc_update' */    key_copy = strdup (key);    if (key_copy == NULL)    {        ERROR ("uc_insert: strdup failed.");        return (-1);    }    ce = cache_alloc (ds->ds_num);    if (ce == NULL)    {        sfree (key_copy);        ERROR ("uc_insert: cache_alloc (%zu) failed.", ds->ds_num);        return (-1);    }    sstrncpy (ce->name, key, sizeof (ce->name));    for (i = 0; i < ds->ds_num; i++)    {        switch (ds->ds[i].type)        {        case DS_TYPE_COUNTER:            ce->values_gauge[i] = NAN;            ce->values_raw[i].counter = vl->values[i].counter;            break;        case DS_TYPE_GAUGE:            ce->values_gauge[i] = vl->values[i].gauge;            ce->values_raw[i].gauge = vl->values[i].gauge;            break;        case DS_TYPE_DERIVE:            ce->values_gauge[i] = NAN;            ce->values_raw[i].derive = vl->values[i].derive;            break;        case DS_TYPE_ABSOLUTE:            ce->values_gauge[i] = NAN;            if (vl->interval > 0)                ce->values_gauge[i] = ((double) vl->values[i].absolute)                                      / CDTIME_T_TO_DOUBLE (vl->interval);            ce->values_raw[i].absolute = vl->values[i].absolute;            break;        default:            /* This shouldn't happen. */            ERROR ("uc_insert: Don't know how to handle data source type %i.",                   ds->ds[i].type);            sfree (key_copy);            cache_free (ce);            return (-1);        } /* switch (ds->ds[i].type) */    } /* for (i) */    /* Prune invalid gauge data */    uc_check_range (ds, ce);    ce->last_time = vl->time;    ce->last_update = cdtime ();    ce->interval = vl->interval;    ce->state = STATE_OKAY;    if (c_avl_insert (cache_tree, key_copy, ce) != 0)    {        sfree (key_copy);        ERROR ("uc_insert: c_avl_insert failed.");        return (-1);    }    DEBUG ("uc_insert: Added %s to the cache.", key);    return (0);} /* int uc_insert */