X-Git-Url: https://git.octo.it/?a=blobdiff_plain;ds=sidebyside;f=src%2Fwrite_kafka.c;h=c120d15d9598bfdbabaa94701f696fbe07c45d0f;hb=06a86a60a7dabc685bdbd81ce3d36ea5f7e2c2d4;hp=e3fc5a26f29670773aeea47f08084079957e30fc;hpb=87e53c90047eb1ff8818105762ff997a5cfea4ad;p=collectd.git diff --git a/src/write_kafka.c b/src/write_kafka.c index e3fc5a26..c120d15d 100644 --- a/src/write_kafka.c +++ b/src/write_kafka.c @@ -31,6 +31,7 @@ #include "utils_cmd_putval.h" #include "utils_format_graphite.h" #include "utils_format_json.h" +#include "utils_random.h" #include #include @@ -42,7 +43,7 @@ struct kafka_topic_context { #define KAFKA_FORMAT_GRAPHITE 2 uint8_t format; unsigned int graphite_flags; - _Bool store_rates; + bool store_rates; rd_kafka_topic_conf_t *conf; rd_kafka_topic_t *topic; rd_kafka_conf_t *kafka_conf; @@ -76,6 +77,14 @@ static void kafka_log(const rd_kafka_t *rkt, int level, const char *fac, } #endif +static rd_kafka_resp_err_t kafka_error() { +#if RD_KAFKA_VERSION >= 0x000b00ff + return rd_kafka_last_error(); +#else + return rd_kafka_errno2err(errno); +#endif +} + static uint32_t kafka_hash(const char *keydata, size_t keylen) { uint32_t hash = 5381; for (; keylen > 0; keylen--) @@ -88,7 +97,7 @@ static uint32_t kafka_hash(const char *keydata, size_t keylen) { #define KAFKA_RANDOM_KEY_BUFFER \ (char[KAFKA_RANDOM_KEY_SIZE]) { "" } static char *kafka_random_key(char buffer[static KAFKA_RANDOM_KEY_SIZE]) { - ssnprintf(buffer, KAFKA_RANDOM_KEY_SIZE, "%08lX", (unsigned long)mrand48()); + snprintf(buffer, KAFKA_RANDOM_KEY_SIZE, "%08" PRIX32, cdrand_u()); return buffer; } @@ -112,12 +121,12 @@ static int kafka_handle(struct kafka_topic_context *ctx) /* {{{ */ rd_kafka_topic_conf_t *topic_conf; if (ctx->kafka != NULL && ctx->topic != NULL) - return (0); + return 0; if (ctx->kafka == NULL) { if ((conf = rd_kafka_conf_dup(ctx->kafka_conf)) == NULL) { ERROR("write_kafka plugin: cannot duplicate kafka config"); - return (1); + return 1; } if ((ctx->kafka = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errbuf, @@ -146,7 +155,7 @@ static int kafka_handle(struct kafka_topic_context *ctx) /* {{{ */ if ((ctx->topic = rd_kafka_topic_new(ctx->kafka, ctx->topic_name, topic_conf)) == NULL) { ERROR("write_kafka plugin: cannot create topic : %s\n", - rd_kafka_err2str(rd_kafka_errno2err(errno))); + rd_kafka_err2str(kafka_error())); return errno; } @@ -157,7 +166,7 @@ static int kafka_handle(struct kafka_topic_context *ctx) /* {{{ */ rd_kafka_topic_name(ctx->topic)); } - return (0); + return 0; } /* }}} int kafka_handle */ @@ -264,7 +273,7 @@ static void kafka_config_topic(rd_kafka_conf_t *conf, } tctx->escape_char = '.'; - tctx->store_rates = 1; + tctx->store_rates = true; tctx->format = KAFKA_FORMAT_JSON; tctx->key = NULL; @@ -397,8 +406,8 @@ static void kafka_config_topic(rd_kafka_conf_t *conf, rd_kafka_topic_conf_set_partitioner_cb(tctx->conf, kafka_partition); rd_kafka_topic_conf_set_opaque(tctx->conf, tctx); - ssnprintf(callback_name, sizeof(callback_name), "write_kafka/%s", - tctx->topic_name); + snprintf(callback_name, sizeof(callback_name), "write_kafka/%s", + tctx->topic_name); status = plugin_register_write( callback_name, kafka_write, @@ -480,7 +489,7 @@ static int kafka_config(oconfig_item_t *ci) /* {{{ */ } if (conf != NULL) rd_kafka_conf_destroy(conf); - return (0); + return 0; errout: if (conf != NULL) rd_kafka_conf_destroy(conf);