diff --git a/Makefile.conf.template b/Makefile.conf.template index c6396cc5f49..0aa9a27cfb2 100644 --- a/Makefile.conf.template +++ b/Makefile.conf.template @@ -18,6 +18,7 @@ #db_oracle= Provides Oracle connectivity for OpenSIPS. | Development library of OCI, typically instantclient-sdk-10.2.0.3 #db_perlvdb= Provides a virtualization framework for OpenSIPS's database access. | Perl library development files, typically libperl-dev #db_postgres= Provides Postgres connectivity for OpenSIPS | PostgreSQL library and development library - typically libpq5 and libpq-dev +#db_redis= Provides Redis (single instance or cluster) row storage for OpenSIPS | Redis client library, hiredis #db_sqlite= Provides SQLite connectivity for OpenSIPS | SQLite library and development library - typically libsqlite3 and libsqlite3-dev #db_unixodbc= Allows to use the unixodbc package with OpenSIPS | ODBC library and ODBC development library #dialplan= Implements generic string translations based on matching and replacement rules | PCRE development library, typically libpcre2-dev @@ -76,7 +77,7 @@ #uuid= UUID generator | uuid-dev # Modules omitted from the default build, generally due to external dependencies. -exclude_modules?= aaa_diameter aaa_radius auth_jwt auth_web3 b2b_logic_xml cachedb_cassandra cachedb_couchbase cachedb_dynamodb cachedb_memcached cachedb_mongodb cachedb_redis carrierroute cgrates compression cpl_c db_berkeley db_http db_mysql db_oracle db_perlvdb db_postgres db_sqlite db_unixodbc dialplan emergency event_rabbitmq event_kafka event_sqs h350 httpd http2d identity jabber json launch_darkly ldap lua mi_xmlrpc mmgeoip opentelemetry osp perl pi_http presence presence_dialoginfo presence_mwi presence_reginfo presence_xml presence_dfks proto_ipsec proto_sctp proto_tls proto_wss pua pua_bla pua_dialoginfo pua_mi pua_reginfo pua_usrloc pua_xmpp python regex rabbitmq_consumer rest_client rls rtp.io siprec sngtc snmpstats stir_shaken tls_mgm tls_openssl tls_wolfssl uuid xcap xcap_client xml xmpp +exclude_modules?= aaa_diameter aaa_radius auth_jwt auth_web3 b2b_logic_xml cachedb_cassandra cachedb_couchbase cachedb_dynamodb cachedb_memcached cachedb_mongodb cachedb_redis carrierroute cgrates compression cpl_c db_berkeley db_http db_mysql db_oracle db_perlvdb db_postgres db_redis db_sqlite db_unixodbc dialplan emergency event_rabbitmq event_kafka event_sqs h350 httpd http2d identity jabber json launch_darkly ldap lua mi_xmlrpc mmgeoip opentelemetry osp perl pi_http presence presence_dialoginfo presence_mwi presence_reginfo presence_xml presence_dfks proto_ipsec proto_sctp proto_tls proto_wss pua pua_bla pua_dialoginfo pua_mi pua_reginfo pua_usrloc pua_xmpp python regex rabbitmq_consumer rest_client rls rtp.io siprec sngtc snmpstats stir_shaken tls_mgm tls_openssl tls_wolfssl uuid xcap xcap_client xml xmpp # Modules forced into the build, even when also listed in exclude_modules. include_modules?= diff --git a/modules/db_redis/Makefile b/modules/db_redis/Makefile new file mode 100644 index 00000000000..d70cb2e5d0c --- /dev/null +++ b/modules/db_redis/Makefile @@ -0,0 +1,11 @@ +# WARNING: do not run this directly, it should be run by the master Makefile + +include ../../Makefile.defs +auto_gen= +NAME=db_redis.so + +# hiredis client library +DEFS+=-I$(LOCALBASE)/include -I$(SYSBASE)/include +LIBS+=-L$(LOCALBASE)/lib -L$(SYSBASE)/lib -lhiredis + +include ../../Makefile.modules diff --git a/modules/db_redis/README b/modules/db_redis/README new file mode 100644 index 00000000000..0b4617d761d --- /dev/null +++ b/modules/db_redis/README @@ -0,0 +1,348 @@ +db_redis Module + __________________________________________________________ + + Table of Contents + + 1. Admin Guide + + 1.1. Overview + 1.2. Data Model + 1.3. Deployment Modes + + 1.3.1. Single endpoint + 1.3.2. Redis Cluster + + 1.4. Dependencies + + 1.4.1. OpenSIPS Modules + 1.4.2. External Libraries or Applications + + 1.5. Exported Parameters + + 1.5.1. connect_timeout (integer) + 1.5.2. query_timeout (integer) + 1.5.3. scan_count (integer) + 1.5.4. mode (string) + + 1.6. Exported Functions + 1.7. Usage + 1.8. Limitations + 1.9. Installation + + 2. Contributors + + 2.1. By Commit Statistics + 2.2. By Commit Activity + + 3. Documentation + + 3.1. Contributors + + List of Tables + + 2.1. Top contributors by DevScore^(1), authored commits^(2) and + lines added/removed^(3) + + 2.2. Most recently active contributors^(1) to this module + + List of Examples + + 1.1. Set connect_timeout parameter + 1.2. Set query_timeout parameter + 1.3. Set scan_count parameter + 1.4. Set mode parameter + 1.5. Persisting the dialog table in a single Redis server + 1.6. Persisting the dialog table in a Redis Cluster + +Chapter 1. Admin Guide + +1.1. Overview + + This module provides a OpenSIPS DB API backend that stores data + in a Redis server or a Redis Cluster. It implements the + structured database interface used by modules such as dialog, + so that their tables can be persisted in Redis instead of a + traditional SQL database. + + Unlike the cachedb_redis module, which exposes Redis through + the key-value NoSQL interface, db_redis presents Redis as a + relational-style backend behind the standard db_url mechanism. + A module is switched to Redis simply by pointing its db_url at + a redis:// URL; no code change is required in the consuming + module. + + The connection type is auto-detected at startup: the module + works either against a single Redis endpoint (a standalone + server, or the current master reached through an external load + balancer/Sentinel setup) or against a fully-enabled Redis + Cluster, where it discovers the whole topology and keeps a + connection open to every master shard. + +1.2. Data Model + + Redis is schemaless, so the module needs the table layout to be + declared out of band. Both the schema definitions and the data + are stored as Redis hashes: + * Schema - the hash “schema:” describes one table. It + holds a “__cols” field with the ordered, space-separated + list of column names, a “__pk” field naming the primary-key + column, and one field per column giving its type as + “type[,null][,auto]”. The supported types are int, bigint, + double, string, blob and datetime. The null attribute marks + a nullable column and auto marks an auto-increment primary + key. + * Rows - each row is the hash “
:”, + with one field per non-NULL column. A NULL column is + represented by the absence of the field. Values of blob + columns are stored verbatim (binary-safe), with no encoding + overhead. + * Sequences - auto-increment primary keys are generated with + INCR on the counter “seq:
”. + * Table version - the standard OpenSIPS “version” table is + provisioned like any other table, with one row per served + table, for example “HSET version:dialog table_name dialog + table_version 12”. + + The key prefixes “schema:” and “seq:” are reserved by the + module and must not be used as table names. Ready-made + provisioning files are shipped under the “scripts/redis/” + directory; see that directory's README for details. A schema is + loaded from Redis on first access to a table and cached for the + lifetime of the connection. + +1.3. Deployment Modes + + At connection time the module issues an “INFO CLUSTER” command + and inspects the “cluster_enabled” field to decide how to + operate. The detected behaviour can be pinned with the mode + parameter. + +1.3.1. Single endpoint + + Used for a standalone Redis server, or when a Redis Sentinel + deployment is fronted by an external load balancer that always + routes to the current master. The module keeps a single + connection and transparently reconnects on I/O errors or when + it receives a “-READONLY” reply (which indicates it is talking + to a demoted master after a failover). Sentinel discovery and + master election are expected to be handled externally. + +1.3.2. Redis Cluster + + When the endpoint is a cluster node, the module fetches the + full topology (via “CLUSTER SHARDS”, falling back to “CLUSTER + SLOTS” on older servers) starting from the single seed host + given in db_url. It then opens and keeps warm a connection to + every master shard, so that a key routed to any shard can be + served immediately over an already established connection + rather than paying a new TCP handshake on each redirect. + + Each key is routed to its shard by computing the CRC16 hash + slot (honouring “{hash tag}” braces). “MOVED” redirects are + served from the warm connection pool and trigger a deferred + topology refresh; “ASK” redirects are followed with a one-shot + “ASKING”; and “TRYAGAIN” replies (seen during slot migration) + are retried after a short back-off. Full-table operations scan + every master and merge the results. + + Only master nodes are used; the module always reads from and + writes to masters to guarantee read-your-writes consistency. + +1.4. Dependencies + +1.4.1. OpenSIPS Modules + + The following modules must be loaded before this module: + * No dependencies on other OpenSIPS modules. + +1.4.2. External Libraries or Applications + + The following libraries or applications must be installed + before running OpenSIPS with this module loaded: + * hiredis - the minimalistic Redis client C library + (development package, typically libhiredis-dev). + +1.5. Exported Parameters + +1.5.1. connect_timeout (integer) + + The timeout, in milliseconds, for establishing a TCP connection + to a Redis node. It also bounds how long a failover blackout + can stall an operation before it is retried or failed. + + Default value is 1000 (1 second). + + Example 1.1. Set connect_timeout parameter +... +modparam("db_redis", "connect_timeout", 500) +... + +1.5.2. query_timeout (integer) + + The send/receive timeout, in milliseconds, applied to commands + issued on an established connection. Keep it small enough that + a stalled node does not block OpenSIPS worker processes for + long, especially when the dialog module is used in realtime + db_mode. + + Default value is 2000 (2 seconds). + + Example 1.2. Set query_timeout parameter +... +modparam("db_redis", "query_timeout", 1000) +... + +1.5.3. scan_count (integer) + + The “COUNT” hint passed to the Redis “SCAN” command when + iterating a whole table (used by full-table loads and by + queries that are not addressed by primary key). Larger values + reduce the number of round-trips at the cost of larger + individual replies. + + Default value is 100. + + Example 1.3. Set scan_count parameter +... +modparam("db_redis", "scan_count", 500) +... + +1.5.4. mode (string) + + Pins the connection mode instead of auto-detecting it. Accepted + values are: + * auto - detect single vs. cluster from “INFO CLUSTER” (the + default). + * single - force single-endpoint mode; startup fails if the + endpoint is a cluster node. + * cluster - force cluster mode; startup fails if the endpoint + has cluster support disabled. + + Pinning the mode turns an accidental misconfiguration (for + example a db_url pointing at the wrong kind of server) into a + clear startup error instead of a silently wrong runtime + behaviour. + + Default value is "auto". + + Example 1.4. Set mode parameter +... +modparam("db_redis", "mode", "cluster") +... + +1.6. Exported Functions + + No function exported to be used from the configuration file. + +1.7. Usage + + The module is used by pointing a consuming module's db_url at a + redis:// URL. The URL has the standard OpenSIPS form + “redis://[:password@]host[:port]/db”. The database path + component is mandatory for the URL parser but is only + meaningful for a standalone server (it selects the logical + database number); it is ignored in cluster mode. A + password-only Redis is authenticated with the “:password@” + form. + + Example 1.5. Persisting the dialog table in a single Redis + server +... +loadmodule "db_redis/db_redis.so" +loadmodule "dialog/dialog.so" + +modparam("dialog", "db_mode", 1) +modparam("dialog", "db_url", "redis://127.0.0.1:6379/0") +... + + Example 1.6. Persisting the dialog table in a Redis Cluster +... +loadmodule "db_redis/db_redis.so" +loadmodule "dialog/dialog.so" + +modparam("dialog", "db_mode", 1) +# a single cluster node is enough as a seed; the rest of the topology +# is discovered automatically +modparam("dialog", "db_url", "redis://:secretpass@10.0.0.20:6379/0") +... + + Before starting OpenSIPS, provision the table schemas into + Redis, for example (add “-c” when talking to a cluster): +redis-cli [-c] < scripts/redis/dialog-create.redis + +1.8. Limitations + + * Queries support equality and the ordering operators (“=”, + “<”, “>”, “<=”, “>=”, “!=”) on any column, combined with + AND or OR, plus a single “ORDER BY” column. Lookups + addressed by the primary key are served directly; every + other query performs a table scan and filters in the + client. There is no “JOIN” support (the OpenSIPS DB API + does not provide one) and no raw-query support. + * Updating the value of the primary-key column of an existing + row is not supported. + * The module is intended for modules that use the structured + DB API with per-row primary-key access, such as dialog, + dispatcher, clusterer and usrloc. Raw SQL queries are not + implemented; a module that relies on them works only if it + also offers a structured-query path (as usrloc does for + contact loading). + * As with the SQL backends, sharing rows between several + OpenSIPS instances through the same Redis store requires + the instances to use a matching configuration (for the + dialog table, an identical hash_size and matching listening + sockets). + +1.9. Installation + + Because it depends on an external library, the db_redis module + is not compiled and installed by default. You can use one of + the next options. + * - edit the "Makefile" and remove "db_redis" from + "excluded_modules" list. Then follow the standard procedure + to install OpenSIPS: "make all; make install". + * - from command line use: 'make all + include_modules="db_redis"; make install + include_modules="db_redis"'. + +Chapter 2. Contributors + +2.1. By Commit Statistics + + Table 2.1. Top contributors by DevScore^(1), authored + commits^(2) and lines added/removed^(3) + Name DevScore Commits Lines ++ Lines -- + 1. Yury Kirsanov 27 2 2802 4 + + (1) DevScore = author_commits + author_lines_added / + (project_lines_added / project_commits) + author_lines_deleted + / (project_lines_deleted / project_commits) + + (2) including any documentation-related commits, excluding + merge commits. Regarding imported patches/code, we do our best + to count the work on behalf of the proper owner, as per the + "fix_authors" and "mod_renames" arrays in + opensips/docs/build-contrib.sh. If you identify any + patches/commits which do not get properly attributed to you, + please submit a pull request which extends "fix_authors" and/or + "mod_renames". + + (3) ignoring whitespace edits, renamed files and auto-generated + files + +2.2. By Commit Activity + + Table 2.2. Most recently active contributors^(1) to this module + Name Commit Activity + 1. Yury Kirsanov Jul 2026 - Jul 2026 + + (1) including any documentation-related commits, excluding + merge commits + +Chapter 3. Documentation + +3.1. Contributors + + Documentation Copyrights: + + Copyright © 2026 www.opensips-solutions.com diff --git a/modules/db_redis/db_redis.c b/modules/db_redis/db_redis.c new file mode 100644 index 00000000000..2d84ab986aa --- /dev/null +++ b/modules/db_redis/db_redis.c @@ -0,0 +1,119 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include "../../sr_module.h" +#include "../../db/db.h" +#include "db_redis.h" +#include "dbase.h" + +int rdb_connect_timeout = 1000; /* ms */ +int rdb_query_timeout = 2000; /* ms */ +int rdb_scan_count = 100; +int rdb_mode = RDB_MODE_AUTO; + +static char *rdb_mode_str; + +static int redis_mod_init(void); + +static const cmd_export_t cmds[] = { + {"db_bind_api", (cmd_function)db_redis_bind_api, {{0,0,0}},0}, + {0,0,{{0,0,0}},0} +}; + +static const param_export_t params[] = { + {"connect_timeout", INT_PARAM, &rdb_connect_timeout}, + {"query_timeout", INT_PARAM, &rdb_query_timeout}, + {"scan_count", INT_PARAM, &rdb_scan_count}, + {"mode", STR_PARAM, &rdb_mode_str}, + {0, 0, 0} +}; + +struct module_exports exports = { + "db_redis", + MOD_TYPE_SQLDB, /* class of this module */ + MODULE_VERSION, + DEFAULT_DLFLAGS, /* dlopen flags */ + 0, /* load function */ + NULL, /* OpenSIPS module dependencies */ + cmds, + 0, /* exported async functions */ + params, /* module parameters */ + 0, /* exported statistics */ + 0, /* exported MI functions */ + 0, /* exported pseudo-variables */ + 0, /* exported transformations */ + 0, /* extra processes */ + 0, /* module pre-initialization function */ + redis_mod_init, /* module initialization function */ + 0, /* response function */ + 0, /* destroy function */ + 0, /* per-child init function */ + 0 /* reload confirm function */ +}; + +static int redis_mod_init(void) +{ + if (rdb_mode_str) { + if (strcasecmp(rdb_mode_str, "auto") == 0) + rdb_mode = RDB_MODE_AUTO; + else if (strcasecmp(rdb_mode_str, "single") == 0) + rdb_mode = RDB_MODE_SINGLE; + else if (strcasecmp(rdb_mode_str, "cluster") == 0) + rdb_mode = RDB_MODE_CLUSTER; + else { + LM_ERR("invalid mode parameter <%s>, use one of " + "auto|single|cluster\n", rdb_mode_str); + return -1; + } + } + + if (rdb_connect_timeout <= 0 || rdb_query_timeout <= 0) { + LM_ERR("connect_timeout and query_timeout must be positive\n"); + return -1; + } + if (rdb_scan_count <= 0) + rdb_scan_count = 100; + + return 0; +} + +int db_redis_bind_api(const str* mod, db_func_t *dbb) +{ + if (dbb == NULL) + return -1; + + memset(dbb, 0, sizeof(db_func_t)); + + dbb->use_table = db_redis_use_table; + dbb->init = db_redis_init; + dbb->close = db_redis_close; + dbb->query = db_redis_query; + dbb->free_result = db_redis_free_result; + dbb->insert = db_redis_insert; + dbb->delete = db_redis_delete; + dbb->update = db_redis_update; + dbb->replace = db_redis_replace; + dbb->insert_update = db_redis_insert_update; + dbb->last_inserted_id = db_redis_last_inserted_id; + /* no raw_query / fetch_result / async in phase 1 - the core derives + * the exported capabilities from the functions set above */ + + return 0; +} diff --git a/modules/db_redis/db_redis.h b/modules/db_redis/db_redis.h new file mode 100644 index 00000000000..49150427dcb --- /dev/null +++ b/modules/db_redis/db_redis.h @@ -0,0 +1,36 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef DB_REDIS_H +#define DB_REDIS_H + +/* connection mode selection (modparam "mode") */ +#define RDB_MODE_AUTO 0 +#define RDB_MODE_SINGLE 1 +#define RDB_MODE_CLUSTER 2 + +extern int rdb_connect_timeout; /* ms */ +extern int rdb_query_timeout; /* ms */ +extern int rdb_scan_count; /* SCAN COUNT hint */ +extern int rdb_mode; /* RDB_MODE_* */ + +int db_redis_bind_api(const str* mod, db_func_t *dbb); + +#endif /* DB_REDIS_H */ diff --git a/modules/db_redis/dbase.c b/modules/db_redis/dbase.c new file mode 100644 index 00000000000..84173ff1642 --- /dev/null +++ b/modules/db_redis/dbase.c @@ -0,0 +1,1339 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include +#include +#include + +#include "../../dprint.h" +#include "../../mem/mem.h" +#include "../../ut.h" +#include "../../db/db.h" +#include "../../db/db_ut.h" +#include "../../db/db_pool.h" +#include "db_redis.h" +#include "redis_con.h" +#include "schema.h" +#include "dbase.h" + +#define RDB_KEY_MAX 320 /*
+ ':' + */ +#define RDB_NUM_MAX 32 /* printed numeric value */ + +enum rdb_opc { + RDB_OP_EQ, RDB_OP_LT, RDB_OP_GT, RDB_OP_LEQ, RDB_OP_GEQ, RDB_OP_NEQ +}; + +struct rdb_filter { + int col; /* schema column index */ + enum rdb_opc op; + const db_val_t *val; +}; + +/* one matched row while collecting; vals[i].s==NULL means SQL NULL, + * non-NULL pointers are individual pkg allocations (NUL-terminated) */ +struct rdb_irow { + struct rdb_irow *next; + str *vals; +}; + + +db_con_t* db_redis_init(const str* _url) +{ + return db_do_init(_url, (void *)db_redis_new_connection); +} + +void db_redis_close(db_con_t* _h) +{ + db_do_close(_h, db_redis_free_connection); +} + +int db_redis_use_table(db_con_t* _h, const str* _t) +{ + return db_use_table(_h, _t); +} + +int db_redis_free_result(db_con_t* _h, db_res_t* _r) +{ + if (!_r) + return -1; + return db_free_result(_r); +} + + +static int rdb_parse_opc(const db_op_t op, enum rdb_opc *out) +{ + if (!op || strcmp(op, OP_EQ) == 0) + *out = RDB_OP_EQ; + else if (strcmp(op, OP_LT) == 0) + *out = RDB_OP_LT; + else if (strcmp(op, OP_GT) == 0) + *out = RDB_OP_GT; + else if (strcmp(op, OP_LEQ) == 0) + *out = RDB_OP_LEQ; + else if (strcmp(op, OP_GEQ) == 0) + *out = RDB_OP_GEQ; + else if (strcmp(op, OP_NEQ) == 0) + *out = RDB_OP_NEQ; + else { + LM_ERR("unsupported operator <%s>\n", op); + return -1; + } + return 0; +} + + +/* render a db value as its storage string; numeric types print into buf */ +static int rdb_val2str(const db_val_t *v, str *out, char *buf) +{ + if (VAL_NULL(v)) { + out->s = NULL; + out->len = 0; + return 0; + } + + out->len = RDB_NUM_MAX; + out->s = buf; + + switch (VAL_TYPE(v)) { + case DB_INT: + if (db_int2str(VAL_INT(v), buf, &out->len) < 0) + return -1; + break; + case DB_BITMAP: + if (db_int2str((int)VAL_BITMAP(v), buf, &out->len) < 0) + return -1; + break; + case DB_BIGINT: + if (db_bigint2str(VAL_BIGINT(v), buf, &out->len) < 0) + return -1; + break; + case DB_DOUBLE: + if (db_double2str(VAL_DOUBLE(v), buf, &out->len) < 0) + return -1; + break; + case DB_DATETIME: + /* stored as a Unix timestamp, so it stays numerically + * comparable and sortable */ + if (db_bigint2str((long long)VAL_TIME(v), buf, &out->len) < 0) + return -1; + break; + case DB_STRING: + out->s = (char *)VAL_STRING(v); + out->len = out->s ? strlen(out->s) : 0; + break; + case DB_STR: + out->s = VAL_STR(v).s; + out->len = VAL_STR(v).len; + break; + case DB_BLOB: + out->s = VAL_BLOB(v).s; + out->len = VAL_BLOB(v).len; + break; + default: + LM_ERR("unsupported value type %d\n", VAL_TYPE(v)); + return -1; + } + return 0; +} + + +static inline int rdb_row_key(const str *table, const str *pkval, + char *buf, str *out) +{ + if (table->len + 1 + pkval->len >= RDB_KEY_MAX) { + LM_ERR("row key too long for table <%.*s>\n", + table->len, table->s); + return -1; + } + memcpy(buf, table->s, table->len); + buf[table->len] = ':'; + memcpy(buf + table->len + 1, pkval->s, pkval->len); + out->s = buf; + out->len = table->len + 1 + pkval->len; + return 0; +} + + +/* look up a field in a HGETALL reply; returns 0 and fills out when + * present, -1 when the field is absent (SQL NULL) */ +static int rdb_reply_field(redisReply *hg, const str *name, str *out) +{ + size_t i; + + for (i = 0; i+1 < hg->elements; i += 2) { + if (hg->element[i]->type != REDIS_REPLY_STRING) + continue; + if ((int)hg->element[i]->len == name->len && + memcmp(hg->element[i]->str, name->s, name->len) == 0) { + if (hg->element[i+1]->type != REDIS_REPLY_STRING) + return -1; + out->s = hg->element[i+1]->str; + out->len = hg->element[i+1]->len; + return 0; + } + } + return -1; +} + + +static inline int rdb_col_is_numeric(const struct rdb_col *col) +{ + return col->type == DB_INT || col->type == DB_BIGINT || + col->type == DB_DOUBLE || col->type == DB_DATETIME || + col->type == DB_BITMAP; +} + + +/* evaluate one filter against a fetched row; SQL comparison semantics: + * NULL compares false to everything, except "= NULL" (IS NULL) and + * "!= NULL" (IS NOT NULL) */ +static int rdb_eval_one(redisReply *hg, const struct rdb_schema *sch, + const struct rdb_filter *f) +{ + const struct rdb_col *col = &sch->cols[f->col]; + str field, want; + char buf[RDB_NUM_MAX]; + int c, have; + + have = (rdb_reply_field(hg, &col->name, &field) == 0); + + if (VAL_NULL(f->val)) { + if (f->op == RDB_OP_EQ) + return !have; + if (f->op == RDB_OP_NEQ) + return have; + return 0; + } + if (!have) + return f->op == RDB_OP_NEQ ? 0 : 0; + + if (rdb_val2str(f->val, &want, buf) < 0) + return 0; + + if (rdb_col_is_numeric(col)) { + if (col->type == DB_DOUBLE) { + double a = strtod(field.s, NULL); + double b = strtod(want.s, NULL); + c = (a < b) ? -1 : (a > b) ? 1 : 0; + } else { + long long a = strtoll(field.s, NULL, 10); + long long b = strtoll(want.s, NULL, 10); + c = (a < b) ? -1 : (a > b) ? 1 : 0; + } + } else { + int min = field.len < want.len ? field.len : want.len; + c = memcmp(field.s, want.s, min); + if (c == 0) + c = (field.len < want.len) ? -1 : + (field.len > want.len) ? 1 : 0; + } + + switch (f->op) { + case RDB_OP_EQ: return c == 0; + case RDB_OP_NEQ: return c != 0; + case RDB_OP_LT: return c < 0; + case RDB_OP_GT: return c > 0; + case RDB_OP_LEQ: return c <= 0; + case RDB_OP_GEQ: return c >= 0; + } + return 0; +} + + +static int rdb_eval(redisReply *hg, const struct rdb_schema *sch, + const struct rdb_filter *flt, int nf, int is_or) +{ + int i, m; + + if (nf == 0) + return 1; + + for (i = 0; i < nf; i++) { + m = rdb_eval_one(hg, sch, &flt[i]); + if (is_or && m) + return 1; + if (!is_or && !m) + return 0; + } + return is_or ? 0 : 1; +} + + +/* build the filter array out of _k/_op/_v; returns 0 or -1 */ +static int rdb_build_filters(const struct rdb_schema *sch, + const db_key_t* _k, const db_op_t* _op, const db_val_t* _v, + int _n, struct rdb_filter *flt) +{ + int i; + + for (i = 0; i < _n; i++) { + flt[i].col = rdb_schema_col(sch, _k[i]); + if (flt[i].col < 0) { + LM_ERR("unknown column <%.*s> in table <%.*s>\n", + _k[i]->len, _k[i]->s, sch->table.len, sch->table.s); + return -1; + } + if (rdb_parse_opc(_op ? _op[i] : NULL, &flt[i].op) < 0) + return -1; + flt[i].val = &_v[i]; + } + return 0; +} + + +/* single "pk = value" fast path detector: returns the filter index of + * the pk-EQ filter usable for direct addressing, or -1 */ +static int rdb_pk_fastpath(const struct rdb_schema *sch, + const struct rdb_filter *flt, int nf, int is_or) +{ + int i; + + if (nf == 0) + return -1; + /* with OR semantics a single condition behaves like AND */ + if (is_or && nf > 1) + return -1; + + for (i = 0; i < nf; i++) + if (flt[i].col == sch->pk && flt[i].op == RDB_OP_EQ && + !VAL_NULL(flt[i].val)) + return i; + return -1; +} + + +/* fetch one row hash by its key; returns reply (may be empty array + * when the row does not exist) or NULL on transport error */ +static redisReply *rdb_fetch_row(struct redis_con *con, const str *rowkey) +{ + const char *argv[2]; + size_t argvlen[2]; + redisReply *reply; + + argv[0] = "HGETALL"; argvlen[0] = 7; + argv[1] = rowkey->s; argvlen[1] = rowkey->len; + + reply = rdb_cmd_key(con, rowkey, 2, argv, argvlen); + if (reply && reply->type == REDIS_REPLY_ERROR) { + LM_ERR("HGETALL <%.*s> failed: %s\n", + rowkey->len, rowkey->s, reply->str); + freeReplyObject(reply); + return NULL; + } + return reply; +} + + +/* ---------------- table scanning ---------------- */ + +typedef int (*rdb_scan_cb)(struct redis_con *con, const str *rowkey, + redisReply *hg, void *arg); + +/* walk all rows of a table across all masters, invoking cb for each + * existing row hash; cb return <0 aborts the scan */ +static int rdb_scan_table(struct redis_con *con, const str *table, + rdb_scan_cb cb, void *arg) +{ + redis_node *node; + redisReply *reply, *hg; + char pattern[RDB_KEY_MAX], cursor[24]; + const char *argv[6]; + size_t argvlen[6], k; + char countbuf[16]; + str rowkey; + int rc; + + if (table->len + 2 >= RDB_KEY_MAX) { + LM_ERR("table name too long\n"); + return -1; + } + memcpy(pattern, table->s, table->len); + pattern[table->len] = ':'; + pattern[table->len+1] = '*'; + + snprintf(countbuf, sizeof countbuf, "%d", rdb_scan_count); + + for (node = con->nodes; node; node = node->next) { + strcpy(cursor, "0"); + do { + argv[0] = "SCAN"; argvlen[0] = 4; + argv[1] = cursor; argvlen[1] = strlen(cursor); + argv[2] = "MATCH"; argvlen[2] = 5; + argv[3] = pattern; argvlen[3] = table->len + 2; + argv[4] = "COUNT"; argvlen[4] = 5; + argv[5] = countbuf; argvlen[5] = strlen(countbuf); + + reply = rdb_cmd_node(con, node, 6, argv, argvlen); + if (!reply || reply->type != REDIS_REPLY_ARRAY || + reply->elements < 2 || + reply->element[0]->type != REDIS_REPLY_STRING || + reply->element[1]->type != REDIS_REPLY_ARRAY) { + LM_ERR("SCAN failed on %s:%u\n", node->host, node->port); + if (reply) + freeReplyObject(reply); + return -1; + } + + strncpy(cursor, reply->element[0]->str, sizeof(cursor)-1); + cursor[sizeof(cursor)-1] = 0; + + for (k = 0; k < reply->element[1]->elements; k++) { + if (reply->element[1]->element[k]->type != + REDIS_REPLY_STRING) + continue; + rowkey.s = reply->element[1]->element[k]->str; + rowkey.len = reply->element[1]->element[k]->len; + + hg = rdb_fetch_row(con, &rowkey); + if (!hg) + continue; + if (hg->type == REDIS_REPLY_ARRAY && hg->elements) { + rc = cb(con, &rowkey, hg, arg); + if (rc < 0) { + freeReplyObject(hg); + freeReplyObject(reply); + return -1; + } + } + freeReplyObject(hg); + } + freeReplyObject(reply); + } while (strcmp(cursor, "0") != 0); + } + + return 0; +} + + +/* ---------------- query ---------------- */ + +struct rdb_query_ctx { + const struct rdb_schema *sch; + const struct rdb_filter *flt; + int nf; + int is_or; + const int *rescols; /* schema col index per result column */ + int nrescols; + struct rdb_irow *head, *tail; + int count; +}; + +static void rdb_free_irows(struct rdb_irow *head, int ncols) +{ + struct rdb_irow *next; + int i; + + while (head) { + next = head->next; + for (i = 0; i < ncols; i++) + if (head->vals[i].s) + pkg_free(head->vals[i].s); + pkg_free(head); + head = next; + } +} + +/* copy the requested columns of a row-hash reply into an irow */ +static struct rdb_irow *rdb_extract_row(redisReply *hg, + const struct rdb_schema *sch, const int *rescols, int nrescols) +{ + struct rdb_irow *row; + str field; + int i; + + row = pkg_malloc(sizeof *row + nrescols * sizeof(str)); + if (!row) { + LM_ERR("no more pkg memory for result row\n"); + return NULL; + } + memset(row, 0, sizeof *row + nrescols * sizeof(str)); + row->vals = (str *)(row + 1); + + for (i = 0; i < nrescols; i++) { + if (rdb_reply_field(hg, &sch->cols[rescols[i]].name, &field) < 0) + continue; /* stays NULL */ + row->vals[i].s = pkg_malloc(field.len + 1); + if (!row->vals[i].s) { + LM_ERR("no more pkg memory for column value\n"); + rdb_free_irows(row, nrescols); + return NULL; + } + memcpy(row->vals[i].s, field.s, field.len); + row->vals[i].s[field.len] = 0; + row->vals[i].len = field.len; + } + return row; +} + +static int rdb_query_scan_cb(struct redis_con *con, const str *rowkey, + redisReply *hg, void *arg) +{ + struct rdb_query_ctx *ctx = (struct rdb_query_ctx *)arg; + struct rdb_irow *row; + + if (!rdb_eval(hg, ctx->sch, ctx->flt, ctx->nf, ctx->is_or)) + return 0; + + row = rdb_extract_row(hg, ctx->sch, ctx->rescols, ctx->nrescols); + if (!row) + return -1; + + if (ctx->tail) + ctx->tail->next = row; + else + ctx->head = row; + ctx->tail = row; + ctx->count++; + return 0; +} + +/* qsort context for ORDER BY (each process is single-threaded) */ +static int rdb_sort_idx; +static int rdb_sort_desc; +static int rdb_sort_numeric; + +static int rdb_sort_cmp(const void *pa, const void *pb) +{ + const struct rdb_irow *a = *(struct rdb_irow * const *)pa; + const struct rdb_irow *b = *(struct rdb_irow * const *)pb; + const str *va = &a->vals[rdb_sort_idx]; + const str *vb = &b->vals[rdb_sort_idx]; + int c, min; + + /* NULLs sort first */ + if (!va->s || !vb->s) + c = (!va->s && !vb->s) ? 0 : (!va->s ? -1 : 1); + else if (rdb_sort_numeric) { + double da = strtod(va->s, NULL), db_ = strtod(vb->s, NULL); + c = (da < db_) ? -1 : (da > db_) ? 1 : 0; + } else { + min = va->len < vb->len ? va->len : vb->len; + c = memcmp(va->s, vb->s, min); + if (c == 0) + c = (va->len < vb->len) ? -1 : (va->len > vb->len) ? 1 : 0; + } + return rdb_sort_desc ? -c : c; +} + +/* fill one db_res value from its stored string representation */ +static int rdb_fill_val(db_val_t *val, db_type_t type, str *stored) +{ + VAL_TYPE(val) = type; + VAL_FREE(val) = 0; + + if (!stored->s) { + VAL_NULL(val) = 1; + return 0; + } + VAL_NULL(val) = 0; + + switch (type) { + case DB_INT: + if (db_str2int(stored->s, &VAL_INT(val)) < 0) + goto badval; + break; + case DB_BIGINT: + if (db_str2bigint(stored->s, &VAL_BIGINT(val)) < 0) + goto badval; + break; + case DB_DOUBLE: + if (db_str2double(stored->s, &VAL_DOUBLE(val)) < 0) + goto badval; + break; + case DB_DATETIME: { + /* stored as a Unix timestamp (see rdb_val2str) */ + long long ts; + if (db_str2bigint(stored->s, &ts) < 0) + goto badval; + VAL_TIME(val) = (time_t)ts; + break; + } + case DB_STR: + VAL_STR(val).s = stored->s; + VAL_STR(val).len = stored->len; + VAL_FREE(val) = 1; + stored->s = NULL; /* ownership moved to the result */ + break; + case DB_BLOB: + VAL_BLOB(val).s = stored->s; + VAL_BLOB(val).len = stored->len; + VAL_FREE(val) = 1; + stored->s = NULL; + break; + default: + LM_ERR("unsupported column type %d\n", type); + VAL_NULL(val) = 1; + return -1; + } + return 0; + +badval: + LM_ERR("failed to parse stored value <%.*s> as type %d\n", + stored->len, stored->s, type); + VAL_NULL(val) = 1; + return -1; +} + + +int db_redis_query(const db_con_t* _h, const db_key_t* _k, const db_op_t* _op, + const db_val_t* _v, const db_key_t* _c, const int _n, const int _nc, + const db_key_t _o, db_res_t** _r) +{ + struct redis_con *con; + struct rdb_schema *sch; + struct rdb_filter *flt = NULL; + struct rdb_query_ctx ctx; + struct rdb_irow **sorted = NULL, *irow; + db_res_t *res = NULL; + redisReply *hg; + int *rescols = NULL; + int i, r, fp, is_or, order_idx; + char numbuf[RDB_NUM_MAX], keybuf[RDB_KEY_MAX]; + str pkval, rowkey, ocol; + char *p; + + if (!_h || !CON_TABLE(_h) || !_r) { + LM_ERR("invalid query parameters\n"); + return -1; + } + *_r = NULL; + memset(&ctx, 0, sizeof ctx); + + con = CON_REDIS(_h); + is_or = (_h->flags & CON_OR_OPERATOR) ? 1 : 0; + CON_OR_RESET(_h); + + rdb_maybe_refresh(con); + + sch = rdb_get_schema(con, CON_TABLE(_h)); + if (!sch) + return -1; + + /* the result column set */ + if (_c && _nc > 0) { + rescols = pkg_malloc(_nc * sizeof(int)); + if (!rescols) + goto oom; + for (i = 0; i < _nc; i++) { + rescols[i] = rdb_schema_col(sch, _c[i]); + if (rescols[i] < 0) { + LM_ERR("unknown column <%.*s> in table <%.*s>\n", + _c[i]->len, _c[i]->s, + sch->table.len, sch->table.s); + goto error; + } + } + ctx.nrescols = _nc; + } else { + rescols = pkg_malloc(sch->nr_cols * sizeof(int)); + if (!rescols) + goto oom; + for (i = 0; i < sch->nr_cols; i++) + rescols[i] = i; + ctx.nrescols = sch->nr_cols; + } + + if (_n > 0) { + flt = pkg_malloc(_n * sizeof *flt); + if (!flt) + goto oom; + if (rdb_build_filters(sch, _k, _op, _v, _n, flt) < 0) + goto error; + } + + ctx.sch = sch; + ctx.flt = flt; + ctx.nf = _n; + ctx.is_or = is_or; + ctx.rescols = rescols; + + fp = flt ? rdb_pk_fastpath(sch, flt, _n, is_or) : -1; + if (fp >= 0) { + /* direct row addressing by primary key */ + if (rdb_val2str(flt[fp].val, &pkval, numbuf) < 0 || + rdb_row_key(CON_TABLE(_h), &pkval, keybuf, &rowkey) < 0) + goto error; + + hg = rdb_fetch_row(con, &rowkey); + if (!hg) + goto error; + if (hg->type == REDIS_REPLY_ARRAY && hg->elements) { + if (rdb_query_scan_cb(con, &rowkey, hg, &ctx) < 0) { + freeReplyObject(hg); + goto error; + } + } + freeReplyObject(hg); + } else { + if (rdb_scan_table(con, CON_TABLE(_h), + rdb_query_scan_cb, &ctx) < 0) + goto error; + } + + /* ORDER BY: single column, optional ASC/DESC */ + if (_o && _o->s && ctx.count > 1) { + ocol.s = _o->s; + p = memchr(_o->s, ' ', _o->len); + ocol.len = p ? (int)(p - _o->s) : _o->len; + + order_idx = -1; + for (i = 0; i < ctx.nrescols; i++) + if (sch->cols[rescols[i]].name.len == ocol.len && + memcmp(sch->cols[rescols[i]].name.s, ocol.s, ocol.len) == 0) { + order_idx = i; + break; + } + if (order_idx < 0) { + LM_WARN("order-by column <%.*s> not in the result set, " + "ignoring ordering\n", ocol.len, ocol.s); + } else { + rdb_sort_idx = order_idx; + rdb_sort_numeric = + rdb_col_is_numeric(&sch->cols[rescols[order_idx]]); + rdb_sort_desc = 0; + if (p) { + while (p < _o->s + _o->len && *p == ' ') p++; + if (_o->s + _o->len - p >= 4 && + strncasecmp(p, "DESC", 4) == 0) + rdb_sort_desc = 1; + } + + sorted = pkg_malloc(ctx.count * sizeof *sorted); + if (!sorted) + goto oom; + for (irow = ctx.head, i = 0; irow; irow = irow->next) + sorted[i++] = irow; + qsort(sorted, ctx.count, sizeof *sorted, rdb_sort_cmp); + /* relink in sorted order */ + for (i = 0; i < ctx.count-1; i++) + sorted[i]->next = sorted[i+1]; + sorted[ctx.count-1]->next = NULL; + ctx.head = sorted[0]; + pkg_free(sorted); + sorted = NULL; + } + } + + /* build the db_res_t */ + res = db_new_result(); + if (!res) + goto oom; + RES_COL_N(res) = ctx.nrescols; + if (db_allocate_columns(res, ctx.nrescols) < 0) + goto oom; + for (i = 0; i < ctx.nrescols; i++) { + RES_NAMES(res)[i] = &sch->cols[rescols[i]].name; + RES_TYPES(res)[i] = sch->cols[rescols[i]].type; + } + + if (ctx.count > 0) { + if (db_allocate_rows(res, ctx.count) < 0) + goto oom; + for (irow = ctx.head, r = 0; irow; irow = irow->next, r++) { + ROW_N(&RES_ROWS(res)[r]) = ctx.nrescols; + for (i = 0; i < ctx.nrescols; i++) + rdb_fill_val(&ROW_VALUES(&RES_ROWS(res)[r])[i], + sch->cols[rescols[i]].type, &irow->vals[i]); + } + } + RES_ROW_N(res) = ctx.count; + RES_NUM_ROWS(res) = ctx.count; + RES_LAST_ROW(res) = ctx.count; + + rdb_free_irows(ctx.head, ctx.nrescols); + if (flt) + pkg_free(flt); + pkg_free(rescols); + + *_r = res; + return 0; + +oom: + LM_ERR("no more pkg memory while running query\n"); +error: + if (res) + db_free_result(res); + if (sorted) + pkg_free(sorted); + rdb_free_irows(ctx.head, ctx.nrescols); + if (flt) + pkg_free(flt); + if (rescols) + pkg_free(rescols); + return -1; +} + + +/* ---------------- write operations ---------------- */ + +/* fetch the next auto-increment value for a table */ +static int rdb_next_id(struct redis_con *con, const str *table, + char *numbuf, str *out) +{ + char keybuf[RDB_KEY_MAX]; + str key; + const char *argv[2]; + size_t argvlen[2]; + redisReply *reply; + + if (table->len + 4 >= RDB_KEY_MAX) + return -1; + memcpy(keybuf, "seq:", 4); + memcpy(keybuf + 4, table->s, table->len); + key.s = keybuf; + key.len = 4 + table->len; + + argv[0] = "INCR"; argvlen[0] = 4; + argv[1] = key.s; argvlen[1] = key.len; + + reply = rdb_cmd_key(con, &key, 2, argv, argvlen); + if (!reply || reply->type != REDIS_REPLY_INTEGER) { + LM_ERR("INCR on <%.*s> failed\n", key.len, key.s); + if (reply) + freeReplyObject(reply); + return -1; + } + con->last_insert_id = reply->integer; + out->len = RDB_NUM_MAX; + out->s = numbuf; + if (db_bigint2str(reply->integer, numbuf, &out->len) < 0) { + freeReplyObject(reply); + return -1; + } + freeReplyObject(reply); + return 0; +} + + +#define RDB_STORE_INSERT 0 /* fail when the row already exists */ +#define RDB_STORE_REPLACE 1 /* delete any existing row first */ +#define RDB_STORE_MERGE 2 /* merge fields into any existing row */ + +static int rdb_store_row(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n, int store_mode) +{ + struct redis_con *con; + struct rdb_schema *sch; + redisReply *reply; + const char **argv = NULL; + size_t *argvlen = NULL; + char (*numbufs)[RDB_NUM_MAX] = NULL; + char keybuf[RDB_KEY_MAX], pknum[RDB_NUM_MAX]; + str pkval = STR_NULL, rowkey, v; + int i, c, argc, pk_arg = -1, rc = -1; + + if (!_h || !CON_TABLE(_h) || !_k || !_v || _n <= 0) { + LM_ERR("invalid insert parameters\n"); + return -1; + } + + con = CON_REDIS(_h); + rdb_maybe_refresh(con); + + sch = rdb_get_schema(con, CON_TABLE(_h)); + if (!sch) + return -1; + + /* locate the primary key among the given columns */ + for (i = 0; i < _n; i++) { + c = rdb_schema_col(sch, _k[i]); + if (c < 0) { + LM_ERR("unknown column <%.*s> in table <%.*s>\n", + _k[i]->len, _k[i]->s, sch->table.len, sch->table.s); + return -1; + } + if (c == sch->pk && !VAL_NULL(&_v[i])) + pk_arg = i; + } + + if (pk_arg >= 0) { + if (rdb_val2str(&_v[pk_arg], &pkval, pknum) < 0) + return -1; + } else { + if (!sch->cols[sch->pk].is_auto) { + LM_ERR("no value for primary key <%.*s> of table <%.*s>\n", + sch->cols[sch->pk].name.len, sch->cols[sch->pk].name.s, + sch->table.len, sch->table.s); + return -1; + } + if (rdb_next_id(con, CON_TABLE(_h), pknum, &pkval) < 0) + return -1; + } + + if (rdb_row_key(CON_TABLE(_h), &pkval, keybuf, &rowkey) < 0) + return -1; + + argv = pkg_malloc((4 + 2*_n) * sizeof *argv); + argvlen = pkg_malloc((4 + 2*_n) * sizeof *argvlen); + numbufs = pkg_malloc(_n * sizeof *numbufs); + if (!argv || !argvlen || !numbufs) { + LM_ERR("no more pkg memory for insert\n"); + goto out; + } + + if (store_mode == RDB_STORE_REPLACE) { + argv[0] = "DEL"; argvlen[0] = 3; + argv[1] = rowkey.s; argvlen[1] = rowkey.len; + reply = rdb_cmd_key(con, &rowkey, 2, argv, argvlen); + if (!reply) { + LM_ERR("failed to clear row for replace\n"); + goto out; + } + freeReplyObject(reply); + } + + if (store_mode != RDB_STORE_MERGE) { + /* reserve the row: HSETNX doubles as + * both the uniqueness check and the guaranteed first field */ + argv[0] = "HSETNX"; argvlen[0] = 6; + argv[1] = rowkey.s; argvlen[1] = rowkey.len; + argv[2] = sch->cols[sch->pk].name.s; + argvlen[2] = sch->cols[sch->pk].name.len; + argv[3] = pkval.s; argvlen[3] = pkval.len; + + reply = rdb_cmd_key(con, &rowkey, 4, argv, argvlen); + if (!reply || reply->type != REDIS_REPLY_INTEGER) { + LM_ERR("row reservation failed for <%.*s>\n", + rowkey.len, rowkey.s); + if (reply) + freeReplyObject(reply); + goto out; + } + if (reply->integer == 0) { + LM_ERR("duplicate primary key <%.*s> in table <%.*s>\n", + pkval.len, pkval.s, sch->table.len, sch->table.s); + freeReplyObject(reply); + goto out; + } + freeReplyObject(reply); + } + + /* store the remaining non-NULL fields */ + argc = 2; + argv[0] = "HSET"; argvlen[0] = 4; + argv[1] = rowkey.s; argvlen[1] = rowkey.len; + + if (store_mode == RDB_STORE_MERGE) { + /* the pk field is written here instead of via HSETNX */ + argv[argc] = sch->cols[sch->pk].name.s; + argvlen[argc] = sch->cols[sch->pk].name.len; + argc++; + argv[argc] = pkval.s; + argvlen[argc] = pkval.len; + argc++; + } + + for (i = 0; i < _n; i++) { + if (i == pk_arg || VAL_NULL(&_v[i])) + continue; + if (rdb_val2str(&_v[i], &v, numbufs[i]) < 0) + goto out; + argv[argc] = _k[i]->s; + argvlen[argc] = _k[i]->len; + argc++; + argv[argc] = v.s ? v.s : ""; + argvlen[argc] = v.len; + argc++; + } + + if (argc > 2) { + reply = rdb_cmd_key(con, &rowkey, argc, argv, argvlen); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LM_ERR("failed to store row <%.*s> (%s)\n", + rowkey.len, rowkey.s, reply ? reply->str : "io error"); + if (reply) + freeReplyObject(reply); + goto out; + } + freeReplyObject(reply); + } + + rc = 0; +out: + if (argv) pkg_free(argv); + if (argvlen) pkg_free(argvlen); + if (numbufs) pkg_free(numbufs); + return rc; +} + + +int db_redis_insert(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n) +{ + return rdb_store_row(_h, _k, _v, _n, RDB_STORE_INSERT); +} + +int db_redis_replace(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n) +{ + return rdb_store_row(_h, _k, _v, _n, RDB_STORE_REPLACE); +} + +int db_redis_insert_update(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n) +{ + return rdb_store_row(_h, _k, _v, _n, RDB_STORE_MERGE); +} + +int db_redis_last_inserted_id(const db_con_t* _h) +{ + if (!_h) + return -1; + return (int)CON_REDIS(_h)->last_insert_id; +} + + +/* ---------------- delete ---------------- */ + +static int rdb_del_key(struct redis_con *con, const str *rowkey) +{ + const char *argv[2]; + size_t argvlen[2]; + redisReply *reply; + + argv[0] = "DEL"; argvlen[0] = 3; + argv[1] = rowkey->s; argvlen[1] = rowkey->len; + + reply = rdb_cmd_key(con, rowkey, 2, argv, argvlen); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LM_ERR("DEL <%.*s> failed\n", rowkey->len, rowkey->s); + if (reply) + freeReplyObject(reply); + return -1; + } + freeReplyObject(reply); + return 0; +} + +struct rdb_del_ctx { + const struct rdb_schema *sch; + const struct rdb_filter *flt; + int nf; + int is_or; + int failed; +}; + +static int rdb_delete_scan_cb(struct redis_con *con, const str *rowkey, + redisReply *hg, void *arg) +{ + struct rdb_del_ctx *ctx = (struct rdb_del_ctx *)arg; + + if (!rdb_eval(hg, ctx->sch, ctx->flt, ctx->nf, ctx->is_or)) + return 0; + if (rdb_del_key(con, rowkey) < 0) + ctx->failed = 1; + return 0; +} + +int db_redis_delete(const db_con_t* _h, const db_key_t* _k, + const db_op_t* _o, const db_val_t* _v, const int _n) +{ + struct redis_con *con; + struct rdb_schema *sch; + struct rdb_filter *flt = NULL; + struct rdb_del_ctx ctx; + char numbuf[RDB_NUM_MAX], keybuf[RDB_KEY_MAX]; + str pkval, rowkey; + int i, is_or, all_pk_eq, rc = -1; + + if (!_h || !CON_TABLE(_h)) { + LM_ERR("invalid delete parameters\n"); + return -1; + } + + con = CON_REDIS(_h); + is_or = (_h->flags & CON_OR_OPERATOR) ? 1 : 0; + CON_OR_RESET(_h); + + rdb_maybe_refresh(con); + + sch = rdb_get_schema(con, CON_TABLE(_h)); + if (!sch) + return -1; + + if (_n > 0) { + flt = pkg_malloc(_n * sizeof *flt); + if (!flt) { + LM_ERR("no more pkg memory for delete\n"); + return -1; + } + if (rdb_build_filters(sch, _k, _o, _v, _n, flt) < 0) + goto out; + + /* pk-only deletes go straight to the row keys: a single + * "pk=X", or an OR-list of "pk=X" conditions (bulk delete) */ + all_pk_eq = 1; + for (i = 0; i < _n; i++) + if (flt[i].col != sch->pk || flt[i].op != RDB_OP_EQ || + VAL_NULL(flt[i].val)) { + all_pk_eq = 0; + break; + } + + if (all_pk_eq && (_n == 1 || is_or)) { + for (i = 0; i < _n; i++) { + if (rdb_val2str(flt[i].val, &pkval, numbuf) < 0 || + rdb_row_key(CON_TABLE(_h), &pkval, keybuf, + &rowkey) < 0) + goto out; + if (rdb_del_key(con, &rowkey) < 0) + goto out; + } + rc = 0; + goto out; + } + } + + ctx.sch = sch; + ctx.flt = flt; + ctx.nf = _n; + ctx.is_or = is_or; + ctx.failed = 0; + + if (rdb_scan_table(con, CON_TABLE(_h), rdb_delete_scan_cb, &ctx) < 0) + goto out; + + rc = ctx.failed ? -1 : 0; +out: + if (flt) + pkg_free(flt); + return rc; +} + + +/* ---------------- update ---------------- */ + +struct rdb_upd_ctx { + const struct rdb_schema *sch; + const struct rdb_filter *flt; + int nf; + int is_or; + /* collected row keys (own pkg copies) */ + str *keys; + int count; + int alloc; +}; + +static int rdb_update_scan_cb(struct redis_con *con, const str *rowkey, + redisReply *hg, void *arg) +{ + struct rdb_upd_ctx *ctx = (struct rdb_upd_ctx *)arg; + str *newkeys; + + if (!rdb_eval(hg, ctx->sch, ctx->flt, ctx->nf, ctx->is_or)) + return 0; + + if (ctx->count == ctx->alloc) { + ctx->alloc = ctx->alloc ? 2*ctx->alloc : 16; + newkeys = pkg_realloc(ctx->keys, ctx->alloc * sizeof(str)); + if (!newkeys) { + LM_ERR("no more pkg memory for update targets\n"); + return -1; + } + ctx->keys = newkeys; + } + ctx->keys[ctx->count].s = pkg_malloc(rowkey->len); + if (!ctx->keys[ctx->count].s) { + LM_ERR("no more pkg memory for update target key\n"); + return -1; + } + memcpy(ctx->keys[ctx->count].s, rowkey->s, rowkey->len); + ctx->keys[ctx->count].len = rowkey->len; + ctx->count++; + return 0; +} + +/* apply the SET clause to one row: HSET the non-NULL values, + * HDEL the NULL ones */ +static int rdb_apply_update(struct redis_con *con, const str *rowkey, + const db_key_t* _uk, const db_val_t* _uv, int _un, + char (*numbufs)[RDB_NUM_MAX]) +{ + const char *argv[2 + 2*_un]; + size_t argvlen[2 + 2*_un]; + redisReply *reply; + str v; + int i, argc; + + /* HSET phase */ + argc = 2; + argv[0] = "HSET"; argvlen[0] = 4; + argv[1] = rowkey->s; argvlen[1] = rowkey->len; + for (i = 0; i < _un; i++) { + if (VAL_NULL(&_uv[i])) + continue; + if (rdb_val2str(&_uv[i], &v, numbufs[i]) < 0) + return -1; + argv[argc] = _uk[i]->s; + argvlen[argc] = _uk[i]->len; + argc++; + argv[argc] = v.s ? v.s : ""; + argvlen[argc] = v.len; + argc++; + } + if (argc > 2) { + reply = rdb_cmd_key(con, rowkey, argc, argv, argvlen); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LM_ERR("failed to update row <%.*s> (%s)\n", rowkey->len, + rowkey->s, reply ? reply->str : "io error"); + if (reply) + freeReplyObject(reply); + return -1; + } + freeReplyObject(reply); + } + + /* HDEL phase for NULL assignments */ + argc = 2; + argv[0] = "HDEL"; argvlen[0] = 4; + argv[1] = rowkey->s; argvlen[1] = rowkey->len; + for (i = 0; i < _un; i++) { + if (!VAL_NULL(&_uv[i])) + continue; + argv[argc] = _uk[i]->s; + argvlen[argc] = _uk[i]->len; + argc++; + } + if (argc > 2) { + reply = rdb_cmd_key(con, rowkey, argc, argv, argvlen); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LM_ERR("failed to clear columns of row <%.*s>\n", + rowkey->len, rowkey->s); + if (reply) + freeReplyObject(reply); + return -1; + } + freeReplyObject(reply); + } + + return 0; +} + +int db_redis_update(const db_con_t* _h, const db_key_t* _k, + const db_op_t* _o, const db_val_t* _v, const db_key_t* _uk, + const db_val_t* _uv, const int _n, const int _un) +{ + struct redis_con *con; + struct rdb_schema *sch; + struct rdb_filter *flt = NULL; + struct rdb_upd_ctx ctx; + redisReply *hg; + char (*numbufs)[RDB_NUM_MAX] = NULL; + char numbuf[RDB_NUM_MAX], keybuf[RDB_KEY_MAX]; + str pkval, rowkey; + int i, c, fp, is_or, rc = -1; + + if (!_h || !CON_TABLE(_h) || !_uk || !_uv || _un <= 0) { + LM_ERR("invalid update parameters\n"); + return -1; + } + + con = CON_REDIS(_h); + is_or = (_h->flags & CON_OR_OPERATOR) ? 1 : 0; + CON_OR_RESET(_h); + + rdb_maybe_refresh(con); + + sch = rdb_get_schema(con, CON_TABLE(_h)); + if (!sch) + return -1; + + /* validate the SET columns; updating the pk would require a key + * rename and is not supported */ + for (i = 0; i < _un; i++) { + c = rdb_schema_col(sch, _uk[i]); + if (c < 0) { + LM_ERR("unknown column <%.*s> in table <%.*s>\n", + _uk[i]->len, _uk[i]->s, sch->table.len, sch->table.s); + return -1; + } + if (c == sch->pk) { + LM_ERR("updating the primary key column <%.*s> is " + "not supported\n", _uk[i]->len, _uk[i]->s); + return -1; + } + } + + memset(&ctx, 0, sizeof ctx); + ctx.sch = sch; + ctx.is_or = is_or; + + if (_n > 0) { + flt = pkg_malloc(_n * sizeof *flt); + if (!flt) { + LM_ERR("no more pkg memory for update\n"); + return -1; + } + if (rdb_build_filters(sch, _k, _o, _v, _n, flt) < 0) + goto out; + ctx.flt = flt; + ctx.nf = _n; + } + + numbufs = pkg_malloc(_un * sizeof *numbufs); + if (!numbufs) { + LM_ERR("no more pkg memory for update values\n"); + goto out; + } + + fp = flt ? rdb_pk_fastpath(sch, flt, _n, is_or) : -1; + if (fp >= 0) { + if (rdb_val2str(flt[fp].val, &pkval, numbuf) < 0 || + rdb_row_key(CON_TABLE(_h), &pkval, keybuf, &rowkey) < 0) + goto out; + + hg = rdb_fetch_row(con, &rowkey); + if (!hg) + goto out; + if (hg->type == REDIS_REPLY_ARRAY && hg->elements && + rdb_eval(hg, sch, flt, _n, is_or)) { + freeReplyObject(hg); + if (rdb_apply_update(con, &rowkey, _uk, _uv, _un, + numbufs) < 0) + goto out; + } else { + /* no matching row - not an error, zero rows updated */ + freeReplyObject(hg); + } + rc = 0; + goto out; + } + + if (rdb_scan_table(con, CON_TABLE(_h), rdb_update_scan_cb, &ctx) < 0) + goto out; + + rc = 0; + for (i = 0; i < ctx.count; i++) + if (rdb_apply_update(con, &ctx.keys[i], _uk, _uv, _un, + numbufs) < 0) + rc = -1; + +out: + for (i = 0; i < ctx.count; i++) + if (ctx.keys[i].s) + pkg_free(ctx.keys[i].s); + if (ctx.keys) + pkg_free(ctx.keys); + if (numbufs) + pkg_free(numbufs); + if (flt) + pkg_free(flt); + return rc; +} diff --git a/modules/db_redis/dbase.h b/modules/db_redis/dbase.h new file mode 100644 index 00000000000..d680619e4d2 --- /dev/null +++ b/modules/db_redis/dbase.h @@ -0,0 +1,57 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef DB_REDIS_DBASE_H +#define DB_REDIS_DBASE_H + +#include "../../db/db_con.h" +#include "../../db/db_res.h" +#include "../../db/db_key.h" +#include "../../db/db_op.h" +#include "../../db/db_val.h" + +db_con_t* db_redis_init(const str* _url); +void db_redis_close(db_con_t* _h); +int db_redis_use_table(db_con_t* _h, const str* _t); +int db_redis_free_result(db_con_t* _h, db_res_t* _r); + +int db_redis_query(const db_con_t* _h, const db_key_t* _k, const db_op_t* _op, + const db_val_t* _v, const db_key_t* _c, const int _n, const int _nc, + const db_key_t _o, db_res_t** _r); + +int db_redis_insert(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n); + +int db_redis_delete(const db_con_t* _h, const db_key_t* _k, + const db_op_t* _o, const db_val_t* _v, const int _n); + +int db_redis_update(const db_con_t* _h, const db_key_t* _k, + const db_op_t* _o, const db_val_t* _v, const db_key_t* _uk, + const db_val_t* _uv, const int _n, const int _un); + +int db_redis_replace(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n); + +int db_redis_insert_update(const db_con_t* _h, const db_key_t* _k, + const db_val_t* _v, const int _n); + +int db_redis_last_inserted_id(const db_con_t* _h); + +#endif /* DB_REDIS_DBASE_H */ diff --git a/modules/db_redis/doc/contributors.xml b/modules/db_redis/doc/contributors.xml new file mode 100644 index 00000000000..f4e23055b44 --- /dev/null +++ b/modules/db_redis/doc/contributors.xml @@ -0,0 +1,79 @@ + + + &contributors; + +
+ By Commit Statistics + +
Top contributors by DevScore<superscript>(1)</superscript>, authored commits<superscript>(2)</superscript> and lines added/removed<superscript>(3)</superscript> + + + + + Name + DevScore + Commits + Lines ++ + Lines -- + + + + + 1. + Yury Kirsanov + 27 + 2 + 2802 + 4 + + + +
+ + + (1) DevScore = author_commits + author_lines_added / (project_lines_added / project_commits) + author_lines_deleted / (project_lines_deleted / project_commits) + + + (2) including any documentation-related commits, excluding merge commits. Regarding imported patches/code, we do our best to count the work on behalf of the proper owner, as per the "fix_authors" and "mod_renames" arrays in opensips/docs/build-contrib.sh. If you identify any patches/commits which do not get properly attributed to you, please submit a pull request which extends "fix_authors" and/or "mod_renames". + + + (3) ignoring whitespace edits, renamed files and auto-generated files + + + +
+ By Commit Activity + + Most recently active contributors<superscript>(1)</superscript> to this module + + + + + Name + Commit Activity + + + + + 1. + Yury Kirsanov + Jul 2026 - Jul 2026 + + + +
+ + + (1) including any documentation-related commits, excluding merge commits + +
+ + + + Documentation +
+ Contributors + +
+ + diff --git a/modules/db_redis/doc/db_redis.xml b/modules/db_redis/doc/db_redis.xml new file mode 100644 index 00000000000..495263444a8 --- /dev/null +++ b/modules/db_redis/doc/db_redis.xml @@ -0,0 +1,30 @@ + + + + + + + +%docentities; + +]> + + + + db_redis Module + &osipsname; + + + + &admin; + &faq; + &contrib; + + &docCopyrights; + ©right; 2026 &osipssol; + + diff --git a/modules/db_redis/doc/db_redis_admin.xml b/modules/db_redis/doc/db_redis_admin.xml new file mode 100644 index 00000000000..4c700cda614 --- /dev/null +++ b/modules/db_redis/doc/db_redis_admin.xml @@ -0,0 +1,393 @@ + + + + + &adminguide; + +
+ Overview + + This module provides a &osips; DB API backend that stores data in + a Redis server or a Redis + Cluster. It implements the structured database interface used by + modules such as dialog, so that their tables + can be persisted in Redis instead of a traditional SQL database. + + + Unlike the cachedb_redis module, which exposes + Redis through the key-value NoSQL interface, db_redis + presents Redis as a relational-style backend behind the standard + db_url mechanism. A module is switched to Redis + simply by pointing its db_url at a + redis:// URL; no code change is required in the + consuming module. + + + The connection type is auto-detected at startup: the module works + either against a single Redis endpoint (a standalone server, or the + current master reached through an external load balancer/Sentinel + setup) or against a fully-enabled Redis Cluster, where it discovers + the whole topology and keeps a connection open to every master + shard. + +
+ +
+ Data Model + + Redis is schemaless, so the module needs the table layout to be + declared out of band. Both the schema definitions and the data are + stored as Redis hashes: + + + + + Schema - the hash + schema:<table> describes one table. It holds + a __cols field with the ordered, space-separated + list of column names, a __pk field naming the + primary-key column, and one field per column giving its type as + type[,null][,auto]. The supported types are + int, bigint, + double, string, + blob and datetime. The + null attribute marks a nullable column and + auto marks an auto-increment primary key. + + + + + Rows - each row is the hash + <table>:<primary key value>, with one + field per non-NULL column. A NULL column is represented by the + absence of the field. Values of blob columns + are stored verbatim (binary-safe), with no encoding overhead. + + + + + Sequences - auto-increment primary keys are + generated with INCR on the counter + seq:<table>. + + + + + Table version - the standard &osips; + version table is provisioned like any other table, + with one row per served table, for example + HSET version:dialog table_name dialog table_version 12. + + + + + The key prefixes schema: and seq: are + reserved by the module and must not be used as table names. Ready-made + provisioning files are shipped under the scripts/redis/ + directory; see that directory's README for details. A schema is + loaded from Redis on first access to a table and cached for the + lifetime of the connection. + +
+ +
+ Deployment Modes + + At connection time the module issues an INFO CLUSTER + command and inspects the cluster_enabled field to + decide how to operate. The detected behaviour can be pinned with the + parameter. + +
+ Single endpoint + + Used for a standalone Redis server, or when a Redis Sentinel + deployment is fronted by an external load balancer that always + routes to the current master. The module keeps a single + connection and transparently reconnects on I/O errors or when it + receives a -READONLY reply (which indicates it is + talking to a demoted master after a failover). Sentinel discovery + and master election are expected to be handled externally. + +
+
+ Redis Cluster + + When the endpoint is a cluster node, the module fetches the full + topology (via CLUSTER SHARDS, falling back to + CLUSTER SLOTS on older servers) starting from the + single seed host given in db_url. It then opens + and keeps warm a connection to every master shard, so that a key + routed to any shard can be served immediately over an already + established connection rather than paying a new TCP handshake on + each redirect. + + + Each key is routed to its shard by computing the CRC16 hash slot + (honouring {hash tag} braces). MOVED + redirects are served from the warm connection pool and trigger a + deferred topology refresh; ASK redirects are + followed with a one-shot ASKING; and + TRYAGAIN replies (seen during slot migration) are + retried after a short back-off. Full-table operations scan every + master and merge the results. + + + Only master nodes are used; the module always reads from and + writes to masters to guarantee read-your-writes consistency. + +
+
+ +
+ Dependencies +
+ &osips; Modules + + The following modules must be loaded before this module: + + + + No dependencies on other &osips; modules. + + + + +
+
+ External Libraries or Applications + + The following libraries or applications must be installed before running + &osips; with this module loaded: + + + + hiredis - the minimalistic Redis client + C library (development package, typically + libhiredis-dev). + + + + +
+
+ +
+ Exported Parameters + +
+ <varname>connect_timeout</varname> (integer) + + The timeout, in milliseconds, for establishing a TCP connection to a + Redis node. It also bounds how long a failover blackout can stall an + operation before it is retried or failed. + + + Default value is 1000 (1 second). + + + Set <varname>connect_timeout</varname> parameter + +... +modparam("db_redis", "connect_timeout", 500) +... + + +
+ +
+ <varname>query_timeout</varname> (integer) + + The send/receive timeout, in milliseconds, applied to commands issued + on an established connection. Keep it small enough that a stalled node + does not block &osips; worker processes for long, especially when the + dialog module is used in realtime db_mode. + + + Default value is 2000 (2 seconds). + + + Set <varname>query_timeout</varname> parameter + +... +modparam("db_redis", "query_timeout", 1000) +... + + +
+ +
+ <varname>scan_count</varname> (integer) + + The COUNT hint passed to the Redis SCAN + command when iterating a whole table (used by full-table loads and by + queries that are not addressed by primary key). Larger values reduce + the number of round-trips at the cost of larger individual replies. + + + Default value is 100. + + + Set <varname>scan_count</varname> parameter + +... +modparam("db_redis", "scan_count", 500) +... + + +
+ +
+ <varname>mode</varname> (string) + + Pins the connection mode instead of auto-detecting it. Accepted + values are: + + + + auto - detect single vs. cluster from + INFO CLUSTER (the default). + + + single - force single-endpoint mode; + startup fails if the endpoint is a cluster node. + + + cluster - force cluster mode; startup + fails if the endpoint has cluster support disabled. + + + + Pinning the mode turns an accidental misconfiguration (for example a + db_url pointing at the wrong kind of server) into a + clear startup error instead of a silently wrong runtime behaviour. + + + Default value is "auto". + + + Set <varname>mode</varname> parameter + +... +modparam("db_redis", "mode", "cluster") +... + + +
+ +
+ +
+ Exported Functions + + No function exported to be used from the configuration file. + +
+ +
+ Usage + + The module is used by pointing a consuming module's + db_url at a redis:// URL. The + URL has the standard &osips; form + redis://[:password@]host[:port]/db. The database path + component is mandatory for the URL parser but is only meaningful for a + standalone server (it selects the logical database number); it is + ignored in cluster mode. A password-only Redis is authenticated with + the :password@ form. + + + Persisting the dialog table in a single Redis server + +... +loadmodule "db_redis/db_redis.so" +loadmodule "dialog/dialog.so" + +modparam("dialog", "db_mode", 1) +modparam("dialog", "db_url", "redis://127.0.0.1:6379/0") +... + + + + Persisting the dialog table in a Redis Cluster + +... +loadmodule "db_redis/db_redis.so" +loadmodule "dialog/dialog.so" + +modparam("dialog", "db_mode", 1) +# a single cluster node is enough as a seed; the rest of the topology +# is discovered automatically +modparam("dialog", "db_url", "redis://:secretpass@10.0.0.20:6379/0") +... + + + + Before starting &osips;, provision the table schemas into Redis, for + example (add -c when talking to a cluster): + + +redis-cli [-c] < scripts/redis/dialog-create.redis + +
+ +
+ Limitations + + + Queries support equality and the ordering operators + (=, <, >, + <=, >=, !=) on + any column, combined with AND or OR, plus a single + ORDER BY column. Lookups addressed by the primary + key are served directly; every other query performs a table scan + and filters in the client. There is no JOIN support + (the &osips; DB API does not provide one) and no raw-query + support. + + + Updating the value of the primary-key column of an existing row + is not supported. + + + The module is intended for modules that use the structured DB API + with per-row primary-key access, such as + dialog, dispatcher, + clusterer and usrloc. + Raw SQL queries are not implemented; a module that relies on them + works only if it also offers a structured-query path (as + usrloc does for contact loading). + + + As with the SQL backends, sharing rows between several &osips; + instances through the same Redis store requires the instances to + use a matching configuration (for the dialog table, an identical + hash_size and matching listening sockets). + + +
+ +
+ Installation + + Because it depends on an external library, the db_redis module is not + compiled and installed by default. You can use one of the next + options. + + + + + - edit the "Makefile" and remove "db_redis" from + "excluded_modules" list. Then follow the standard procedure to + install &osips;: "make all; make install". + + + + + - from command line use: 'make all include_modules="db_redis"; + make install include_modules="db_redis"'. + + + +
+ diff --git a/modules/db_redis/redis_con.c b/modules/db_redis/redis_con.c new file mode 100644 index 00000000000..dc0beba525b --- /dev/null +++ b/modules/db_redis/redis_con.c @@ -0,0 +1,826 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include +#include + +#include "../../dprint.h" +#include "../../mem/mem.h" +#include "../../ut.h" +#include "../../db/db.h" +#include "db_redis.h" +#include "redis_con.h" +#include "schema.h" + +#define RDB_MAX_REDIRECTS 3 + +/* CRC16 (CCITT) as used by Redis Cluster for key hash slots, + * reference implementation from the Redis Cluster specification */ +static const uint16_t crc16tab[256] = { + 0x0000,0x1021,0x2042,0x3063,0x4084,0x50a5,0x60c6,0x70e7, + 0x8108,0x9129,0xa14a,0xb16b,0xc18c,0xd1ad,0xe1ce,0xf1ef, + 0x1231,0x0210,0x3273,0x2252,0x52b5,0x4294,0x72f7,0x62d6, + 0x9339,0x8318,0xb37b,0xa35a,0xd3bd,0xc39c,0xf3ff,0xe3de, + 0x2462,0x3443,0x0420,0x1401,0x64e6,0x74c7,0x44a4,0x5485, + 0xa56a,0xb54b,0x8528,0x9509,0xe5ee,0xf5cf,0xc5ac,0xd58d, + 0x3653,0x2672,0x1611,0x0630,0x76d7,0x66f6,0x5695,0x46b4, + 0xb75b,0xa77a,0x9719,0x8738,0xf7df,0xe7fe,0xd79d,0xc7bc, + 0x48c4,0x58e5,0x6886,0x78a7,0x0840,0x1861,0x2802,0x3823, + 0xc9cc,0xd9ed,0xe98e,0xf9af,0x8948,0x9969,0xa90a,0xb92b, + 0x5af5,0x4ad4,0x7ab7,0x6a96,0x1a71,0x0a50,0x3a33,0x2a12, + 0xdbfd,0xcbdc,0xfbbf,0xeb9e,0x9b79,0x8b58,0xbb3b,0xab1a, + 0x6ca6,0x7c87,0x4ce4,0x5cc5,0x2c22,0x3c03,0x0c60,0x1c41, + 0xedae,0xfd8f,0xcdec,0xddcd,0xad2a,0xbd0b,0x8d68,0x9d49, + 0x7e97,0x6eb6,0x5ed5,0x4ef4,0x3e13,0x2e32,0x1e51,0x0e70, + 0xff9f,0xefbe,0xdfdd,0xcffc,0xbf1b,0xaf3a,0x9f59,0x8f78, + 0x9188,0x81a9,0xb1ca,0xa1eb,0xd10c,0xc12d,0xf14e,0xe16f, + 0x1080,0x00a1,0x30c2,0x20e3,0x5004,0x4025,0x7046,0x6067, + 0x83b9,0x9398,0xa3fb,0xb3da,0xc33d,0xd31c,0xe37f,0xf35e, + 0x02b1,0x1290,0x22f3,0x32d2,0x4235,0x5214,0x6277,0x7256, + 0xb5ea,0xa5cb,0x95a8,0x8589,0xf56e,0xe54f,0xd52c,0xc50d, + 0x34e2,0x24c3,0x14a0,0x0481,0x7466,0x6447,0x5424,0x4405, + 0xa7db,0xb7fa,0x8799,0x97b8,0xe75f,0xf77e,0xc71d,0xd73c, + 0x26d3,0x36f2,0x0691,0x16b0,0x6657,0x7676,0x4615,0x5634, + 0xd94c,0xc96d,0xf90e,0xe92f,0x99c8,0x89e9,0xb98a,0xa9ab, + 0x5844,0x4865,0x7806,0x6827,0x18c0,0x08e1,0x3882,0x28a3, + 0xcb7d,0xdb5c,0xeb3f,0xfb1e,0x8bf9,0x9bd8,0xabbb,0xbb9a, + 0x4a75,0x5a54,0x6a37,0x7a16,0x0af1,0x1ad0,0x2ab3,0x3a92, + 0xfd2e,0xed0f,0xdd6c,0xcd4d,0xbdaa,0xad8b,0x9de8,0x8dc9, + 0x7c26,0x6c07,0x5c64,0x4c45,0x3ca2,0x2c83,0x1ce0,0x0cc1, + 0xef1f,0xff3e,0xcf5d,0xdf7c,0xaf9b,0xbfba,0x8fd9,0x9ff8, + 0x6e17,0x7e36,0x4e55,0x5e74,0x2e93,0x3eb2,0x0ed1,0x1ef0 +}; + +static uint16_t rdb_crc16(const char *buf, int len) +{ + int i; + uint16_t crc = 0; + + for (i = 0; i < len; i++) + crc = (crc << 8) ^ crc16tab[((crc >> 8) ^ *buf++) & 0x00FF]; + return crc; +} + +/* hash slot of a key, honoring {hash tags} as per the cluster spec */ +unsigned int rdb_key_slot(const char *key, int len) +{ + int s, e; + + for (s = 0; s < len; s++) + if (key[s] == '{') + break; + + if (s < len) { + for (e = s+1; e < len; e++) + if (key[e] == '}') + break; + if (e < len && e != s+1) + return rdb_crc16(key+s+1, e-s-1) % RDB_NR_SLOTS; + } + + return rdb_crc16(key, len) % RDB_NR_SLOTS; +} + + +static redisContext *rdb_connect_ctx(const char *host, unsigned short port) +{ + struct timeval tv; + redisContext *ctx; + + tv.tv_sec = rdb_connect_timeout / 1000; + tv.tv_usec = (rdb_connect_timeout % 1000) * 1000; + + ctx = redisConnectWithTimeout(host, port, tv); + if (!ctx) { + LM_ERR("failed to allocate redis context for %s:%u\n", host, port); + return NULL; + } + if (ctx->err) { + LM_ERR("failed to connect to redis %s:%u (%s)\n", + host, port, ctx->errstr); + redisFree(ctx); + return NULL; + } + + tv.tv_sec = rdb_query_timeout / 1000; + tv.tv_usec = (rdb_query_timeout % 1000) * 1000; + redisSetTimeout(ctx, tv); + + return ctx; +} + + +/* authenticate + select database on a fresh context, using the db URL id */ +static int rdb_setup_ctx(struct redis_con *con, redisContext *ctx) +{ + redisReply *reply; + int dbno; + str s; + + if (con->id->password) { + if (con->id->username && con->id->username[0]) + reply = redisCommand(ctx, "AUTH %s %s", + con->id->username, con->id->password); + else + reply = redisCommand(ctx, "AUTH %s", con->id->password); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LM_ERR("redis AUTH failed (%s)\n", + reply ? reply->str : ctx->errstr); + if (reply) freeReplyObject(reply); + return -1; + } + freeReplyObject(reply); + } + + /* database number only makes sense outside cluster mode */ + if (con->mode != RDB_MODE_CLUSTER && + con->id->database && con->id->database[0]) { + s.s = con->id->database; + s.len = strlen(s.s); + if (str2sint(&s, &dbno) == 0 && dbno > 0) { + reply = redisCommand(ctx, "SELECT %d", dbno); + if (!reply || reply->type == REDIS_REPLY_ERROR) { + LM_ERR("redis SELECT %d failed (%s)\n", dbno, + reply ? reply->str : ctx->errstr); + if (reply) freeReplyObject(reply); + return -1; + } + freeReplyObject(reply); + } + } + + return 0; +} + + +static redis_node *rdb_find_node(struct redis_con *con, + const char *host, unsigned short port) +{ + redis_node *n; + + for (n = con->nodes; n; n = n->next) + if (n->port == port && strcmp(n->host, host) == 0) + return n; + return NULL; +} + + +/* add a node to the pool and (eagerly) connect it */ +static redis_node *rdb_add_node(struct redis_con *con, + const char *host, unsigned short port) +{ + redis_node *n; + + n = pkg_malloc(sizeof *n); + if (!n) { + LM_ERR("no more pkg memory for redis node\n"); + return NULL; + } + memset(n, 0, sizeof *n); + strncpy(n->host, host, RDB_HOST_MAX-1); + n->port = port; + + n->ctx = rdb_connect_ctx(n->host, n->port); + if (n->ctx && rdb_setup_ctx(con, n->ctx) < 0) { + redisFree(n->ctx); + n->ctx = NULL; + } + if (!n->ctx) + LM_WARN("redis node %s:%u added disconnected, " + "will retry on first use\n", n->host, n->port); + + n->next = con->nodes; + con->nodes = n; + return n; +} + + +static void rdb_disconnect_node(redis_node *n) +{ + if (n->ctx) { + redisFree(n->ctx); + n->ctx = NULL; + } +} + + +/* (re)establish the TCP connection of a node */ +static int rdb_reconnect_node(struct redis_con *con, redis_node *n) +{ + rdb_disconnect_node(n); + n->ctx = rdb_connect_ctx(n->host, n->port); + if (!n->ctx) + return -1; + if (rdb_setup_ctx(con, n->ctx) < 0) { + rdb_disconnect_node(n); + return -1; + } + return 0; +} + + +/* assign [start,end] slot range to a node */ +static void rdb_map_slots(struct redis_con *con, redis_node *n, + int start, int end) +{ + int i; + + if (start < 0 || end >= RDB_NR_SLOTS || start > end) + return; + for (i = start; i <= end; i++) + con->slot_map[i] = n; +} + + +/* parse one "host port" pair out of a CLUSTER SHARDS node map */ +static int rdb_shards_node_endpoint(redisReply *node_map, + char *host, size_t host_len, unsigned short *port, int *is_master, + int *is_online) +{ + size_t i; + redisReply *k, *v; + + host[0] = 0; + *port = 0; + *is_master = 0; + *is_online = 1; + + if (node_map->type != REDIS_REPLY_ARRAY && + node_map->type != REDIS_REPLY_MAP) + return -1; + + for (i = 0; i+1 < node_map->elements; i += 2) { + k = node_map->element[i]; + v = node_map->element[i+1]; + if (k->type != REDIS_REPLY_STRING) + continue; + if (strcmp(k->str, "ip") == 0 || strcmp(k->str, "endpoint") == 0) { + if (v->type == REDIS_REPLY_STRING && v->str[0] && !host[0]) { + strncpy(host, v->str, host_len-1); + host[host_len-1] = 0; + } + } else if (strcmp(k->str, "port") == 0) { + if (v->type == REDIS_REPLY_INTEGER) + *port = (unsigned short)v->integer; + } else if (strcmp(k->str, "role") == 0) { + if (v->type == REDIS_REPLY_STRING && + strcmp(v->str, "master") == 0) + *is_master = 1; + } else if (strcmp(k->str, "health") == 0) { + if (v->type == REDIS_REPLY_STRING && + strcmp(v->str, "online") != 0) + *is_online = 0; + } + } + + return (host[0] && *port) ? 0 : -1; +} + + +/* build topology from a CLUSTER SHARDS reply (Redis 7.0+) */ +static int rdb_parse_shards(struct redis_con *con, redisReply *reply) +{ + size_t i, j, r; + redisReply *shard, *k, *v, *slots = NULL, *rnodes = NULL; + redis_node *n; + char host[RDB_HOST_MAX]; + unsigned short port; + int is_master, is_online, mapped = 0; + + if (reply->type != REDIS_REPLY_ARRAY || reply->elements == 0) + return -1; + + for (i = 0; i < reply->elements; i++) { + shard = reply->element[i]; + if (shard->type != REDIS_REPLY_ARRAY && + shard->type != REDIS_REPLY_MAP) + continue; + + slots = NULL; + rnodes = NULL; + for (j = 0; j+1 < shard->elements; j += 2) { + k = shard->element[j]; + v = shard->element[j+1]; + if (k->type != REDIS_REPLY_STRING) + continue; + if (strcmp(k->str, "slots") == 0) + slots = v; + else if (strcmp(k->str, "nodes") == 0) + rnodes = v; + } + if (!slots || !rnodes || rnodes->type != REDIS_REPLY_ARRAY) + continue; + + /* find the online master of this shard */ + n = NULL; + for (r = 0; r < rnodes->elements; r++) { + if (rdb_shards_node_endpoint(rnodes->element[r], host, + sizeof host, &port, &is_master, &is_online) < 0) + continue; + if (!is_master || !is_online) + continue; + n = rdb_find_node(con, host, port); + if (!n) + n = rdb_add_node(con, host, port); + break; + } + if (!n) + continue; + + /* slots is a flat array of start,end pairs */ + if (slots->type == REDIS_REPLY_ARRAY) { + for (r = 0; r+1 < slots->elements; r += 2) { + if (slots->element[r]->type != REDIS_REPLY_INTEGER || + slots->element[r+1]->type != REDIS_REPLY_INTEGER) + continue; + rdb_map_slots(con, n, + (int)slots->element[r]->integer, + (int)slots->element[r+1]->integer); + mapped = 1; + } + } + } + + return mapped ? 0 : -1; +} + + +/* build topology from a CLUSTER SLOTS reply (Redis 3.0+) */ +static int rdb_parse_slots(struct redis_con *con, redisReply *reply) +{ + size_t i; + redisReply *range, *master; + redis_node *n; + int start, end, mapped = 0; + + if (reply->type != REDIS_REPLY_ARRAY || reply->elements == 0) + return -1; + + for (i = 0; i < reply->elements; i++) { + range = reply->element[i]; + if (range->type != REDIS_REPLY_ARRAY || range->elements < 3) + continue; + if (range->element[0]->type != REDIS_REPLY_INTEGER || + range->element[1]->type != REDIS_REPLY_INTEGER) + continue; + start = (int)range->element[0]->integer; + end = (int)range->element[1]->integer; + + /* element[2] is the master: [host, port, id...] */ + master = range->element[2]; + if (master->type != REDIS_REPLY_ARRAY || master->elements < 2 || + master->element[0]->type != REDIS_REPLY_STRING || + master->element[1]->type != REDIS_REPLY_INTEGER) + continue; + + n = rdb_find_node(con, master->element[0]->str, + (unsigned short)master->element[1]->integer); + if (!n) + n = rdb_add_node(con, master->element[0]->str, + (unsigned short)master->element[1]->integer); + if (!n) + continue; + + rdb_map_slots(con, n, start, end); + mapped = 1; + } + + return mapped ? 0 : -1; +} + + +/* fetch and apply the full cluster topology using any usable node; + * newly discovered masters are eagerly connected */ +static int rdb_load_topology(struct redis_con *con, redisContext *seed_ctx) +{ + redisReply *reply; + int rc = -1; + + /* CLUSTER SHARDS first (7.0+), CLUSTER SLOTS as fallback */ + reply = redisCommand(seed_ctx, "CLUSTER SHARDS"); + if (reply && reply->type == REDIS_REPLY_ARRAY && reply->elements) { + rc = rdb_parse_shards(con, reply); + freeReplyObject(reply); + if (rc == 0) { + LM_DBG("cluster topology loaded via CLUSTER SHARDS\n"); + return 0; + } + } else if (reply) { + freeReplyObject(reply); + } + + reply = redisCommand(seed_ctx, "CLUSTER SLOTS"); + if (reply && reply->type == REDIS_REPLY_ARRAY && reply->elements) { + rc = rdb_parse_slots(con, reply); + freeReplyObject(reply); + if (rc == 0) { + LM_DBG("cluster topology loaded via CLUSTER SLOTS\n"); + return 0; + } + return -1; + } + if (reply) + freeReplyObject(reply); + + return -1; +} + + +/* refresh the slot map from any connected master; prune masters that + * no longer own slots */ +static int rdb_refresh_topology(struct redis_con *con) +{ + redis_node *n, **prev; + int i, rc = -1; + + memset(con->slot_map, 0, sizeof con->slot_map); + + for (n = con->nodes; n; n = n->next) { + if (!n->ctx && rdb_reconnect_node(con, n) < 0) + continue; + rc = rdb_load_topology(con, n->ctx); + if (rc == 0) + break; + } + if (rc < 0) { + LM_ERR("failed to refresh cluster topology from any known node\n"); + return -1; + } + + /* drop nodes that own no slots anymore (demoted/removed) */ + prev = &con->nodes; + while ((n = *prev) != NULL) { + for (i = 0; i < RDB_NR_SLOTS; i++) + if (con->slot_map[i] == n) + break; + if (i == RDB_NR_SLOTS) { + LM_INFO("dropping redis node %s:%u (owns no slots)\n", + n->host, n->port); + *prev = n->next; + rdb_disconnect_node(n); + pkg_free(n); + } else { + prev = &n->next; + } + } + + con->need_refresh = 0; + return 0; +} + + +void rdb_maybe_refresh(struct redis_con *con) +{ + if (con->mode == RDB_MODE_CLUSTER && con->need_refresh) + rdb_refresh_topology(con); +} + + +/* log the discovered master pool and how many slots each owns; also + * reports whether the warm TCP connection to each master is up */ +static void rdb_log_pool(struct redis_con *con) +{ + redis_node *n; + int i, slots, total_masters = 0, up = 0; + + for (n = con->nodes; n; n = n->next) { + slots = 0; + for (i = 0; i < RDB_NR_SLOTS; i++) + if (con->slot_map[i] == n) + slots++; + total_masters++; + if (n->ctx) + up++; + LM_DBG("redis cluster master %s:%u - %d slots, connection %s\n", + n->host, n->port, slots, n->ctx ? "up" : "down"); + } + LM_INFO("redis cluster pool ready: %d masters discovered, " + "%d warm connections\n", total_masters, up); +} + + +/* determine whether the endpoint is a cluster; the "cluster_enabled" + * field lives in the "Cluster" section of the general INFO command + * (it is not part of CLUSTER INFO) and is reliably reported as + * cluster_enabled:0/1 by both standalone and cluster instances; on any + * error assume non-cluster */ +static int rdb_probe_cluster(redisContext *ctx) +{ + redisReply *reply; + int enabled = 0; + + reply = redisCommand(ctx, "INFO CLUSTER"); + if (!reply) + return 0; + if (reply->type == REDIS_REPLY_STRING || reply->type == REDIS_REPLY_STATUS) { + if (reply->str && strstr(reply->str, "cluster_enabled:1")) + enabled = 1; + } + freeReplyObject(reply); + return enabled; +} + + +struct redis_con* db_redis_new_connection(const struct db_id* id) +{ + struct redis_con *con; + redisContext *seed; + unsigned short port; + int is_cluster; + + if (!id || !id->host) { + LM_ERR("invalid db URL for redis connection\n"); + return NULL; + } + + con = pkg_malloc(sizeof *con); + if (!con) { + LM_ERR("no more pkg memory for redis connection\n"); + return NULL; + } + memset(con, 0, sizeof *con); + con->id = (struct db_id *)id; + con->ref = 1; + con->mode = RDB_MODE_SINGLE; /* until probed otherwise */ + + port = id->port ? id->port : 6379; + + seed = rdb_connect_ctx(id->host, port); + if (!seed) + goto error; + if (rdb_setup_ctx(con, seed) < 0) { + redisFree(seed); + goto error; + } + + is_cluster = rdb_probe_cluster(seed); + + if (rdb_mode == RDB_MODE_SINGLE && is_cluster) { + LM_ERR("mode pinned to single but %s:%u is a redis cluster node\n", + id->host, port); + redisFree(seed); + goto error; + } + if (rdb_mode == RDB_MODE_CLUSTER && !is_cluster) { + LM_ERR("mode pinned to cluster but %s:%u has cluster " + "support disabled\n", id->host, port); + redisFree(seed); + goto error; + } + + if (is_cluster) { + con->mode = RDB_MODE_CLUSTER; + /* discover all masters and connect them eagerly, so MOVED + * redirects can be served over already-open connections */ + if (rdb_load_topology(con, seed) < 0) { + LM_ERR("failed to load cluster topology from %s:%u\n", + id->host, port); + redisFree(seed); + goto error; + } + /* the seed served its purpose; the pool holds the masters + * (the seed itself is in the pool if it is a master) */ + redisFree(seed); + LM_INFO("connected to redis cluster via seed %s:%u\n", + id->host, port); + rdb_log_pool(con); + } else { + con->mode = RDB_MODE_SINGLE; + /* the seed becomes the single pooled node */ + if (!rdb_add_node(con, id->host, port)) { + redisFree(seed); + goto error; + } + /* rdb_add_node opened its own ctx; keep that one, drop seed */ + redisFree(seed); + if (!con->nodes->ctx) { + LM_ERR("failed to connect single redis node %s:%u\n", + id->host, port); + goto error; + } + LM_INFO("connected to redis server %s:%u\n", id->host, port); + } + + return con; + +error: + db_redis_free_connection((struct pool_con *)con); + return NULL; +} + + +void db_redis_free_connection(struct pool_con* pcon) +{ + struct redis_con *con = (struct redis_con *)pcon; + redis_node *n, *next; + + if (!con) + return; + + for (n = con->nodes; n; n = next) { + next = n->next; + rdb_disconnect_node(n); + pkg_free(n); + } + rdb_free_schemas(con); + pkg_free(con); +} + + +/* pick the node a key routes to */ +static redis_node *rdb_route(struct redis_con *con, const str *key) +{ + redis_node *n; + + if (con->mode != RDB_MODE_CLUSTER) + return con->nodes; + + n = con->slot_map[rdb_key_slot(key->s, key->len)]; + if (!n) { + /* unknown slot owner - any node will answer or redirect us */ + con->need_refresh = 1; + n = con->nodes; + } + return n; +} + + +/* parse "MOVED :" / "ASK :" */ +static int rdb_parse_redirect(const char *err, int *slot, + char *host, size_t host_len, unsigned short *port) +{ + const char *p, *colon; + size_t hlen; + + p = strchr(err, ' '); + if (!p) + return -1; + *slot = atoi(p+1); + p = strchr(p+1, ' '); + if (!p) + return -1; + p++; + colon = strrchr(p, ':'); + if (!colon || colon == p) + return -1; + hlen = colon - p; + if (hlen >= host_len) + return -1; + memcpy(host, p, hlen); + host[hlen] = 0; + *port = (unsigned short)atoi(colon+1); + return (*port != 0) ? 0 : -1; +} + + +redisReply *rdb_cmd_node(struct redis_con *con, redis_node *node, + int argc, const char **argv, const size_t *argvlen) +{ + redisReply *reply; + int retried = 0; + +again: + if (!node->ctx && rdb_reconnect_node(con, node) < 0) + return NULL; + + reply = redisCommandArgv(node->ctx, argc, argv, argvlen); + if (!reply) { + /* I/O error - reconnect once (covers LB-reaped idle + * connections and failed-over endpoints) */ + LM_DBG("redis I/O error on %s:%u (%s), reconnecting\n", + node->host, node->port, + node->ctx ? node->ctx->errstr : "no ctx"); + rdb_disconnect_node(node); + if (!retried) { + retried = 1; + goto again; + } + LM_ERR("redis command failed on %s:%u after reconnect\n", + node->host, node->port); + return NULL; + } + + /* -READONLY: we are talking to a demoted master (e.g. the LB has + * not yet cut over) - reconnect through the same endpoint once */ + if (reply->type == REDIS_REPLY_ERROR && reply->str && + strncmp(reply->str, "READONLY", 8) == 0 && !retried) { + LM_INFO("redis node %s:%u went read-only, reconnecting\n", + node->host, node->port); + freeReplyObject(reply); + rdb_disconnect_node(node); + if (con->mode == RDB_MODE_CLUSTER) + con->need_refresh = 1; + retried = 1; + goto again; + } + + return reply; +} + + +redisReply *rdb_cmd_key(struct redis_con *con, const str *key, + int argc, const char **argv, const size_t *argvlen) +{ + redisReply *reply; + redis_node *node, *target; + char host[RDB_HOST_MAX]; + unsigned short port; + int slot, redirects; + const char *asking = "ASKING"; + size_t asking_len = 6; + redisReply *ask_reply; + + node = rdb_route(con, key); + if (!node) { + LM_ERR("no redis node available\n"); + return NULL; + } + + for (redirects = 0; redirects <= RDB_MAX_REDIRECTS; redirects++) { + + reply = rdb_cmd_node(con, node, argc, argv, argvlen); + if (!reply) + return NULL; + + if (reply->type != REDIS_REPLY_ERROR || !reply->str) + return reply; + + if (strncmp(reply->str, "MOVED ", 6) == 0 && + con->mode == RDB_MODE_CLUSTER) { + if (rdb_parse_redirect(reply->str, &slot, host, + sizeof host, &port) < 0) { + LM_ERR("unparsable MOVED reply <%s>\n", reply->str); + return reply; + } + freeReplyObject(reply); + + /* serve the redirect from the warm pool when possible */ + target = rdb_find_node(con, host, port); + if (!target) + target = rdb_add_node(con, host, port); + if (!target) + return NULL; + if (slot >= 0 && slot < RDB_NR_SLOTS) + con->slot_map[slot] = target; + /* the map is stale beyond this slot - refresh before + * the next operation, not in the hot path */ + con->need_refresh = 1; + node = target; + continue; + } + + if (strncmp(reply->str, "ASK ", 4) == 0 && + con->mode == RDB_MODE_CLUSTER) { + if (rdb_parse_redirect(reply->str, &slot, host, + sizeof host, &port) < 0) { + LM_ERR("unparsable ASK reply <%s>\n", reply->str); + return reply; + } + freeReplyObject(reply); + + /* one-shot redirect: ASKING + command, no map change */ + target = rdb_find_node(con, host, port); + if (!target) + target = rdb_add_node(con, host, port); + if (!target) + return NULL; + ask_reply = rdb_cmd_node(con, target, 1, &asking, &asking_len); + if (ask_reply) + freeReplyObject(ask_reply); + node = target; + continue; + } + + if (strncmp(reply->str, "TRYAGAIN", 8) == 0 && + con->mode == RDB_MODE_CLUSTER) { + /* slot migration in progress; brief backoff and retry */ + freeReplyObject(reply); + usleep(20000); + continue; + } + + /* genuine command error - hand it to the caller */ + return reply; + } + + LM_ERR("redis redirect limit exceeded for key <%.*s>\n", + key->len, key->s); + return NULL; +} diff --git a/modules/db_redis/redis_con.h b/modules/db_redis/redis_con.h new file mode 100644 index 00000000000..4fbec487a98 --- /dev/null +++ b/modules/db_redis/redis_con.h @@ -0,0 +1,88 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef REDIS_CON_H +#define REDIS_CON_H + +#include + +/* RESP3 map replies only exist in hiredis >= 1.0; with the default RESP2 + * protocol these arrive as flat arrays, so on older hiredis the constant + * just needs to exist for the (never-matching) comparisons */ +#ifndef REDIS_REPLY_MAP +#define REDIS_REPLY_MAP 9 +#endif + +#include "../../db/db_pool.h" +#include "../../db/db_id.h" +#include "../../str.h" + +#define RDB_NR_SLOTS 16384 + +#define RDB_HOST_MAX 128 + +/* one TCP connection to one Redis server (a cluster master, or the + * single endpoint in non-cluster mode) */ +typedef struct redis_node { + char host[RDB_HOST_MAX]; + unsigned short port; + redisContext *ctx; /* NULL when disconnected */ + struct redis_node *next; +} redis_node; + +struct rdb_schema; + +struct redis_con { + struct db_id* id; /**< Connection identifier */ + unsigned int ref; /**< Reference count */ + struct pool_con *async_pool; /**< Subpool of identical database handles */ + int no_transfers; /**< Number of async queries to this backend */ + struct db_transfer *transfers; /**< Array of ongoing async operations */ + struct pool_con *next; /**< Next element in the pool (different id) */ + + int mode; /* RDB_MODE_SINGLE or RDB_MODE_CLUSTER */ + redis_node *nodes; /* master pool (single mode: one node) */ + redis_node *slot_map[RDB_NR_SLOTS]; /* cluster: slot owner (NULL=unknown) */ + int need_refresh; /* topology refresh requested (cluster) */ + + struct rdb_schema *schemas; /* cached table schemas */ + long long last_insert_id; /* per-connection last auto-generated id */ +}; + +#define CON_REDIS(db_con) ((struct redis_con *)((db_con)->tail)) + +struct redis_con* db_redis_new_connection(const struct db_id* id); +void db_redis_free_connection(struct pool_con* con); + +/* run a command routed by 'key' (row key), redirect- and reconnect-aware; + * returns a reply that the caller must freeReplyObject(), or NULL */ +redisReply *rdb_cmd_key(struct redis_con *con, const str *key, + int argc, const char **argv, const size_t *argvlen); + +/* run a command on an explicit node, with one reconnect attempt */ +redisReply *rdb_cmd_node(struct redis_con *con, redis_node *node, + int argc, const char **argv, const size_t *argvlen); + +/* attempt a topology refresh if one was requested (cluster mode) */ +void rdb_maybe_refresh(struct redis_con *con); + +unsigned int rdb_key_slot(const char *key, int len); + +#endif /* REDIS_CON_H */ diff --git a/modules/db_redis/schema.c b/modules/db_redis/schema.c new file mode 100644 index 00000000000..424b749ec26 --- /dev/null +++ b/modules/db_redis/schema.c @@ -0,0 +1,288 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include + +#include "../../dprint.h" +#include "../../mem/mem.h" +#include "../../ut.h" +#include "schema.h" + +#define SCHEMA_KEY_PREFIX "schema:" +#define SCHEMA_KEY_PREFIX_LEN (sizeof(SCHEMA_KEY_PREFIX)-1) + +static int rdb_parse_type(const char *s, size_t len, struct rdb_col *col) +{ + const char *p, *end = s + len; + + col->nullable = 0; + col->is_auto = 0; + + p = memchr(s, ',', len); + if (!p) + p = end; + + if (p-s == 3 && strncmp(s, "int", 3) == 0) + col->type = DB_INT; + else if (p-s == 6 && strncmp(s, "bigint", 6) == 0) + col->type = DB_BIGINT; + else if (p-s == 6 && strncmp(s, "double", 6) == 0) + col->type = DB_DOUBLE; + else if (p-s == 6 && strncmp(s, "string", 6) == 0) + col->type = DB_STR; + else if (p-s == 4 && strncmp(s, "blob", 4) == 0) + col->type = DB_BLOB; + else if (p-s == 8 && strncmp(s, "datetime", 8) == 0) + col->type = DB_DATETIME; + else + return -1; + + while (p < end) { + p++; /* skip ',' */ + if (end-p >= 4 && strncmp(p, "null", 4) == 0) + col->nullable = 1; + else if (end-p >= 4 && strncmp(p, "auto", 4) == 0) + col->is_auto = 1; + p = memchr(p, ',', end-p); + if (!p) + break; + } + + return 0; +} + + +static struct rdb_schema *rdb_load_schema(struct redis_con *con, + const str *table) +{ + struct rdb_schema *sch = NULL; + redisReply *reply = NULL, *k, *v; + char keybuf[SCHEMA_KEY_PREFIX_LEN + 64]; + str key; + const char *argv[2]; + size_t argvlen[2]; + str cols_spec = STR_NULL, pk_name = STR_NULL; + str name; + char *p, *end; + size_t i; + int c; + + if (table->len > 64) { + LM_ERR("table name too long <%.*s>\n", table->len, table->s); + return NULL; + } + + memcpy(keybuf, SCHEMA_KEY_PREFIX, SCHEMA_KEY_PREFIX_LEN); + memcpy(keybuf + SCHEMA_KEY_PREFIX_LEN, table->s, table->len); + key.s = keybuf; + key.len = SCHEMA_KEY_PREFIX_LEN + table->len; + + argv[0] = "HGETALL"; argvlen[0] = 7; + argv[1] = key.s; argvlen[1] = key.len; + + reply = rdb_cmd_key(con, &key, 2, argv, argvlen); + if (!reply) { + LM_ERR("failed to fetch schema for table <%.*s>\n", + table->len, table->s); + return NULL; + } + if (reply->type != REDIS_REPLY_ARRAY || reply->elements == 0) { + LM_ERR("no schema provisioned for table <%.*s> " + "(missing hash <%.*s>)\n", + table->len, table->s, key.len, key.s); + goto error; + } + + /* first pass: locate __cols and __pk */ + for (i = 0; i+1 < reply->elements; i += 2) { + k = reply->element[i]; + v = reply->element[i+1]; + if (k->type != REDIS_REPLY_STRING || v->type != REDIS_REPLY_STRING) + continue; + if (k->len == 6 && strncmp(k->str, "__cols", 6) == 0) { + cols_spec.s = v->str; + cols_spec.len = v->len; + } else if (k->len == 4 && strncmp(k->str, "__pk", 4) == 0) { + pk_name.s = v->str; + pk_name.len = v->len; + } + } + if (!cols_spec.s || !pk_name.s) { + LM_ERR("schema of table <%.*s> lacks __cols or __pk\n", + table->len, table->s); + goto error; + } + + sch = pkg_malloc(sizeof *sch + table->len); + if (!sch) + goto oom; + memset(sch, 0, sizeof *sch); + sch->table.s = (char *)(sch + 1); + sch->table.len = table->len; + memcpy(sch->table.s, table->s, table->len); + sch->pk = -1; + + /* count columns in __cols */ + p = cols_spec.s; + end = cols_spec.s + cols_spec.len; + while (p < end) { + while (p < end && *p == ' ') p++; + if (p == end) break; + sch->nr_cols++; + while (p < end && *p != ' ') p++; + } + if (sch->nr_cols == 0) { + LM_ERR("empty __cols in schema of table <%.*s>\n", + table->len, table->s); + goto error; + } + + sch->cols = pkg_malloc(sch->nr_cols * sizeof *sch->cols); + if (!sch->cols) + goto oom; + memset(sch->cols, 0, sch->nr_cols * sizeof *sch->cols); + + /* second pass: fill the ordered columns, resolving each type */ + p = cols_spec.s; + c = 0; + while (p < end && c < sch->nr_cols) { + while (p < end && *p == ' ') p++; + if (p == end) break; + name.s = p; + while (p < end && *p != ' ') p++; + name.len = p - name.s; + + sch->cols[c].name.s = pkg_malloc(name.len + 1); + if (!sch->cols[c].name.s) + goto oom; + memcpy(sch->cols[c].name.s, name.s, name.len); + sch->cols[c].name.s[name.len] = 0; + sch->cols[c].name.len = name.len; + + /* find the matching type field */ + sch->cols[c].type = DB_INT; + for (i = 0; i+1 < reply->elements; i += 2) { + k = reply->element[i]; + v = reply->element[i+1]; + if (k->type != REDIS_REPLY_STRING || + v->type != REDIS_REPLY_STRING) + continue; + if ((int)k->len == name.len && + memcmp(k->str, name.s, name.len) == 0) { + if (rdb_parse_type(v->str, v->len, &sch->cols[c]) < 0) { + LM_ERR("bad type <%.*s> for column <%.*s> in " + "table <%.*s>\n", (int)v->len, v->str, + name.len, name.s, table->len, table->s); + goto error; + } + break; + } + } + if (i+1 >= reply->elements) { + LM_ERR("column <%.*s> in __cols has no type field in " + "schema of table <%.*s>\n", + name.len, name.s, table->len, table->s); + goto error; + } + + if (name.len == pk_name.len && + memcmp(name.s, pk_name.s, name.len) == 0) + sch->pk = c; + c++; + } + + if (sch->pk < 0) { + LM_ERR("__pk column <%.*s> is not part of __cols in " + "table <%.*s>\n", pk_name.len, pk_name.s, + table->len, table->s); + goto error; + } + + freeReplyObject(reply); + + sch->next = con->schemas; + con->schemas = sch; + + LM_DBG("loaded schema for table <%.*s>: %d columns, pk <%.*s>\n", + table->len, table->s, sch->nr_cols, + sch->cols[sch->pk].name.len, sch->cols[sch->pk].name.s); + return sch; + +oom: + LM_ERR("no more pkg memory while loading schema for <%.*s>\n", + table->len, table->s); +error: + if (reply) + freeReplyObject(reply); + if (sch) { + if (sch->cols) { + for (c = 0; c < sch->nr_cols; c++) + if (sch->cols[c].name.s) + pkg_free(sch->cols[c].name.s); + pkg_free(sch->cols); + } + pkg_free(sch); + } + return NULL; +} + + +struct rdb_schema *rdb_get_schema(struct redis_con *con, const str *table) +{ + struct rdb_schema *sch; + + for (sch = con->schemas; sch; sch = sch->next) + if (sch->table.len == table->len && + memcmp(sch->table.s, table->s, table->len) == 0) + return sch; + + return rdb_load_schema(con, table); +} + + +int rdb_schema_col(const struct rdb_schema *sch, const str *name) +{ + int i; + + for (i = 0; i < sch->nr_cols; i++) + if (sch->cols[i].name.len == name->len && + memcmp(sch->cols[i].name.s, name->s, name->len) == 0) + return i; + return -1; +} + + +void rdb_free_schemas(struct redis_con *con) +{ + struct rdb_schema *sch, *next; + int c; + + for (sch = con->schemas; sch; sch = next) { + next = sch->next; + if (sch->cols) { + for (c = 0; c < sch->nr_cols; c++) + if (sch->cols[c].name.s) + pkg_free(sch->cols[c].name.s); + pkg_free(sch->cols); + } + pkg_free(sch); + } + con->schemas = NULL; +} diff --git a/modules/db_redis/schema.h b/modules/db_redis/schema.h new file mode 100644 index 00000000000..a8b31487f26 --- /dev/null +++ b/modules/db_redis/schema.h @@ -0,0 +1,57 @@ +/* + * Copyright (C) 2026 OpenSIPS Solutions + * + * This file is part of opensips, a free SIP server. + * + * opensips is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 2 of the License, or + * (at your option) any later version + * + * opensips is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef DB_REDIS_SCHEMA_H +#define DB_REDIS_SCHEMA_H + +#include "../../str.h" +#include "../../db/db_val.h" +#include "redis_con.h" + +/* a table schema is provisioned as the hash "schema:" with: + * __cols -> ordered, space-separated column names + * __pk -> name of the primary-key column + * -> type[,null][,auto] + * where type is one of: int, bigint, double, string, blob, datetime */ + +struct rdb_col { + str name; + db_type_t type; + int nullable; + int is_auto; /* auto-increment (emulated via INCR seq:
) */ +}; + +struct rdb_schema { + str table; + int nr_cols; + struct rdb_col *cols; + int pk; /* index into cols[] */ + struct rdb_schema *next; +}; + +/* cached lookup; loads "schema:
" from redis on first use */ +struct rdb_schema *rdb_get_schema(struct redis_con *con, const str *table); + +/* index of a column by name, -1 if not part of the schema */ +int rdb_schema_col(const struct rdb_schema *sch, const str *name); + +void rdb_free_schemas(struct redis_con *con); + +#endif /* DB_REDIS_SCHEMA_H */ diff --git a/modules/usrloc/dlist.c b/modules/usrloc/dlist.c index 3dd2db2b59d..5910d8672ba 100644 --- a/modules/usrloc/dlist.c +++ b/modules/usrloc/dlist.c @@ -129,6 +129,7 @@ static int get_domain_db_ucontacts(udomain_t *d, void *buf, int *len, unsigned int dbflags; int needed; int shortage = 0; + int client_side_partition = 0; uint64_t contact_id; void *record_start; @@ -149,51 +150,99 @@ static int get_domain_db_ucontacts(udomain_t *d, void *buf, int *len, goto error; } - i = snprintf(query_buf, sizeof query_buf, "select %.*s, %.*s, %.*s," + if (DB_CAPABILITY(ul_dbf, DB_CAP_RAW_QUERY)) { + /* the contact_id modulo partitioning is done in the query */ + i = snprintf(query_buf, sizeof query_buf, "select %.*s, %.*s, %.*s," #ifdef ORACLE_USRLOC " %.*s, %.*s, %.*s from %s where %.*s > %lld and mod(contact_id, %u) = %u", #else " %.*s, %.*s, %.*s from %s where %.*s > %lld and contact_id %% %u = %u", #endif - received_col.len, received_col.s, - contact_col.len, contact_col.s, - sock_col.len, sock_col.s, - cflags_col.len, cflags_col.s, - path_col.len, path_col.s, - contactid_col.len, contactid_col.s, - d->name->s, - expires_col.len, expires_col.s, - (long long)now, - part_max, part_idx); - - LM_DBG("query: %.*s\n", (int)(sizeof query_buf), query_buf); - if (i >= sizeof query_buf) { - LM_ERR("DB query too long\n"); - goto error; - } + received_col.len, received_col.s, + contact_col.len, contact_col.s, + sock_col.len, sock_col.s, + cflags_col.len, cflags_col.s, + path_col.len, path_col.s, + contactid_col.len, contactid_col.s, + d->name->s, + expires_col.len, expires_col.s, + (long long)now, + part_max, part_idx); + + LM_DBG("query: %.*s\n", (int)(sizeof query_buf), query_buf); + if (i >= sizeof query_buf) { + LM_ERR("DB query too long\n"); + goto error; + } + + query_str.s = query_buf; + query_str.len = i; + + if (DB_CAPABILITY(ul_dbf, DB_CAP_FETCH)) { + if (ul_dbf.raw_query(ul_dbh, &query_str, 0) < 0) { + LM_ERR("raw_query failed\n"); + goto error; + } - query_str.s = query_buf; - query_str.len = i; + no_rows = estimate_available_rows(20+128+20+128+64, 5); + if (no_rows == 0) + no_rows = 10; - if (DB_CAPABILITY(ul_dbf, DB_CAP_FETCH)) { - if (ul_dbf.raw_query(ul_dbh, &query_str, 0) < 0) { + LM_DBG("fetching %d rows\n", no_rows); + + if (ul_dbf.fetch_result(ul_dbh, &res, no_rows) < 0) { + LM_ERR("Error fetching rows\n"); + goto error; + } + } else if (ul_dbf.raw_query(ul_dbh, &query_str, &res) < 0) { LM_ERR("raw_query failed\n"); goto error; } + } else { + /* the DB backend has no raw query support (e.g. db_redis): run + * the equivalent structured query and apply the contact_id + * modulo partitioning while iterating the resulting rows */ + db_key_t cols[6]; + db_key_t keys[1]; + db_op_t ops[1]; + db_val_t vals[1]; + + cols[0] = &received_col; + cols[1] = &contact_col; + cols[2] = &sock_col; + cols[3] = &cflags_col; + cols[4] = &path_col; + cols[5] = &contactid_col; + + keys[0] = &expires_col; + ops[0] = OP_GT; + memset(vals, 0, sizeof vals); + VAL_TYPE(vals) = DB_BIGINT; + VAL_BIGINT(vals) = (long long)now; + + client_side_partition = (part_max > 1); + + if (DB_CAPABILITY(ul_dbf, DB_CAP_FETCH)) { + if (ul_dbf.query(ul_dbh, keys, ops, vals, cols, 1, 6, 0, 0) < 0) { + LM_ERR("query failed\n"); + goto error; + } - no_rows = estimate_available_rows(20+128+20+128+64, 5); - if (no_rows == 0) - no_rows = 10; + no_rows = estimate_available_rows(20+128+20+128+64, 5); + if (no_rows == 0) + no_rows = 10; - LM_DBG("fetching %d rows\n", no_rows); + LM_DBG("fetching %d rows\n", no_rows); - if (ul_dbf.fetch_result(ul_dbh, &res, no_rows) < 0) { - LM_ERR("Error fetching rows\n"); + if (ul_dbf.fetch_result(ul_dbh, &res, no_rows) < 0) { + LM_ERR("Error fetching rows\n"); + goto error; + } + } else if (ul_dbf.query(ul_dbh, keys, ops, vals, cols, + 1, 6, 0, &res) < 0) { + LM_ERR("query failed\n"); goto error; } - } else if (ul_dbf.raw_query(ul_dbh, &query_str, &res) < 0) { - LM_ERR("raw_query failed\n"); - goto error; } do { @@ -246,6 +295,12 @@ static int get_domain_db_ucontacts(udomain_t *d, void *buf, int *len, /* contact id*/ contact_id = VAL_BIGINT(ROW_VALUES(row) + 5); + /* when the query could not partition by contact_id (no raw + * query support), do the modulo partitioning here */ + if (client_side_partition && + contact_id % part_max != part_idx) + continue; + needed = (int)(p_len + sizeof p_len + r_len + sizeof r_len + p1_len + sizeof p1_len + sizeof sock + sizeof dbflags + sizeof next_hop); diff --git a/modules/usrloc/ul_mod.c b/modules/usrloc/ul_mod.c index 85445fdd12c..ab8ed8bc8f4 100644 --- a/modules/usrloc/ul_mod.c +++ b/modules/usrloc/ul_mod.c @@ -883,9 +883,10 @@ int ul_check_db(void) return -1; } + /* DB_CAP_RAW_QUERY is not required: contact loading has a + * structured-query fallback for backends without raw query + * support (e.g. db_redis) */ db_caps = DB_CAP_ALL; - if (cluster_mode == CM_SQL_ONLY) - db_caps |= DB_CAP_RAW_QUERY; if (!DB_CAPABILITY(ul_dbf, db_caps)) { LM_ERR("database module does not implement all functions" diff --git a/scripts/redis/README b/scripts/redis/README new file mode 100644 index 00000000000..c857e78f4b1 --- /dev/null +++ b/scripts/redis/README @@ -0,0 +1,50 @@ +db_redis provisioning files +=========================== + +These files provision table schemas for the db_redis module. Load the +ones for the modules you intend to back with Redis (add -c when talking +to a Redis Cluster): + + redis-cli [-c] < dialog-create.redis + +Note: redis-cli piped input does not support comment lines, so these +files contain commands only. + +Available files +--------------- + + dialog-create.redis - dialog module (dialog table, v12) + dispatcher-create.redis - dispatcher module (dispatcher table, v9) + usrloc-create.redis - usrloc module (location table, v1013) + clusterer-create.redis - clusterer module (clusterer table v4, + clusterer_bridge v1) + +Each file is self-contained: it also (re)creates the small "version" +table schema, so loading any file on its own is enough. + +Notes on specific modules: + + - The tm module keeps no database table and therefore has no schema + here. + +Data model +---------- + +A table schema is the hash "schema:
" holding: + + __cols -> ordered, space-separated column names + __pk -> primary-key column name + -> type[,null][,auto] + +where type is one of: int, bigint, double, string, blob, datetime. +The "auto" attribute marks an auto-increment primary key, emulated by +the module through INCR on "seq:
". + +Rows are stored as hashes at "
:", one field per +non-NULL column. The "version" table is provisioned like any other +table, with one row per served table, e.g.: + + HSET version:dialog table_name dialog table_version 12 + +The key prefixes "schema:" and "seq:" are reserved by the module and +must not be used as table names. diff --git a/scripts/redis/clusterer-create.redis b/scripts/redis/clusterer-create.redis new file mode 100644 index 00000000000..eee8f64bd52 --- /dev/null +++ b/scripts/redis/clusterer-create.redis @@ -0,0 +1,5 @@ +HSET schema:version __cols "table_name table_version" __pk table_name table_name string table_version int +HSET schema:clusterer __cols "id cluster_id node_id url state no_ping_retries priority sip_addr flags description" __pk id id int,auto cluster_id int node_id int url string state int no_ping_retries int priority int sip_addr string,null flags string,null description string,null +HSET version:clusterer table_name clusterer table_version 4 +HSET schema:clusterer_bridge __cols "id cluster_a cluster_b send_shtag dst_node_csv" __pk id id int,auto cluster_a int cluster_b int send_shtag string dst_node_csv string,null +HSET version:clusterer_bridge table_name clusterer_bridge table_version 1 diff --git a/scripts/redis/dialog-create.redis b/scripts/redis/dialog-create.redis new file mode 100644 index 00000000000..98d9e9ddfef --- /dev/null +++ b/scripts/redis/dialog-create.redis @@ -0,0 +1,3 @@ +HSET schema:version __cols "table_name table_version" __pk table_name table_name string table_version int +HSET schema:dialog __cols "dlg_id callid from_uri from_tag to_uri to_tag mangled_from_uri mangled_to_uri caller_cseq callee_cseq caller_ping_cseq callee_ping_cseq caller_route_set callee_route_set caller_contact callee_contact caller_sock callee_sock state start_time timeout vars profiles script_flags module_flags flags rt_on_answer rt_on_timeout rt_on_hangup" __pk dlg_id dlg_id bigint callid string from_uri string from_tag string to_uri string to_tag string mangled_from_uri string,null mangled_to_uri string,null caller_cseq string callee_cseq string caller_ping_cseq int callee_ping_cseq int caller_route_set string,null callee_route_set string,null caller_contact string,null callee_contact string,null caller_sock string callee_sock string state int start_time int timeout int vars blob,null profiles string,null script_flags string,null module_flags int flags int rt_on_answer string,null rt_on_timeout string,null rt_on_hangup string,null +HSET version:dialog table_name dialog table_version 12 diff --git a/scripts/redis/dispatcher-create.redis b/scripts/redis/dispatcher-create.redis new file mode 100644 index 00000000000..66960a17ebb --- /dev/null +++ b/scripts/redis/dispatcher-create.redis @@ -0,0 +1,3 @@ +HSET schema:version __cols "table_name table_version" __pk table_name table_name string table_version int +HSET schema:dispatcher __cols "id setid destination socket state probe_mode weight priority attrs description" __pk id id int,auto setid int destination string socket string,null state int probe_mode int weight string priority int attrs string,null description string,null +HSET version:dispatcher table_name dispatcher table_version 9 diff --git a/scripts/redis/usrloc-create.redis b/scripts/redis/usrloc-create.redis new file mode 100644 index 00000000000..da2a094c172 --- /dev/null +++ b/scripts/redis/usrloc-create.redis @@ -0,0 +1,3 @@ +HSET schema:version __cols "table_name table_version" __pk table_name table_name string table_version int +HSET schema:location __cols "contact_id username domain contact received path expires q callid cseq last_modified flags cflags user_agent socket methods sip_instance kv_store attr" __pk contact_id contact_id bigint,auto username string domain string,null contact string received string,null path string,null expires int q double callid string cseq int last_modified datetime flags int cflags string,null user_agent string socket string,null methods int,null sip_instance string,null kv_store string,null attr string,null +HSET version:location table_name location table_version 1013