Merge branch 'collectd-5.4' into collectd-5.5
[collectd.git] / src / write_kafka.c
1 /**
2  * collectd - src/write_kafka.c
3  * Copyright (C) 2014       Pierre-Yves Ritschard
4  *
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:
11  *
12  * The above copyright notice and this permission notice shall be included in
13  * all copies or substantial portions of the Software.
14  *
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.
22  *
23  * Authors:
24  *   Pierre-Yves Ritschard <pyr at spootnik.org>
25  */
26
27 #include "collectd.h"
28 #include "plugin.h"
29 #include "common.h"
30 #include "configfile.h"
31 #include "utils_cache.h"
32 #include "utils_cmd_putval.h"
33 #include "utils_format_graphite.h"
34 #include "utils_format_json.h"
35 #include "utils_crc32.h"
36
37 #include <sys/types.h>
38 #include <librdkafka/rdkafka.h>
39 #include <pthread.h>
40 #include <zlib.h>
41 #include <errno.h>
42
43 struct kafka_topic_context {
44 #define KAFKA_FORMAT_JSON        0
45 #define KAFKA_FORMAT_COMMAND     1
46 #define KAFKA_FORMAT_GRAPHITE    2
47     u_int8_t                     format;
48     unsigned int                 graphite_flags;
49     _Bool                        store_rates;
50     rd_kafka_topic_conf_t       *conf;
51     rd_kafka_topic_t            *topic;
52     rd_kafka_conf_t             *kafka_conf;
53     rd_kafka_t                  *kafka;
54     int                          has_key;
55     u_int32_t                    key;
56     char                        *prefix;
57     char                        *postfix;
58     char                         escape_char;
59     char                        *topic_name;
60     pthread_mutex_t             lock;
61 };
62
63 static int kafka_handle(struct kafka_topic_context *);
64 static int kafka_write(const data_set_t *, const value_list_t *, user_data_t *);
65 static int32_t kafka_partition(const rd_kafka_topic_t *, const void *, size_t,
66                                int32_t, void *, void *);
67
68 #if defined HAVE_LIBRDKAFKA_LOGGER || defined HAVE_LIBRDKAFKA_LOG_CB
69 static void kafka_log(const rd_kafka_t *, int, const char *, const char *);
70
71 static void kafka_log(const rd_kafka_t *rkt, int level,
72                       const char *fac, const char *msg)
73 {
74     plugin_log(level, "%s", msg);
75 }
76 #endif
77
78 static int32_t kafka_partition(const rd_kafka_topic_t *rkt,
79                                const void *keydata, size_t keylen,
80                                int32_t partition_cnt, void *p, void *m)
81 {
82     u_int32_t key = *((u_int32_t *)keydata );
83     u_int32_t target = key % partition_cnt;
84     int32_t   i = partition_cnt;
85
86     while (--i > 0 && !rd_kafka_topic_partition_available(rkt, target)) {
87         target = (target + 1) % partition_cnt;
88     }
89     return target;
90 }
91
92 static int kafka_handle(struct kafka_topic_context *ctx) /* {{{ */
93 {
94     char                         errbuf[1024];
95     rd_kafka_conf_t             *conf;
96     rd_kafka_topic_conf_t       *topic_conf;
97
98     if (ctx->kafka != NULL && ctx->topic != NULL)
99         return(0);
100
101     if (ctx->kafka == NULL) {
102         if ((conf = rd_kafka_conf_dup(ctx->kafka_conf)) == NULL) {
103             ERROR("write_kafka plugin: cannot duplicate kafka config");
104             return(1);
105         }
106
107         if ((ctx->kafka = rd_kafka_new(RD_KAFKA_PRODUCER, conf,
108                                     errbuf, sizeof(errbuf))) == NULL) {
109                 ERROR("write_kafka plugin: cannot create kafka handle.");
110                 return 1;
111         }
112
113         rd_kafka_conf_destroy(ctx->kafka_conf);
114         ctx->kafka_conf = NULL;
115
116         INFO ("write_kafka plugin: created KAFKA handle : %s", rd_kafka_name(ctx->kafka));
117
118 #ifdef HAVE_LIBRDKAFKA_LOGGER
119         rd_kafka_set_logger(ctx->kafka, kafka_log);
120 #endif
121     }
122
123     if (ctx->topic == NULL ) {
124         if ((topic_conf = rd_kafka_topic_conf_dup(ctx->conf)) == NULL) {
125             ERROR("write_kafka plugin: cannot duplicate kafka topic config");
126             return 1;
127         }
128
129         if ((ctx->topic = rd_kafka_topic_new(ctx->kafka, ctx->topic_name,
130                                                 topic_conf)) == NULL) {
131                 ERROR("write_kafka plugin: cannot create topic : %s\n", 
132                         rd_kafka_err2str(rd_kafka_errno2err(errno)));
133                 return errno;
134         }
135
136         rd_kafka_topic_conf_destroy(ctx->conf);
137         ctx->conf = NULL;
138
139         INFO ("write_kafka plugin: handle created for topic : %s", rd_kafka_topic_name(ctx->topic));
140     }
141
142     return(0);
143
144 } /* }}} int kafka_handle */
145
146 static int kafka_write(const data_set_t *ds, /* {{{ */
147               const value_list_t *vl,
148               user_data_t *ud)
149 {
150         int                      status = 0;
151     u_int32_t    key;
152     char         buffer[8192];
153     size_t bfree = sizeof(buffer);
154     size_t bfill = 0;
155     size_t blen = 0;
156         struct kafka_topic_context      *ctx = ud->data;
157
158     if ((ds == NULL) || (vl == NULL) || (ctx == NULL))
159         return EINVAL;
160
161     pthread_mutex_lock (&ctx->lock);
162     status = kafka_handle(ctx);
163     pthread_mutex_unlock (&ctx->lock);
164     if( status != 0 )
165         return status;
166
167     bzero(buffer, sizeof(buffer));
168
169     switch (ctx->format) {
170     case KAFKA_FORMAT_COMMAND:
171         status = create_putval(buffer, sizeof(buffer), ds, vl);
172         if (status != 0) {
173             ERROR("write_kafka plugin: create_putval failed with status %i.",
174                   status);
175             return status;
176         }
177         blen = strlen(buffer);
178         break;
179     case KAFKA_FORMAT_JSON:
180
181         format_json_initialize(buffer, &bfill, &bfree);
182         format_json_value_list(buffer, &bfill, &bfree, ds, vl,
183                                ctx->store_rates);
184         format_json_finalize(buffer, &bfill, &bfree);
185         blen = strlen(buffer);
186         break;
187     case KAFKA_FORMAT_GRAPHITE:
188         status = format_graphite(buffer, sizeof(buffer), ds, vl,
189                                  ctx->prefix, ctx->postfix, ctx->escape_char,
190                                  ctx->graphite_flags);
191         if (status != 0) {
192             ERROR("write_kafka plugin: format_graphite failed with status %i.",
193                   status);
194             return status;
195         }
196         blen = strlen(buffer);
197         break;
198     default:
199         ERROR("write_kafka plugin: invalid format %i.", ctx->format);
200         return -1;
201     }
202
203     /*
204      * We partition our stream by metric name
205      */
206     if (ctx->has_key)
207         key = ctx->key;
208     else
209         key = rand();
210
211     rd_kafka_produce(ctx->topic, RD_KAFKA_PARTITION_UA,
212                      RD_KAFKA_MSG_F_COPY, buffer, blen,
213                      &key, sizeof(key), NULL);
214
215         return status;
216 } /* }}} int kafka_write */
217
218 static void kafka_topic_context_free(void *p) /* {{{ */
219 {
220         struct kafka_topic_context *ctx = p;
221
222         if (ctx == NULL)
223                 return;
224
225     if (ctx->topic_name != NULL)
226         sfree(ctx->topic_name);
227     if (ctx->topic != NULL)
228         rd_kafka_topic_destroy(ctx->topic);
229     if (ctx->conf != NULL)
230         rd_kafka_topic_conf_destroy(ctx->conf);
231     if (ctx->kafka_conf != NULL)
232         rd_kafka_conf_destroy(ctx->kafka_conf);
233     if (ctx->kafka != NULL)
234         rd_kafka_destroy(ctx->kafka);
235
236     sfree(ctx);
237 } /* }}} void kafka_topic_context_free */
238
239 static void kafka_config_topic(rd_kafka_conf_t *conf, oconfig_item_t *ci) /* {{{ */
240 {
241     int                          status;
242     int                          i;
243     struct kafka_topic_context  *tctx;
244     char                        *key = NULL;
245     char                        *val;
246     char                         callback_name[DATA_MAX_NAME_LEN];
247     char                         errbuf[1024];
248     user_data_t                  ud;
249         oconfig_item_t              *child;
250     rd_kafka_conf_res_t          ret;
251
252         if ((tctx = calloc(1, sizeof (*tctx))) == NULL) {
253                 ERROR ("write_kafka plugin: calloc failed.");
254         return;
255         }
256
257     tctx->escape_char = '.';
258     tctx->store_rates = 1;
259     tctx->format = KAFKA_FORMAT_JSON;
260
261     if ((tctx->kafka_conf = rd_kafka_conf_dup(conf)) == NULL) {
262         sfree(tctx);
263         ERROR("write_kafka plugin: cannot allocate memory for kafka config");
264         return;
265     }
266
267 #ifdef HAVE_LIBRDKAFKA_LOG_CB
268     rd_kafka_conf_set_log_cb(tctx->kafka_conf, kafka_log);
269 #endif
270
271     if ((tctx->conf = rd_kafka_topic_conf_new()) == NULL) {
272         rd_kafka_conf_destroy(tctx->kafka_conf);
273         sfree(tctx);
274         ERROR ("write_kafka plugin: cannot create topic configuration.");
275         return;
276     }
277
278     if (ci->values_num != 1) {
279         WARNING("kafka topic name needed.");
280         goto errout;
281     }
282
283     if (ci->values[0].type != OCONFIG_TYPE_STRING) {
284         WARNING("kafka topic needs a string argument.");
285         goto errout;
286     }
287
288     if ((tctx->topic_name = strdup(ci->values[0].value.string)) == NULL) {
289         ERROR("write_kafka plugin: cannot copy topic name.");
290         goto errout;
291     }
292
293         for (i = 0; i < ci->children_num; i++) {
294                 /*
295                  * The code here could be simplified but makes room
296                  * for easy adding of new options later on.
297                  */
298                 child = &ci->children[i];
299                 status = 0;
300
301                 if (strcasecmp ("Property", child->key) == 0) {
302                         if (child->values_num != 2) {
303                                 WARNING("kafka properties need both a key and a value.");
304                 goto errout;
305                         }
306                         if (child->values[0].type != OCONFIG_TYPE_STRING ||
307                             child->values[1].type != OCONFIG_TYPE_STRING) {
308                                 WARNING("kafka properties needs string arguments.");
309                 goto errout;
310                         }
311             key = child->values[0].value.string;
312             val = child->values[1].value.string;
313             ret = rd_kafka_topic_conf_set(tctx->conf,key, val,
314                                           errbuf, sizeof(errbuf));
315             if (ret != RD_KAFKA_CONF_OK) {
316                                 WARNING("cannot set kafka topic property %s to %s: %s.",
317                         key, val, errbuf);
318                 goto errout;
319                         }
320
321         } else if (strcasecmp ("Key", child->key) == 0)  {
322             char *tmp_buf = NULL;
323             status = cf_util_get_string(child, &tmp_buf);
324             if (status != 0) {
325                 WARNING("write_kafka plugin: invalid key supplied");
326                 break;
327             }
328
329             if (strcasecmp(tmp_buf, "Random") != 0) {
330                 tctx->has_key = 1;
331                 tctx->key = crc32_buffer((u_char *)tmp_buf, strlen(tmp_buf));
332             }
333             sfree(tmp_buf);
334
335         } else if (strcasecmp ("Format", child->key) == 0) {
336             status = cf_util_get_string(child, &key);
337             if (status != 0)
338                 goto errout;
339
340             assert(key != NULL);
341
342             if (strcasecmp(key, "Command") == 0) {
343                 tctx->format = KAFKA_FORMAT_COMMAND;
344
345             } else if (strcasecmp(key, "Graphite") == 0) {
346                 tctx->format = KAFKA_FORMAT_GRAPHITE;
347
348             } else if (strcasecmp(key, "Json") == 0) {
349                 tctx->format = KAFKA_FORMAT_JSON;
350
351             } else {
352                 WARNING ("write_kafka plugin: Invalid format string: %s",
353                          key);
354             }
355
356             sfree(key);
357
358         } else if (strcasecmp ("StoreRates", child->key) == 0) {
359             status = cf_util_get_boolean (child, &tctx->store_rates);
360             (void) cf_util_get_flag (child, &tctx->graphite_flags,
361                                      GRAPHITE_STORE_RATES);
362
363         } else if (strcasecmp ("GraphiteSeparateInstances", child->key) == 0) {
364             status = cf_util_get_flag (child, &tctx->graphite_flags,
365                                        GRAPHITE_SEPARATE_INSTANCES);
366
367         } else if (strcasecmp ("GraphiteAlwaysAppendDS", child->key) == 0) {
368             status = cf_util_get_flag (child, &tctx->graphite_flags,
369                                        GRAPHITE_ALWAYS_APPEND_DS);
370
371         } else if (strcasecmp ("GraphitePrefix", child->key) == 0) {
372             status = cf_util_get_string (child, &tctx->prefix);
373         } else if (strcasecmp ("GraphitePostfix", child->key) == 0) {
374             status = cf_util_get_string (child, &tctx->postfix);
375         } else if (strcasecmp ("GraphiteEscapeChar", child->key) == 0) {
376             char *tmp_buff = NULL;
377             status = cf_util_get_string (child, &tmp_buff);
378             if (strlen (tmp_buff) > 1)
379                 WARNING ("write_kafka plugin: The option \"GraphiteEscapeChar\" handles "
380                         "only one character. Others will be ignored.");
381             tctx->escape_char = tmp_buff[0];
382             sfree (tmp_buff);
383         } else {
384             WARNING ("write_kafka plugin: Invalid directive: %s.", child->key);
385         }
386
387         if (status != 0)
388             break;
389     }
390
391     rd_kafka_topic_conf_set_partitioner_cb(tctx->conf, kafka_partition);
392     rd_kafka_topic_conf_set_opaque(tctx->conf, tctx);
393
394     ssnprintf(callback_name, sizeof(callback_name),
395               "write_kafka/%s", tctx->topic_name);
396
397     ud.data = tctx;
398     ud.free_func = kafka_topic_context_free;
399
400         status = plugin_register_write (callback_name, kafka_write, &ud);
401         if (status != 0) {
402                 WARNING ("write_kafka plugin: plugin_register_write (\"%s\") "
403                                 "failed with status %i.",
404                                 callback_name, status);
405         goto errout;
406     }
407
408     pthread_mutex_init (&tctx->lock, /* attr = */ NULL);
409
410     return;
411  errout:
412     if (tctx->topic_name != NULL)
413         free(tctx->topic_name);
414     if (tctx->conf != NULL)
415         rd_kafka_topic_conf_destroy(tctx->conf);
416     if (tctx->kafka_conf != NULL)
417                 rd_kafka_conf_destroy(tctx->kafka_conf);
418     sfree(tctx);
419 } /* }}} int kafka_config_topic */
420
421 static int kafka_config(oconfig_item_t *ci) /* {{{ */
422 {
423         int                          i;
424         oconfig_item_t              *child;
425     rd_kafka_conf_t             *conf;
426     rd_kafka_conf_res_t          ret;
427     char                         errbuf[1024];
428
429     if ((conf = rd_kafka_conf_new()) == NULL) {
430         WARNING("cannot allocate kafka configuration.");
431         return -1;
432     }
433         for (i = 0; i < ci->children_num; i++)  {
434                 child = &ci->children[i];
435
436                 if (strcasecmp("Topic", child->key) == 0) {
437                         kafka_config_topic (conf, child);
438                 } else if (strcasecmp(child->key, "Property") == 0) {
439                         char *key = NULL;
440                         char *val = NULL;
441
442                         if (child->values_num != 2) {
443                                 WARNING("kafka properties need both a key and a value.");
444                 goto errout;
445                         }
446                         if (child->values[0].type != OCONFIG_TYPE_STRING ||
447                             child->values[1].type != OCONFIG_TYPE_STRING) {
448                                 WARNING("kafka properties needs string arguments.");
449                 goto errout;
450                         }
451                         if ((key = strdup(child->values[0].value.string)) == NULL) {
452                                 WARNING("cannot allocate memory for attribute key.");
453                 goto errout;
454                         }
455                         if ((val = strdup(child->values[1].value.string)) == NULL) {
456                                 WARNING("cannot allocate memory for attribute value.");
457                 goto errout;
458                         }
459             ret = rd_kafka_conf_set(conf, key, val, errbuf, sizeof(errbuf));
460             if (ret != RD_KAFKA_CONF_OK) {
461                 WARNING("cannot set kafka property %s to %s: %s",
462                         key, val, errbuf);
463                 goto errout;
464             }
465                         sfree(key);
466                         sfree(val);
467                 } else {
468                         WARNING ("write_kafka plugin: Ignoring unknown "
469                                  "configuration option \"%s\" at top level.",
470                                  child->key);
471                 }
472         }
473     if (conf != NULL)
474         rd_kafka_conf_destroy(conf);
475         return (0);
476  errout:
477     if (conf != NULL)
478         rd_kafka_conf_destroy(conf);
479     return -1;
480 } /* }}} int kafka_config */
481
482 void module_register(void)
483 {
484         plugin_register_complex_config ("write_kafka", kafka_config);
485 }
486
487 /* vim: set sw=8 sts=8 ts=8 noet : */