MT#55283 Adding redis-subscribe parameter

Aim is to decouple redis read/write/subscribe.

Closes #1881

Change-Id: I9bfc46fa738898ab7fcac4f6a4691a6def6f77ba

daemon/main.c

@@ -500,6 +500,7 @@ static void options(int *argc, char ***argv, GHashTable *templates) {
 	g_autoptr(char) graphite_prefix_s = NULL;
 	g_autoptr(char) redisps = NULL;
 	g_autoptr(char) redisps_write = NULL;
+	g_autoptr(char) redisps_subscribe = NULL;
 	g_autoptr(char) log_facility_cdr_s = NULL;
 	g_autoptr(char) log_facility_rtcp_s = NULL;
 	g_autoptr(char) log_facility_dtmf_s = NULL;
@@ -571,6 +572,7 @@ static void options(int *argc, char ***argv, GHashTable *templates) {
 		{ "port-max",	'M', 0, G_OPTION_ARG_INT,	&rtpe_config.port_max,	"Highest port to use for RTP",	"INT"		},
 		{ "redis",	'r', 0, G_OPTION_ARG_STRING,	&redisps,	"Connect to Redis database",	"[PW@]IP:PORT/INT"	},
 		{ "redis-write",'w', 0, G_OPTION_ARG_STRING,    &redisps_write, "Connect to Redis write database",      "[PW@]IP:PORT/INT"       },
+		{ "redis-subscribe", 0, 0, G_OPTION_ARG_STRING, &redisps_subscribe, "Connect to Redis subscribe database",      "[PW@]IP:PORT/INT"       },
 		{ "redis-resolve-on-reconnect", 0,0,	G_OPTION_ARG_NONE,	&rtpe_config.redis_resolve_on_reconnect,	"Re-resolve given FQDN on each re-connect to the redis server.",	NULL },
 		{ "redis-num-threads", 0, 0, G_OPTION_ARG_INT, &rtpe_config.redis_num_threads, "Number of Redis restore threads",      "INT"       },
 		{ "redis-expires", 0, 0, G_OPTION_ARG_INT, &rtpe_config.redis_expires_secs, "Expire time in seconds for redis keys",      "INT"       },
@@ -894,6 +896,13 @@ static void options(int *argc, char ***argv, GHashTable *templates) {
 		}
 	}
 
+	if (redisps_subscribe)
+		if (redis_ep_parse(&rtpe_config.redis_subscribe_ep, &rtpe_config.redis_subscribe_db, &rtpe_config.redis_subscribe_hostname,
+					&rtpe_config.redis_subscribe_auth,"RTPENGINE_REDIS_SUBSCRIBE_AUTH_PW", redisps_subscribe))
+		{
+			die("Invalid Redis endpoint [IP:PORT/INT] '%s' (--redis-subscribe)", redisps_subscribe);
+		}
+
 	if (rtpe_config.fmt > 2)
 		die("Invalid XMLRPC format");
 
@@ -1392,6 +1401,24 @@ static void create_everything(void) {
 				endpoint_print_buf(&rtpe_config.redis_write_ep));
 	}
 
+	if (!is_addr_unspecified(&rtpe_config.redis_subscribe_ep.address)) {
+		rtpe_redis_notify = redis_new(&rtpe_config.redis_subscribe_ep,
+				rtpe_config.redis_subscribe_db,
+				rtpe_config.redis_subscribe_hostname,
+				rtpe_config.redis_subscribe_auth,
+				ANY_REDIS_ROLE,
+				rtpe_config.no_redis_required,
+				rtpe_config.redis_resolve_on_reconnect);
+
+		if (!rtpe_redis_notify)
+			die("Cannot start up without running Redis %s subscribe database! See also NO_REDIS_REQUIRED parameter.",
+				endpoint_print_buf(&rtpe_config.redis_subscribe_ep));
+		// subscribed-kespaces takes precedence over db in notify ep
+		if (!rtpe_config.redis_subscribed_keyspaces.length) {
+			g_queue_push_tail(&rtpe_config.redis_subscribed_keyspaces, GINT_TO_POINTER(rtpe_config.redis_subscribe_db));
+		}
+	}
+
 	if (!is_addr_unspecified(&rtpe_config.redis_ep.address)) {
 		rtpe_redis = redis_new(&rtpe_config.redis_ep,
 				rtpe_config.redis_db,
@@ -1406,7 +1433,7 @@ static void create_everything(void) {
 					"See also NO_REDIS_REQUIRED parameter.",
 				endpoint_print_buf(&rtpe_config.redis_ep));
 
-		if (rtpe_config.redis_subscribed_keyspaces.length) {
+		if (rtpe_config.redis_subscribed_keyspaces.length && !rtpe_redis_notify) {
 			rtpe_redis_notify = redis_new(&rtpe_config.redis_ep,
 					rtpe_config.redis_db,
 					rtpe_config.redis_hostname,
@@ -1546,7 +1573,7 @@ int main(int argc, char **argv) {
 	if (!is_addr_unspecified(&rtpe_config.redis_ep.address) && initial_rtpe_config.redis_delete_async)
 		thread_create_detach(redis_delete_async_loop, NULL, "redis async");
 
-	if (!is_addr_unspecified(&rtpe_config.redis_ep.address) && rtpe_redis_notify)
+	if (rtpe_redis_notify)
 		thread_create_detach(redis_notify_loop, NULL, "redis notify");
 
 	do_redis_restore();
@@ -1595,7 +1622,7 @@ int main(int argc, char **argv) {
 	if (!is_addr_unspecified(&rtpe_config.redis_ep.address) && initial_rtpe_config.redis_delete_async)
 		redis_async_event_base_action(rtpe_redis_write, EVENT_BASE_LOOPBREAK);
 
-	if (!is_addr_unspecified(&rtpe_config.redis_ep.address) && rtpe_redis_notify)
+	if (rtpe_redis_notify)
 		redis_async_event_base_action(rtpe_redis_notify, EVENT_BASE_LOOPBREAK);
 
 	threads_join_all(true);

docs/rtpengine.md

@@ -535,10 +535,23 @@ call to inject-DTMF won't be sent to __\-\-dtmf-log-dest=__ or __\-\-listen-tcp-
     When both options are given, __rtpengine__ will start and use the Redis
     database regardless of the database's role (master or slave).
 
+- __\-\-redis-subscribe=__\[*PW*@\]*IP*:*PORT*/*INT*
+
+    Configures a Redis database for subscribing and receiving notifications.
+    This option takes precedence over __\-\-redis__, if configured.
+    When __\-\-subscribe-keyspace__ is also configured, the keyspace part of
+    __\-\-redis-subscribe=__ is not used, the former takes precedence.
+
+    For password protected Redis servers, the environment variable for the
+    password is __RTPENGINE\*REDIS\*SUBSCRIBE\*AUTH\*PW__.
+
 - __-k__, __\-\-subscribe-keyspace=__*INT*
 
-    List of redis keyspaces to subscribe.
-    If this is not present, no keyspaces are subscribed (default behaviour).
+    List of redis keyspaces to subscribe. When it and __\-\-redis-subscribe=__
+    are not present, no keyspaces are subscribed (default behaviour).
+    In conjunction with __\-\-redis-subscribe=__, __\-\-subscribe-keyspace=__
+    overwrites the keyspaces to subscribe to.
+
     Further subscriptions could be added/removed via __rtpengine-ctl ksadd/ksrm__.
     This may lead to enabling/disabling of the redis keyspace notification feature.
 

etc/rtpengine.conf

@@ -67,6 +67,7 @@ recording-method = proc
 
 # redis = 127.0.0.1:6379/5
 # redis-write = password@12.23.34.45:6379/42
+# redis-subscribe = password@12.23.34.45:6379/43
 # redis-num-threads = 8
 # no-redis-required = false
 # redis-expires = 86400

include/main.h

@@ -53,6 +53,7 @@ enum endpoint_learning {
 	X(port_max) \
 	X(redis_db) \
 	X(redis_write_db) \
+	X(redis_subscribe_db) \
 	X(redis_allowed_errors) \
 	X(redis_disable_time) \
 	X(redis_cmd_timeout) \
@@ -118,8 +119,10 @@ enum endpoint_learning {
 	X(b2b_url) \
 	X(redis_auth) \
 	X(redis_write_auth) \
+	X(redis_subscribe_auth) \
 	X(redis_hostname) \
 	X(redis_write_hostname) \
+	X(redis_subscribe_hostname) \
 	X(spooldir) \
 	X(rec_method) \
 	X(rec_format) \
@@ -152,6 +155,7 @@ enum endpoint_learning {
 	X(graphite_ep) \
 	X(redis_ep) \
 	X(redis_write_ep) \
+	X(redis_subscribe_ep) \
 	X(homer_ep) \
 	X(dtmf_udp_ep)