sr-dev September 2024

sr-dev@lists.kamailio.org

15 participants
169 discussions

git:master:556682f8: kafka: Fix per topic stats bug

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: 556682f8e2cf7e064b7df1ff31fdc496dc51b7ac URL: https://github.com/kamailio/kamailio/commit/556682f8e2cf7e064b7df1ff31fdc49… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: Fix per topic stats bug Bug added by me in commit 0b5e6ac310b7acdf96da0ec6bcbdca021a5089a2. --- Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/556682f8e2cf7e064b7df1ff31fdc49… Patch: https://github.com/kamailio/kamailio/commit/556682f8e2cf7e064b7df1ff31fdc49… --- diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index d9e8a8b9f6d..f8e9379da80 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -1045,6 +1045,8 @@ static kfk_stats_t *kfk_stats_topic_new(str *topic, rd_kafka_resp_err_t err) goto error; } memcpy(st->topic_name->s, topic->s, topic->len); + st->topic_name->s[topic->len] = '\0'; + st->topic_name->len = topic->len; st->total++; if(err) {

2 months, 1 week

git:master:bb5bd6e1: kafka: Add interceptor for broker state

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: bb5bd6e137c758e06736764d823f7437b78d6207 URL: https://github.com/kamailio/kamailio/commit/bb5bd6e137c758e06736764d823f743… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: Add interceptor for broker state So far there is no message logged by librdkafka or kamailio when broker is back up again. The only way to find out that broker is back up is to constatntly check kamcmd stats when errors stop growing. Broker state interceptor is available since librdkafka v2.0.0-RC1 (Dec 21 2022). So only if you are using that librdkafka version, or newer than that, this code will be enabled. --- Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/bb5bd6e137c758e06736764d823f743… Patch: https://github.com/kamailio/kamailio/commit/bb5bd6e137c758e06736764d823f743… --- diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index b2999a7d721..d9e8a8b9f6d 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -220,6 +220,31 @@ static void kfk_msg_delivered( } } +#if RD_KAFKA_VERSION >= 0x020000ff +static rd_kafka_resp_err_t ic_broker_state_change(rd_kafka_t *rk, + int32_t broker_id, const char *secproto, const char *name, int port, + const char *state, void *ic_opaque) +{ + if(strcmp(state, "UP") == 0) { + LM_NOTICE("Connected broker: id: %d, proto: %s, name: %s, port: %d", + broker_id, secproto, name, port); + } + return RD_KAFKA_RESP_ERR_NO_ERROR; +} +#endif + +static rd_kafka_resp_err_t ic_on_new(rd_kafka_t *rk, + const rd_kafka_conf_t *conf, void *ic_opaque, char *errstr, + size_t errstr_size) +{ + +#if RD_KAFKA_VERSION >= 0x020000ff + rd_kafka_interceptor_add_on_broker_state_change( + rk, "ic_broker_state_change", ic_broker_state_change, NULL); +#endif + return RD_KAFKA_RESP_ERR_NO_ERROR; +} + /** * \brief Initialize kafka functionality. * @@ -256,6 +281,9 @@ int kfk_init(char *brokers) /* Set message delivery callback. */ rd_kafka_conf_set_dr_msg_cb(rk_conf, kfk_msg_delivered); + /* Set interceptors init function. */ + rd_kafka_conf_interceptor_add_on_new(rk_conf, "ic_on_new", ic_on_new, NULL); + /* Configure properties: */ if(kfk_conf_configure()) { LM_ERR("Failed to configure general properties\n");

2 months, 1 week

git:master:fa92c728: kafka: Add brokers via librdkafka config object

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: fa92c728c10752824ef991d6828c1ddeea92fbe8 URL: https://github.com/kamailio/kamailio/commit/fa92c728c10752824ef991d6828c1dd… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: Add brokers via librdkafka config object This is because rd_kafka_brokers_add() is deprecated and will log a CONFWARN message when kamailio starts. --- Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/fa92c728c10752824ef991d6828c1dd… Patch: https://github.com/kamailio/kamailio/commit/fa92c728c10752824ef991d6828c1dd… --- diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index 7ac2b57f49c..b2999a7d721 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -240,6 +240,16 @@ int kfk_init(char *brokers) */ rk_conf = rd_kafka_conf_new(); + /* Add brokers */ + LM_DBG("Adding brokers: %s\n", brokers); + if(rd_kafka_conf_set( + rk_conf, "bootstrap.servers", brokers, errstr, sizeof(errstr)) + != RD_KAFKA_CONF_OK) { + LM_ERR("No valid brokers specified: %s\n", brokers); + return -1; + } + LM_DBG("Added brokers: %s\n", brokers); + /* Set logger */ rd_kafka_conf_set_log_cb(rk_conf, kfk_logger); @@ -267,14 +277,6 @@ int kfk_init(char *brokers) rk_conf = NULL; /* Now owned by producer. */ LM_DBG("Producer handle created\n"); - LM_DBG("Adding broker: %s\n", brokers); - /* Add brokers */ - if(rd_kafka_brokers_add(rk, brokers) == 0) { - LM_ERR("No valid brokers specified: %s\n", brokers); - return -1; - } - LM_DBG("Added broker: %s\n", brokers); - /* Topic creation and configuration. */ if(kfk_topic_list_configure()) { LM_ERR("Failed to configure topics\n");

2 months, 1 week

git:master:c31c91fc: kafka: Update doc with the new modparams

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: c31c91fc36dda736b0c19f66ac38b1cc500bfe8b URL: https://github.com/kamailio/kamailio/commit/c31c91fc36dda736b0c19f66ac38b1c… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: Update doc with the new modparams --- Modified: src/modules/kafka/doc/kafka_admin.xml --- Diff: https://github.com/kamailio/kamailio/commit/c31c91fc36dda736b0c19f66ac38b1c… Patch: https://github.com/kamailio/kamailio/commit/c31c91fc36dda736b0c19f66ac38b1c… --- diff --git a/src/modules/kafka/doc/kafka_admin.xml b/src/modules/kafka/doc/kafka_admin.xml index 8ada5e9fe07..666e7c2fc3d 100644 --- a/src/modules/kafka/doc/kafka_admin.xml +++ b/src/modules/kafka/doc/kafka_admin.xml @@ -154,6 +154,70 @@ modparam("kafka", "topic", "name=third_topic") </programlisting> </example> </section> + + <section id="kafka.p.init_without_kafka"> + <title><varname>init_without_kafka</varname> (string)</title> + <para> + Set to anything but 0, specifies if kamailio starts even when kafka brokers are not available at startup time. + </para> + <para> + Default value is <emphasis>0</emphasis> (disabled). + </para> + <example> + <title>Set <varname>init_without_kafka</varname> parameter</title> + <programlisting format="linespecific"> +... +modparam("kafka", "init_without_kafka", 1) +... + </programlisting> + </example> + </section> + + <section id="kafka.p.metadata_timeout"> + <title><varname>metadata_timeout</varname> (string)</title> + <para> + Specifies, in milliseconds, how much time kamailio waits to get topic metadata info at startup time. + </para> + <para> + Default value is <emphasis>2000 milliseconds</emphasis> (2 seconds). + </para> + <example> + <title>Set <varname>metadata_timeout</varname> parameter</title> + <programlisting format="linespecific"> +... +modparam("kafka", "metadata_timeout", 1000) +... + </programlisting> + </example> + </section> + + <section id="kafka.p.log_without_overflow"> + <title><varname>log_without_overflow</varname> (string)</title> + <para> + Set to anything but 0, will skip logging most of the error messages that may happen to each kafka message sent to the broker. + This is useful when e.g. kafka broker goes down, not to overflow syslog with error messages. + </para> + <para> + One can alwyas check this module's stats e.g. via RPC commands and see that errors happened or not. + Those errors can have 2 causes: + <itemizedlist> + <listitem>Some errors happened with the config functions kafka_send()/kafka_send_key(). This means that the message wasn't even enqueued by librdkafka.</listitem> + <listitem>Some errors happened on delivery callback, managed by librdkafka. This means that the message was enqueued by librdkafka, but not managed to be delivered to the broker.</listitem> + </itemizedlist> + </para> + <para> + Default value is <emphasis>0</emphasis> (disabled). + </para> + <example> + <title>Set <varname>log_without_overflow</varname> parameter</title> + <programlisting format="linespecific"> +... +modparam("kafka", "log_without_overflow", 1) +... + </programlisting> + </example> + </section> + </section> <section> <title>Functions</title>

2 months, 1 week

git:master:43f56e47: kafka: Update kfk_stats_add() to use str topic

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: 43f56e4787c00f78b5ec2c2fb64ebe69c089c440 URL: https://github.com/kamailio/kamailio/commit/43f56e4787c00f78b5ec2c2fb64ebe6… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: Update kfk_stats_add() to use str topic Also count stats on kfk_message_send() errors. Also call rd_kafka_poll() even on kfk_message_send() errors. --- Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/43f56e4787c00f78b5ec2c2fb64ebe6… Patch: https://github.com/kamailio/kamailio/commit/43f56e4787c00f78b5ec2c2fb64ebe6…

2 months, 1 week

git:master:6a81d90c: kafka: export statistics for total/err messages

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: 6a81d90c8c0ec0ca7a3c1823c53efb51b0556ed0 URL: https://github.com/kamailio/kamailio/commit/6a81d90c8c0ec0ca7a3c1823c53efb5… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: export statistics for total/err messages --- Modified: src/modules/kafka/kafka_mod.c Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/6a81d90c8c0ec0ca7a3c1823c53efb5… Patch: https://github.com/kamailio/kamailio/commit/6a81d90c8c0ec0ca7a3c1823c53efb5… --- diff --git a/src/modules/kafka/kafka_mod.c b/src/modules/kafka/kafka_mod.c index 43504c7b7cc..108ff3c126b 100644 --- a/src/modules/kafka/kafka_mod.c +++ b/src/modules/kafka/kafka_mod.c @@ -45,6 +45,7 @@ #include "../../core/kemi.h" #include "../../core/rpc.h" #include "../../core/rpc_lookup.h" +#include "../../core/counters.h" #include "kfk.h" @@ -64,6 +65,8 @@ static int w_kafka_send_key( /* * Variables and functions to deal with module parameters. */ +stat_var *total_messages; +stat_var *total_messages_err; int child_init_ok = 0; int init_without_kafka = 0; int log_without_overflow = 0; @@ -105,6 +108,10 @@ struct module_exports exports = { mod_destroy /* destroy function */ }; +/*! \brief We expose internal variables via the statistic framework below.*/ +stat_export_t mod_stats[] = {{"total_messages", 0, &total_messages}, + {"total_messages_err", 0, &total_messages_err}, {0, 0, 0}}; + static int mod_init(void) { /* Register RPC commands. */ @@ -119,6 +126,14 @@ static int mod_init(void) return -1; } +#ifdef STATISTICS + /* register statistics */ + if(register_module_stats("kafka", mod_stats) != 0) { + LM_ERR("Failed to register core statistics\n"); + return -1; + } +#endif + return 0; } diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index ed406385e5e..eb9b166d28c 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -35,7 +35,10 @@ #include "../../core/mem/pkg.h" #include "../../core/mem/shm_mem.h" #include "../../core/locking.h" +#include "../../core/counters.h" +extern stat_var *total_messages; +extern stat_var *total_messages_err; extern int child_init_ok; extern int init_without_kafka; extern int log_without_overflow; @@ -1045,9 +1048,11 @@ static int kfk_stats_add(const char *topic, rd_kafka_resp_err_t err) lock_get(stats_lock); stats_general->total++; + update_stat(total_messages, 1); if(err) { stats_general->error++; + update_stat(total_messages_err, 1); } LM_DBG("General stats: total = %" PRIu64 " error = %" PRIu64 "\n",

2 months, 1 week

git:master:d104324e: kafka: add modparam metadata_timeout

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: d104324ec86190d8ce0886f11fea9b131f46a877 URL: https://github.com/kamailio/kamailio/commit/d104324ec86190d8ce0886f11fea9b1… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: add modparam metadata_timeout --- Modified: src/modules/kafka/kafka_mod.c Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/d104324ec86190d8ce0886f11fea9b1… Patch: https://github.com/kamailio/kamailio/commit/d104324ec86190d8ce0886f11fea9b1… --- diff --git a/src/modules/kafka/kafka_mod.c b/src/modules/kafka/kafka_mod.c index 89f3fcd3c42..43504c7b7cc 100644 --- a/src/modules/kafka/kafka_mod.c +++ b/src/modules/kafka/kafka_mod.c @@ -67,6 +67,7 @@ static int w_kafka_send_key( int child_init_ok = 0; int init_without_kafka = 0; int log_without_overflow = 0; +int metadata_timeout = 2000; char *brokers_param = NULL; /**< List of brokers. */ static int kafka_conf_param(modparam_t type, void *val); static int kafka_topic_param(modparam_t type, void *val); @@ -88,7 +89,8 @@ static param_export_t params[] = {{"brokers", PARAM_STRING, &brokers_param}, (void *)kafka_conf_param}, {"topic", PARAM_STRING | USE_FUNC_PARAM, (void *)kafka_topic_param}, {"init_without_kafka", PARAM_INT, &init_without_kafka}, - {"log_without_overflow", PARAM_INT, &log_without_overflow}, {0, 0, 0}}; + {"log_without_overflow", PARAM_INT, &log_without_overflow}, + {"metadata_timeout", PARAM_INT, &metadata_timeout}, {0, 0, 0}}; /** * \brief Kafka :: Module interface diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index 68014b8b4fb..ed406385e5e 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -39,6 +39,7 @@ extern int child_init_ok; extern int init_without_kafka; extern int log_without_overflow; +extern int metadata_timeout; /** * \brief data type for a configuration property. @@ -730,11 +731,6 @@ static int kfk_topic_list_configure() return 0; } -/* -1 means RD_POLL_INFINITE */ -/* 100000 means 100 seconds */ -#define METADATA_TIMEOUT \ - 100000 /**< Timeout when asking for metadata in milliseconds. */ - /** * \brief check that a topic exists in cluster. * @@ -756,7 +752,7 @@ static int kfk_topic_exist(str *topic_name) /* Get metadata for all topics. */ rd_kafka_resp_err_t res; - res = rd_kafka_metadata(rk, 1, NULL, &metadatap, METADATA_TIMEOUT); + res = rd_kafka_metadata(rk, 1, NULL, &metadatap, metadata_timeout); if(res != RD_KAFKA_RESP_ERR_NO_ERROR) { LM_ERR("Failed to get metadata: %s\n", rd_kafka_err2str(res)); goto error;

2 months, 1 week

git:master:a544b14e: kafka: add modparam log_without_overflow

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: a544b14e0bdc331820a505517a52477879734e7c URL: https://github.com/kamailio/kamailio/commit/a544b14e0bdc331820a505517a52477… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: add modparam log_without_overflow --- Modified: src/modules/kafka/kafka_mod.c Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/a544b14e0bdc331820a505517a52477… Patch: https://github.com/kamailio/kamailio/commit/a544b14e0bdc331820a505517a52477… --- diff --git a/src/modules/kafka/kafka_mod.c b/src/modules/kafka/kafka_mod.c index e88bd37384b..89f3fcd3c42 100644 --- a/src/modules/kafka/kafka_mod.c +++ b/src/modules/kafka/kafka_mod.c @@ -66,6 +66,7 @@ static int w_kafka_send_key( */ int child_init_ok = 0; int init_without_kafka = 0; +int log_without_overflow = 0; char *brokers_param = NULL; /**< List of brokers. */ static int kafka_conf_param(modparam_t type, void *val); static int kafka_topic_param(modparam_t type, void *val); @@ -86,7 +87,8 @@ static param_export_t params[] = {{"brokers", PARAM_STRING, &brokers_param}, {"configuration", PARAM_STRING | USE_FUNC_PARAM, (void *)kafka_conf_param}, {"topic", PARAM_STRING | USE_FUNC_PARAM, (void *)kafka_topic_param}, - {"init_without_kafka", PARAM_INT, &init_without_kafka}, {0, 0, 0}}; + {"init_without_kafka", PARAM_INT, &init_without_kafka}, + {"log_without_overflow", PARAM_INT, &log_without_overflow}, {0, 0, 0}}; /** * \brief Kafka :: Module interface diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index 76a5fea75d9..68014b8b4fb 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -38,6 +38,7 @@ extern int child_init_ok; extern int init_without_kafka; +extern int log_without_overflow; /** * \brief data type for a configuration property. @@ -124,6 +125,12 @@ static void kfk_logger( const rd_kafka_t *rk, int level, const char *fac, const char *buf) { + if(log_without_overflow && strstr(buf, "Connection refused") != NULL) { + // libkafka will keep retrying to connect if kafka server is down + // FIX: ignore these types of errors not to get overflowed + return; + } + switch(level) { case LOG_EMERG: LM_NPRL("RDKAFKA fac: %s : %s : %s\n", fac, @@ -190,8 +197,14 @@ static void kfk_msg_delivered( kfk_stats_add(topic_name, rkmessage->err); if(rkmessage->err) { - LM_ERR("RDKAFKA Message delivery failed: %s\n", - rd_kafka_err2str(rkmessage->err)); + if(log_without_overflow) { + // libkafka will log all undelivered msgs as ERR + // FIX: ignore these types of errors not to get overflowed; check stats instead + ; + } else { + LM_ERR("RDKAFKA Message delivery failed: %s\n", + rd_kafka_err2str(rkmessage->err)); + } } else { LM_DBG("RDKAFKA Message delivered (%zd bytes, offset %" PRId64 ", " "partition %" PRId32 "): %.*s\n", @@ -865,7 +878,11 @@ int kfk_message_send(str *topic_name, str *message, str *key) NULL) == -1) { rd_kafka_resp_err_t err = rd_kafka_last_error(); - LM_ERR("Error sending message: %s\n", rd_kafka_err2str(err)); + if(!log_without_overflow) { + LM_ERR("Error sending message: %s\n", rd_kafka_err2str(err)); + } else { + return 0; + } return -1; }

2 months, 1 week

git:master:741bb148: kafka: add modparam init_without_kafka

by Daniel-Constantin Mierla

Module: kamailio Branch: master Commit: 741bb148ddf4311679cfa6e379fa8bdbc8fac5e5 URL: https://github.com/kamailio/kamailio/commit/741bb148ddf4311679cfa6e379fa8bd… Author: Stefan Mititelu <stefan.mititelu(a)net2phone.com> Committer: Daniel-Constantin Mierla <miconda(a)gmail.com> Date: 2024-09-18T10:01:10+02:00 kafka: add modparam init_without_kafka --- Modified: src/modules/kafka/kafka_mod.c Modified: src/modules/kafka/kfk.c --- Diff: https://github.com/kamailio/kamailio/commit/741bb148ddf4311679cfa6e379fa8bd… Patch: https://github.com/kamailio/kamailio/commit/741bb148ddf4311679cfa6e379fa8bd… --- diff --git a/src/modules/kafka/kafka_mod.c b/src/modules/kafka/kafka_mod.c index 26369d6ca31..e88bd37384b 100644 --- a/src/modules/kafka/kafka_mod.c +++ b/src/modules/kafka/kafka_mod.c @@ -64,6 +64,8 @@ static int w_kafka_send_key( /* * Variables and functions to deal with module parameters. */ +int child_init_ok = 0; +int init_without_kafka = 0; char *brokers_param = NULL; /**< List of brokers. */ static int kafka_conf_param(modparam_t type, void *val); static int kafka_topic_param(modparam_t type, void *val); @@ -84,7 +86,7 @@ static param_export_t params[] = {{"brokers", PARAM_STRING, &brokers_param}, {"configuration", PARAM_STRING | USE_FUNC_PARAM, (void *)kafka_conf_param}, {"topic", PARAM_STRING | USE_FUNC_PARAM, (void *)kafka_topic_param}, - {0, 0, 0}}; + {"init_without_kafka", PARAM_INT, &init_without_kafka}, {0, 0, 0}}; /** * \brief Kafka :: Module interface @@ -125,9 +127,15 @@ static int child_init(int rank) if(rank == PROC_INIT || rank == PROC_TCP_MAIN) return 0; + child_init_ok = 1; if(kfk_init(brokers_param)) { - LM_ERR("Failed to initialize Kafka\n"); - return -1; + child_init_ok = 0; + if(init_without_kafka) { + LM_ERR("Failed to initialize Kafka - continue\n"); + } else { + LM_ERR("Failed to initialize Kafka\n"); + return -1; + } } return 0; } diff --git a/src/modules/kafka/kfk.c b/src/modules/kafka/kfk.c index c6dc8398da2..76a5fea75d9 100644 --- a/src/modules/kafka/kfk.c +++ b/src/modules/kafka/kfk.c @@ -36,6 +36,9 @@ #include "../../core/mem/shm_mem.h" #include "../../core/locking.h" +extern int child_init_ok; +extern int init_without_kafka; + /** * \brief data type for a configuration property. */ @@ -587,7 +590,9 @@ static int kfk_topic_configure(kfk_topic_t *ktopic) } int topic_found = kfk_topic_exist(ktopic->topic_name); - if(topic_found == -1) { + if(init_without_kafka) { + ; + } else if(topic_found == -1) { LM_ERR("Failed to search for topic %.*s in cluster\n", ktopic->topic_name->len, ktopic->topic_name->s); goto error; @@ -828,6 +833,12 @@ int kfk_message_send(str *topic_name, str *message, str *key) /* Get topic from name. */ rd_kafka_topic_t *rkt = kfk_topic_get(topic_name); + if(!child_init_ok) { + LM_ERR("kafka module is unusable: child init NOT ok! Skip sending " + "message, message lost!"); + return -1; + } + if(!rkt) { LM_ERR("Topic not found: %.*s\n", topic_name->len, topic_name->s); return -1;

2 months, 1 week

[kamailio/kamailio] topos_redis : show redis error for post mortem analysis (PR #3965)

by ThomasSevestre

#### Pre-Submission Checklist    - [ ] Commit message has the format required by CONTRIBUTING guide - [ ] Commits are split per component (core, individual modules, libs, utils, ...) - [ ] Each component has a single commit (if not, squash them into one commit) - [ ] No commits to README files for modules (changes must be done to docbook files in `doc/` subfolder, the README file is autogenerated) #### Type Of Change - [ ] Small bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds new functionality) - [ ] Breaking change (fix or feature that would change existing functionality) #### Checklist:  - [ ] PR should be backported to stable branches - [ ] Tested changes locally - [ ] Related to issue #XXXX (replace XXXX with an open issue number) #### Description Small log improvement, in order to make easier to find the origin of redis errors. Log sample : ``` WARNING: topos_redis [topos_redis_storage.c:834]: tps_redis_load_branch(): invalid redis result type: 6 WARNING: topos [tps_storage.c:626]: tps_storage_record(): no local address - do record routing for all initial requests ERROR: topos [tps_storage.c:637]: tps_storage_record(): failed to store ``` The root cause is probaly in redis error message and it whould be nice to log it You can view, comment on, or merge this pull request online at: https://github.com/kamailio/kamailio/pull/3965 -- Commit Summary -- * topos_redis : show redis error for post mortem analysis -- File Changes -- M src/modules/topos_redis/topos_redis_storage.c (9) -- Patch Links -- https://github.com/kamailio/kamailio/pull/3965.patch https://github.com/kamailio/kamailio/pull/3965.diff -- Reply to this email directly or view it on GitHub: https://github.com/kamailio/kamailio/pull/3965 You are receiving this because you are subscribed to this thread. Message ID: <kamailio/kamailio/pull/3965(a)github.com>

2 months, 1 week

← Newer
1
...
4
5
6
7
8
9
10
...
17
Older →

Jump to page:

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

2008

sr-dev September 2024