2 * collectd - src/write_redis.c
3 * Copyright (C) 2010-2015 Florian Forster
5 * Permission is hereby granted, free of charge, to any person obtaining a
6 * copy of this software and associated documentation files (the "Software"),
7 * to deal in the Software without restriction, including without limitation
8 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
9 * and/or sell copies of the Software, and to permit persons to whom the
10 * Software is furnished to do so, subject to the following conditions:
12 * The above copyright notice and this permission notice shall be included in
13 * all copies or substantial portions of the Software.
15 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
20 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
21 * DEALINGS IN THE SOFTWARE.
24 * Florian Forster <ff at octo.it>
30 #include "utils/common/common.h"
32 #include <hiredis/hiredis.h>
35 #ifndef REDIS_DEFAULT_PREFIX
36 #define REDIS_DEFAULT_PREFIX "collectd/"
40 char name[DATA_MAX_NAME_LEN];
44 struct timeval timeout;
54 typedef struct wr_node_s wr_node_t;
59 static int wr_write(const data_set_t *ds, /* {{{ */
60 const value_list_t *vl, user_data_t *ud) {
61 wr_node_t *node = ud->data;
64 char value[512] = {0};
71 status = FORMAT_VL(ident, sizeof(ident), vl);
74 ssnprintf(key, sizeof(key), "%s%s",
75 (node->prefix != NULL) ? node->prefix : REDIS_DEFAULT_PREFIX,
77 ssnprintf(time, sizeof(time), "%.9f", CDTIME_T_TO_DOUBLE(vl->time));
79 value_size = sizeof(value);
80 value_ptr = &value[0];
81 status = format_values(value_ptr, value_size, ds, vl, node->store_rates);
85 pthread_mutex_lock(&node->lock);
87 if (node->conn == NULL) {
89 redisConnectWithTimeout((char *)node->host, node->port, node->timeout);
90 if (node->conn == NULL) {
91 ERROR("write_redis plugin: Connecting to host \"%s\" (port %i) failed: "
93 (node->host != NULL) ? node->host : "localhost",
94 (node->port != 0) ? node->port : 6379);
95 pthread_mutex_unlock(&node->lock);
97 } else if (node->conn->err) {
99 "write_redis plugin: Connecting to host \"%s\" (port %i) failed: %s",
100 (node->host != NULL) ? node->host : "localhost",
101 (node->port != 0) ? node->port : 6379, node->conn->errstr);
102 pthread_mutex_unlock(&node->lock);
106 rr = redisCommand(node->conn, "SELECT %d", node->database);
108 WARNING("SELECT command error. database:%d message:%s", node->database,
114 rr = redisCommand(node->conn, "ZADD %s %s %s", key, time, value);
116 WARNING("ZADD command error. key:%s message:%s", key, node->conn->errstr);
120 if (node->max_set_size >= 0) {
121 rr = redisCommand(node->conn, "ZREMRANGEBYRANK %s %d %d", key, 0,
122 (-1 * node->max_set_size) - 1);
124 WARNING("ZREMRANGEBYRANK command error. key:%s message:%s", key,
130 if (node->max_set_duration > 0) {
132 * remove element, scored less than 'current-max_set_duration'
133 * '(...' indicates 'less than' in redis CLI.
135 rr = redisCommand(node->conn, "ZREMRANGEBYSCORE %s -1 (%.9f", key,
136 (CDTIME_T_TO_DOUBLE(vl->time) - node->max_set_duration));
138 WARNING("ZREMRANGEBYSCORE command error. key:%s message:%s", key,
144 /* TODO(octo): This is more overhead than necessary. Use the cache and
145 * metadata to determine if it is a new metric and call SADD only once for
148 node->conn, "SADD %svalues %s",
149 (node->prefix != NULL) ? node->prefix : REDIS_DEFAULT_PREFIX, ident);
151 WARNING("SADD command error. ident:%s message:%s", ident,
156 pthread_mutex_unlock(&node->lock);
159 } /* }}} int wr_write */
161 static void wr_config_free(void *ptr) /* {{{ */
163 wr_node_t *node = ptr;
168 if (node->conn != NULL) {
169 redisFree(node->conn);
175 } /* }}} void wr_config_free */
177 static int wr_config_node(oconfig_item_t *ci) /* {{{ */
183 node = calloc(1, sizeof(*node));
188 node->timeout.tv_sec = 1;
189 node->timeout.tv_usec = 0;
193 node->max_set_size = -1;
194 node->max_set_duration = -1;
195 node->store_rates = true;
196 pthread_mutex_init(&node->lock, /* attr = */ NULL);
198 status = cf_util_get_string_buffer(ci, node->name, sizeof(node->name));
204 for (int i = 0; i < ci->children_num; i++) {
205 oconfig_item_t *child = ci->children + i;
207 if (strcasecmp("Host", child->key) == 0)
208 status = cf_util_get_string(child, &node->host);
209 else if (strcasecmp("Port", child->key) == 0) {
210 status = cf_util_get_port_number(child);
215 } else if (strcasecmp("Timeout", child->key) == 0) {
216 status = cf_util_get_int(child, &timeout);
218 node->timeout.tv_usec = timeout * 1000;
219 node->timeout.tv_sec = node->timeout.tv_usec / 1000000L;
220 node->timeout.tv_usec %= 1000000L;
222 } else if (strcasecmp("Prefix", child->key) == 0) {
223 status = cf_util_get_string(child, &node->prefix);
224 } else if (strcasecmp("Database", child->key) == 0) {
225 status = cf_util_get_int(child, &node->database);
226 } else if (strcasecmp("MaxSetSize", child->key) == 0) {
227 status = cf_util_get_int(child, &node->max_set_size);
228 } else if (strcasecmp("MaxSetDuration", child->key) == 0) {
229 status = cf_util_get_int(child, &node->max_set_duration);
230 } else if (strcasecmp("StoreRates", child->key) == 0) {
231 status = cf_util_get_boolean(child, &node->store_rates);
233 WARNING("write_redis plugin: Ignoring unknown config option \"%s\".",
238 } /* for (i = 0; i < ci->children_num; i++) */
241 char cb_name[sizeof("write_redis/") + DATA_MAX_NAME_LEN];
243 ssnprintf(cb_name, sizeof(cb_name), "write_redis/%s", node->name);
245 status = plugin_register_write(cb_name, wr_write,
248 .free_func = wr_config_free,
253 wr_config_free(node);
256 } /* }}} int wr_config_node */
258 static int wr_config(oconfig_item_t *ci) /* {{{ */
260 for (int i = 0; i < ci->children_num; i++) {
261 oconfig_item_t *child = ci->children + i;
263 if (strcasecmp("Node", child->key) == 0)
264 wr_config_node(child);
266 WARNING("write_redis plugin: Ignoring unknown "
267 "configuration option \"%s\" at top level.",
272 } /* }}} int wr_config */
274 void module_register(void) {
275 plugin_register_complex_config("write_redis", wr_config);