sr-dev September 2024

sr-dev@lists.kamailio.org

15 participants
170 discussions

[kamailio/kamailio] Kafka producer only tasks (PR #3951)
by Stefan Mititelu 18 Sep '24

18 Sep '24

#### Pre-Submission Checklist    - [X] Commit message has the format required by CONTRIBUTING guide - [X] Commits are split per component (core, individual modules, libs, utils, ...) - [X] Each component has a single commit (if not, squash them into one commit) - [X] 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) - [X] 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 - [X] Tested changes locally - [ ] Related to issue #XXXX (replace XXXX with an open issue number) #### Description  Hello, I recently did some tests with kamailio's kafka module, with focus on what happens when kafka broker goes down. First thing I noticed is that kamailio won't start initially if kafka broker is down. Second thing I noticed syslog overflowing with err logs, when kafka broker is down. Third, I noticed that when it starts, kamailio queries topic metadata and waited up to 100 seconds for any answer. I've added modparams for all these three to change behavior. Updated doc for all the new modparams. I've also updated stats errors counting when exposed cfg functions failed for any reason. And some rework to stats function to accept str* instead of char* topic. Also moved librdkafka's poll() function at the begining of the cfg function logic, so polling happens even when some errors happen with the cfg function. You can view, comment on, or merge this pull request online at: https://github.com/kamailio/kamailio/pull/3951 -- Commit Summary -- * Makefile.defs: version set to 5.8.0-rc0 * pkg/kamailio/deb: version set 5.8.0~rc0 * github: set branch to 5.8 [skip ci] * pkg/docker: set branch 5.8 and refresh [skip ci] * tls: restore early init for other modules that use TLS * tls: fix restore early init * tls: fix OpenSSL 1.1.1 compatibility * OpenSSL integration: manage curl_global_init(...) used by modules * Revert ac4f1be039 - split into per module commits * core/rthreads.h: add thread executor for curl_global_init() * http_async_client: libssl thread executor for curl_global_init() * http_client: libssl thread executor for curl_global_init() * xcap_client: libssl thread executor for curl_global_init() * http_async_client: libssl refactor thread executors for curl * file_out: Add header guards * file_out: Fix leaks * tls: fix OpenSSL 1.1.1 engine keys * tls: make explicit ENGINE deprecation in OpenSSL 3 * tls: clean-up of ENGINE * tlsa: removed the map files used in the past for tls engine * tls: remove unused ENGINE define * tls: basic OpenSSL 3 support of provider keys (replaces ENGINE) * core/rthreads.h: add new option tls_threads_mode = 2 * tls: new option tls_threads_mode = 2 * etc/kamailio.cfg: document tls_threads_mode = 2 to use atfork handler * tls: NULL safety check * ndb_redis: reformat exported structures * evrexec: reformat exported structures * microhttpd: docs - note about variables available in the event route * ndb_redis: init enabled for tcp-main and postchildinit callbacks * tcpops: variables to get active connection attributes * tcpops: if connection not found, return attributes from sip msg rcv * ims_usrloc: make matching of username in contact conditional * ims_registrar_pcscf: update registered state to pending registration if contact exists * core: rthreads - init return variable * influxdbc: copy with buffer size limit * core: ppcfg - jump to end when defexp eval does not get a str val * core: select - handle ws and wss inside select_ip_port() * core: parser rr - check for rr value before serializing * acc: cdr - do not free static empty string in case of error * ctl: reset log prefix on reading ctl traffic * core: main - init local variables used for cli param parsing * presence: docs - remove extra section end tag * tls/docs: Update tls.reload docs * tls/docs: Update certificate generation docs * modules: readme files regenerated - modules ... [skip ci] * ChangeLog: added content from 5.8.0 developent * pkg/kamailio/deb: version set 5.8.0 [skip ci] * pkg/kamailio: version set 5.8.0 for rpms and alpine * Makefile.defs: version set to 5.8.0 * ChangeLog: content set for v5.8.0 * pkg/kamailio/obs: packaged README.file_out * core: set null-terminated char for modparam (backport from master) * Avoid 'qm_strnstr' defined but not used warning * topos: small rearangement by removing else after return in the true block * topos: properly handle cases of no user in contact for mode 1 * pkg/kamailio/obs: Allow builders to disable wolfssl module [skip ci] * file_out: Fix var position and initialize * pdb: Return negative carrierid for failures * pdb server: Return negative carrierid for failures. Clang format * ctl: handle int input for double storage * dispatcher: refine documentation * ims_ipsec_pcscf: TLS support in ipsec_forward() improved * acc: more log messages of failure arsing extra acc string * keepalive: proper storing of last up/down timestamps * keepalive: small formatting updates and comments to the ka dest structure * ndb_redis: clang-format for WITH_SSL code * ndb_redis: fix index on parsing sentinel address inside redisc_reconnect_server() * ndb_redis: break on finding a server via sentinel inside redisc_reconnect_server() * modules: readme files regenerated - modules ... [skip ci] * dmq: fix inter-lock on error * file_out: additional checks for silenting warnings * file_out: Cast to correct type of str * tls: Fix ‘for’ loop initial declarations error * file_out: FIx ‘for’ loop initial declarations error * gcrypt: docs - module name used for section ids * core: socket info - parse advertise ip part to get address family * core: resolve/create_srv_pref_list() - insert at correct position in list * corex: fix conditions for dns_cache modparam srv attributes * keepalive: reformat module exports structures * db_mysql: MYSQL_OPT_RECONNECT is deprecated by MySQL 8.0.34 * db_mysql: right position for define KSR_MYSQL_OPT_RECONNECT * smsops: reformat module exports structures * smsops: short reference section about variables * blst: reformat exported structures * modules: readme files regenerated - modules ... [skip ci] * pkg/kamailio: version set 5.8.1 for rpms and alpine * pkg: deb specs updated for v5.8.1 * core: xavp - added missing closing parenthesis in log messages * core: msg translator - duplicate reply via xavp params and handle errors * Makefile.defs: version set to 5.8.1 * ChangeLog: content updated for v5.8.1 * pdb server: allow carrier id 0 * db_redis: docs - refine docs regarding client certificates [skip ci] * ndb_redis: docs - refine docs regarding client certificates [skip ci] * sdpops: reformat the codecs map * core: parser/sdp - rename parameter to suggest better its role * siputils: tel2sip2() - memset to 0 pkg allocs and recompute len of new uri * ims_ipsec_pcscf: log error message made info * corex: Fix unknown af in list_sockets * rtpengine: fix pkg mem leak in send_rtpp_command() * tcp_main: Add protocol argument for searching tcp/tls connections * tcp_main: Add proto argument to tcpconn_exists function * tcp_main: Update comment docs * core/forward: Match protocol when forwarding * tcp_main: Match wss protocol * usrloc: check on db delete the return value of memchr * core: tcp - keep connection id on event route cb structure * tcpops: reworked some debug messages * core: tcp - set a global shortcut to tcp connection for event route * pv: use global tcp connection callback for event route on $conid * core: fmsg - parse loopback ip when initializing faked msg structure * regex: init globals for safer shutdown on failure * regex: allocate pcres array * regex: reformat exported structures * tcpops: removed unused variable introduced recently * modules: readme files regenerated - modules ... [skip ci] * tls: clear thread-local variables up to tls_pthreads_key_mark * core: spelling fix in comment and log message * uid_auth_db: spelling fix in documentation * ims_usrloc_scscf: reformat exported structures * ims_usrloc_scscf: enclose examples in dotted lines for better visibility * modules/statsd: ensure statsd failures do not stop execution (#3819) * ims_registrar_scscf: init variable to fix analyzer warning * ims_ipsec_pcscf: decouple IPSEC_FORWARD_USEVIA and IPSEC_TCPPORT_UEC for requests * ims_ipsec_pcscf: docs updated for ipsec_forward() flags * ims_ipsec_pcscf: updated comments for ipsec_forward() flags * ims_ipsec_pcscf: handle tls for replies * ims_registrar_scscf: fix uninitialized arguments in save() * ims_registrar_scscf: refactor save_fixup4 * ims_registrar_scscf: fix typo in function name uint_fixup * ims_registrar_scscf: free_uint_fixup: fix memleak/freeing the wrong parameter * ims_ipsec_pcscf: cascade on options to get the uri for fill_contact() * sipcapture: docs - fix default value for parameter * tm: free new buffer on drop for local requests * tm: cb flag to mark local uac on request drop * tm: cb_flags made unsigned for uac_req * pv_headers: print also the new TMCB_LOCAL_REQUEST_DROP in debug function * presence: check if the local notify was dropped in event route * uac: check if local requests are dropped and free the param * pua: check if local requests are dropped and free the param * core: parser/contact - add limit for max number of contacts * core: parser/contact - basic validation tests for contact uri * core: parse/contact: jump to error on too many contacts * ctl: removed level and function names from log messages * ctl: info log on ECONNRESET (connection reset by peer) * modules: readme files regenerated - modules ... [skip ci] * tcpops: note about execution of event routes by TCP main process * modules: readme files regenerated - modules ... [skip ci] * sipcapture: fix default DB URL docs * modules: readme files regenerated - modules ... [skip ci] * etc/kamailio.cfg: remove executable flag * tm: add mandatory Max-Forwards header to local AC * db_cluster: Update log levels * dialplan: don't call free on empty pcre pointer * tcpops: reformat exported structures * sipcapture: use local static buffers for ip address and port * db_mysql: fixed build on CentOS 7 * usrloc: delete location_attrs in db_only mode on delete_urecord * htable: error on not finding htable in pv set * htable: error on not finding htable in pv get * rtpengine: Fix extra process in pkg.stats * siputils: free head of params list for get_uri_param() * siputils: free head of params list for get_uri_param() * regex: fix pcre2 migration * geoip2: init MMDB_s global pointer and check on destroy * geoip2: do not close global MMDB_s on destroy * geoip2: rename global lock variable * geoip2: keep global structure per process * geoip2: clang format module file * geoip2: Reload database before accessing it * iims_registrar_scscf: include header for common c functions * core/resolve: Check dns_cache_init and choose appropriate functions * nathelper: fixed handle_ruri_alias_mode(1) cannot proper handle multi alias * nathelper: remove magic constant * dmq_usrloc: reformat exported structures * ims_ipsec_pcscf: Fix typo * pua_dialoginfo: Fix ruri length * modules: readme files regenerated - modules ... [skip ci] * pkg/kamailio: version set 5.8.2 for rpms and alpine * pkg: deb specs updated for v5.8.2 * Makefile.defs: version set to 5.8.2 * ChangeLog: content updated for v5.8.2 * kafka: add modparam init_without_kafka * kafka: add modparam log_without_overflow * kafka: add modparam metadata_timeout * kafka: export statistics for total/err messages * kafka: Update kfk_stats_add() to use str topic * kafka: Update doc with the new modparams -- File Changes -- M .devcontainer/devcontainer.json (2) M .github/workflows/devcontainer.yaml (2) M .github/workflows/main.yml (2) M .gitmodules (1) M ChangeLog (16562) M etc/kamailio.cfg (5) M pkg/docker (2) M pkg/kamailio/alpine/APKBUILD (2) M pkg/kamailio/deb/bionic/changelog (24) M pkg/kamailio/deb/bookworm/changelog (24) M pkg/kamailio/deb/bullseye/changelog (24) M pkg/kamailio/deb/buster/changelog (24) M pkg/kamailio/deb/debian/changelog (24) M pkg/kamailio/deb/focal/changelog (24) M pkg/kamailio/deb/jammy/changelog (24) M pkg/kamailio/deb/jessie/changelog (24) M pkg/kamailio/deb/precise/changelog (24) M pkg/kamailio/deb/sid/changelog (24) M pkg/kamailio/deb/stretch/changelog (24) M pkg/kamailio/deb/trusty/changelog (24) M pkg/kamailio/deb/wheezy/changelog (24) M pkg/kamailio/deb/xenial/changelog (24) M pkg/kamailio/obs/kamailio.spec (24) M src/Makefile.defs (4) M src/core/fmsg.c (11) M src/core/forward.h (3) M src/core/mem/q_malloc.c (2) M src/core/modparam.c (1) M src/core/msg_translator.c (18) M src/core/parser/contact/contact.c (29) M src/core/parser/parse_rr.c (24) M src/core/parser/sdp/sdp_helpr_funcs.c (14) M src/core/ppcfg.c (2) M src/core/resolve.c (60) M src/core/resolve.h (14) M src/core/rthreads.h (58) M src/core/select_core.c (12) M src/core/socket_info.c (5) M src/core/switch.c (2) M src/core/tcp_conn.h (6) M src/core/tcp_main.c (52) M src/core/ut.h (2) M src/core/xavp.c (6) M src/main.c (6) M src/modules/acc/acc_cdr.c (4) M src/modules/acc/acc_extra.c (16) M src/modules/blst/blst.c (49) M src/modules/corex/corex_mod.c (6) M src/modules/corex/corex_rpc.c (4) M src/modules/ctl/binrpc.h (15) M src/modules/ctl/fifo_server.c (1) M src/modules/ctl/io_listener.c (90) M src/modules/db_cluster/dbcl_api.c (83) M src/modules/db_mysql/km_my_con.c (21) M src/modules/db_redis/README (17) M src/modules/db_redis/doc/db_redis_admin.xml (10) M src/modules/dialplan/dp_repl.c (3) M src/modules/dispatcher/README (13) M src/modules/dispatcher/doc/dispatcher_admin.xml (9) M src/modules/dmq/notification_peer.c (6) M src/modules/dmq_usrloc/dmq_usrloc.c (44) M src/modules/evrexec/evrexec_mod.c (40) M src/modules/file_out/file_out.c (69) M src/modules/file_out/types.c (14) M src/modules/file_out/types.h (5) M src/modules/gcrypt/doc/gcrypt_admin.xml (4) M src/modules/geoip2/geoip2_mod.c (25) M src/modules/geoip2/geoip2_pv.c (81) M src/modules/geoip2/geoip2_pv.h (1) M src/modules/htable/ht_var.c (10) M src/modules/http_async_client/http_async_client_mod.c (115) M src/modules/http_client/http_client.c (23) M src/modules/ims_ipsec_pcscf/README (13) M src/modules/ims_ipsec_pcscf/cmd.c (130) M src/modules/ims_ipsec_pcscf/doc/ims_ipsec_pcscf_admin.xml (12) M src/modules/ims_registrar_pcscf/save.c (11) M src/modules/ims_registrar_scscf/ims_registrar_scscf_mod.c (55) M src/modules/ims_registrar_scscf/lookup.c (2) M src/modules/ims_registrar_scscf/userdata_parser.c (5) M src/modules/ims_usrloc_pcscf/udomain.c (58) M src/modules/ims_usrloc_scscf/README (29) M src/modules/ims_usrloc_scscf/doc/ims_usrloc_scscf_admin.xml (129) M src/modules/ims_usrloc_scscf/ims_usrloc_scscf_mod.c (78) M src/modules/influxdbc/ic.c (16) M src/modules/kafka/doc/kafka_admin.xml (64) M src/modules/kafka/kafka_mod.c (33) M src/modules/kafka/kfk.c (117) M src/modules/keepalive/keepalive.h (12) M src/modules/keepalive/keepalive_core.c (4) M src/modules/keepalive/keepalive_mod.c (56) M src/modules/microhttpd/doc/microhttpd_admin.xml (5) M src/modules/nathelper/nathelper.c (4) M src/modules/ndb_redis/README (17) M src/modules/ndb_redis/doc/ndb_redis_admin.xml (10) M src/modules/ndb_redis/ndb_redis_mod.c (128) M src/modules/ndb_redis/redis_client.c (34) M src/modules/pdb/pdb.c (14) M src/modules/presence/doc/presence_admin.xml (1) M src/modules/presence/notify.c (5) M src/modules/pua/pua.c (5) M src/modules/pua/send_publish.c (3) M src/modules/pua_dialoginfo/pua_dialoginfo.c (1) M src/modules/pv/pv_core.c (7) M src/modules/pv_headers/pv_headers.c (2) M src/modules/regex/regex_mod.c (78) M src/modules/rtpengine/rtpengine.c (7) M src/modules/sdpops/sdpops_data.c (28) M src/modules/sipcapture/README (5) M src/modules/sipcapture/doc/sipcapture_admin.xml (4) M src/modules/sipcapture/hep.c (13) M src/modules/siputils/checks.c (14) M src/modules/smsops/README (13) M src/modules/smsops/doc/smsops_admin.xml (14) M src/modules/smsops/smsops.c (47) M src/modules/statsd/statsd.c (51) M src/modules/tcpops/README (8) M src/modules/tcpops/doc/eventroutes.xml (12) M src/modules/tcpops/tcpops.c (6) M src/modules/tcpops/tcpops_mod.c (155) M src/modules/tls/README (12) M src/modules/tls/doc/rpc.xml (4) M src/modules/tls/doc/tls.xml (10) M src/modules/tls/tls_domain.c (94) D src/modules/tls/tls_map.c (213) D src/modules/tls/tls_map.h (77) M src/modules/tls/tls_mod.c (223) M src/modules/tls/tls_rand.h (2) M src/modules/tls/tls_server.c (24) D src/modules/tlsa/tls_map.c (27) D src/modules/tlsa/tls_map.h (27) M src/modules/tm/t_hooks.h (4) M src/modules/tm/t_msgbuilder.c (5) M src/modules/tm/uac.c (3) M src/modules/tm/uac.h (2) M src/modules/topos/tps_storage.c (71) M src/modules/uac/uac_reg.c (7) M src/modules/uac/uac_send.c (9) M src/modules/uid_auth_db/auth_db.xml (2) M src/modules/usrloc/ucontact.c (36) M src/modules/usrloc/urecord.c (18) M src/modules/xcap_client/xcap_client.c (5) M utils/pdbt/common.h (2) M utils/pdbt/pdb_server.c (330) M utils/pdbt/pdb_server_backend.c (21) -- Patch Links -- https://github.com/kamailio/kamailio/pull/3951.patch https://github.com/kamailio/kamailio/pull/3951.diff -- Reply to this email directly or view it on GitHub: https://github.com/kamailio/kamailio/pull/3951 You are receiving this because you are subscribed to this thread. Message ID: <kamailio/kamailio/pull/3951(a)github.com>

2 8

git:master:556682f8: kafka: Fix per topic stats bug
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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) {

1 0

git:master:bb5bd6e1: kafka: Add interceptor for broker state
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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");

1 0

git:master:fa92c728: kafka: Add brokers via librdkafka config object
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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");

1 0

git:master:c31c91fc: kafka: Update doc with the new modparams
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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>

1 0

git:master:43f56e47: kafka: Update kfk_stats_add() to use str topic
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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…

1 0

git:master:6a81d90c: kafka: export statistics for total/err messages
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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",

1 0

git:master:d104324e: kafka: add modparam metadata_timeout
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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;

1 0

git:master:a544b14e: kafka: add modparam log_without_overflow
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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; }

1 0

git:master:741bb148: kafka: add modparam init_without_kafka
by Daniel-Constantin Mierla 18 Sep '24

18 Sep '24

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;

1 0

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

Jump to page:

2025

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

2010

2009

2008

sr-dev September 2024