TT#34705 merge https://github.com/sipwise/prosody/pull/3

squashed commit of

commit 4a38d9ad2b (HEAD -> master)
Author: Victor Seva <linuxmaniac@torreviejawireless.org>
Date:   Mon Mar 26 14:15:41 2018 +0200

    [sipwise_redis_sessions] clean previous local sessions on start

commit a514e6b60d
Author: Victor Seva <linuxmaniac@torreviejawireless.org>
Date:   Mon Mar 26 14:08:31 2018 +0200

    [mod_shard] clean redis info on shard/send error

    if We can't connect to a shard node, clean that info from
    redis depending on the stanza

    MUC, persistent rooms are always defined in muc module on start, so don't
    check for them locally, set as local in redis if no other node has it

closes #3

Change-Id: I33b32dacb6aa4010d520c1d04ef81736d6555370

plugins/mod_shard.lua

@@ -41,14 +41,36 @@ local function get_local_rooms(host)
     return rooms;
 end
 
+local function is_persistent_room(jid)
+    local node, host, _ = jid_split(jid);
+    local bare = node..'@'..host;
+
+    if hosts[host] and hosts[host].muc.rooms[bare] then
+        return hosts[host].muc.rooms[bare]._data.persistent;
+    end
+    return false;
+end
+
+local function check_redis_info(room)
+    local rhost = redis_mucs.get_room_host(room);
+
+    if rhost ~= shard_name then
+        module:log("info", "clean wrong shard info for room[%s]", room);
+        redis_mucs.clean_room_host(room, rhost);
+        redis_mucs.set_room_host(room, shard_name);
+    end
+end
+
 local function handle_room_event(event)
     local to = event.stanza.attr.to;
     local node, host, _ = jid_split(to);
-    local rhost;
+    local rhost, bare;
 
     if node then
-        if hosts[host].muc.rooms[node] then
-            module:log("debug", "room[%s] is hosted here. Nothing to do", node);
+        bare = node..'@'..host;
+        if not is_persistent_room(bare) and hosts[host].muc.rooms[bare] then
+            module:log("debug", "room[%s] is hosted here. Nothing to do", bare);
+            check_redis_info(bare);
             return nil;
         end
         module:log("debug", "looking up target room shard for %s", to);
@@ -65,7 +87,14 @@ local function handle_room_event(event)
     end
 
     if not rhost then
-        module:log("debug", "room not found. Nothing to do");
+        assert(bare);
+        if is_persistent_room(bare) then
+            module:log("info",
+                "restore missing info for persistent room[%s]", bare);
+            redis_mucs.set_room_host(bare, shard_name);
+        else
+            module:log("debug", "room not found. Nothing to do");
+        end
         return nil;
     end
 
@@ -128,6 +157,21 @@ local function handle_event (event)
     return stop_process_local;
 end
 
+local function handle_shard_error(event)
+    local server_id = event.shard
+    local stanza = event.stanza
+    local jid = stanza.attr.to;
+    local _, host, _ = jid_split(jid);
+
+    if hosts[host].muc then
+        module:log("debug",
+            "to MUC %s detected, clean conference %s", host, jid);
+        redis_mucs.clean_room_host(jid, server_id)
+    else
+        redis_sessions.clean_host(jid, server_id)
+    end
+end
+
 module:hook("iq/bare", handle_event, 1000);
 module:hook("iq/full", handle_event, 1000);
 module:hook("iq/host", handle_event, 1000);
@@ -137,4 +181,5 @@ module:hook("message/host", handle_event, 1000);
 module:hook("presence/bare", handle_event, 1000);
 module:hook("presence/full", handle_event, 1000);
 module:hook("presence/host", handle_event, 1000);
+module:hook_global("shard/error", handle_shard_error);
 module:log("debug", "hooked at %s", module:get_host());

plugins/mod_shard_client.lua

@@ -199,12 +199,12 @@ local function handle_send(event)
         conn, err = connect(shard);
         if not conn then
             module:log("error", "couldn't connect to "..shard..": "..err);
+            module:fire_event("shard/error", {shard = shard, stanza = stanza});
             return;
         end
         conns[shard] = conn;
         queue[shard] = {};
     end
-
     if stanza.attr.via then
         local via = ut.string.explode(';', stanza.attr.via);
         module:log("debug", "via:%s", ut.table.tostring(via));

plugins/mod_sipwise_redis_mucs.lua

@@ -33,35 +33,16 @@ end
 
 local function muc_created(event)
 	local room = event.room;
-	local node, host, _ = jid.split(room.jid)
 
+	redis_mucs.set_room_host(room.jid, redis_config.server_id);
 	module:log("debug", "muc-room-created %s", room.jid);
-
-	if not test_connection() then client_connect() end
-	-- TODO: check that there is no other "room.jid" value?
-	if redis_client:set(room.jid, redis_config.server_id) then
-		module:log("debug", "save [%s]=%s", room.jid, redis_config.server_id);
-	end
-	if redis_client:sadd(host, redis_config.server_id..":"..node) > 0 then
-		module:log("debug", "append [%s]=>%s:%s", host,
-			redis_config.server_id, node);
-	end
 end
 
 local function muc_destroyed(event)
 	local room = event.room;
-	local node, host, _ = jid.split(room.jid)
 
+	redis_mucs.clean_room_host(room.jid, redis_config.server_id);
 	module:log("debug", "muc-room-destroyed %s", room.jid);
-
-	if not test_connection() then client_connect() end
-	if redis_client:del(room.jid) > 0 then
-		module:log("debug", "remove [%s]=%s", room.jid, redis_config.server_id);
-	end
-	if redis_client:srem(host, redis_config.server_id..":"..node) > 0 then
-		module:log("debug", "remove [%s]=>%s:%s", host,
-			redis_config.server_id, node);
-	end
 end
 
 local function split_key(key)
@@ -85,8 +66,23 @@ function redis_mucs.get_rooms(host)
 	return res;
 end
 
+function redis_mucs.set_room_host(room_jid, server_id)
+	local node, host, _ = jid.split(room_jid);
+	local bare_jid = node.."@"..host;
+
+	if not test_connection() then client_connect() end
+	-- TODO: check that there is no other "bare_jid" value?
+	if redis_client:set(bare_jid, server_id) then
+		module:log("debug", "save [%s]=%s", bare_jid, server_id);
+	end
+	if redis_client:sadd(host, server_id..":"..node) > 0 then
+		module:log("debug", "append [%s]=>%s:%s", host,
+			server_id, node);
+	end
+end
+
 function redis_mucs.get_room_host(room_jid)
-	local node, domain = jid.split(room_jid);
+	local node, domain, _ = jid.split(room_jid);
 	local bare_jid = node.."@"..domain;
 
 	module:log("debug", "search room:%s host", bare_jid);
@@ -94,6 +90,19 @@ function redis_mucs.get_room_host(room_jid)
 	return redis_client:get(bare_jid);
 end
 
+function redis_mucs.clean_room_host(room_jid, server_id)
+	local node, host, _ = jid.split(room_jid);
+	local bare_jid = node.."@"..host;
+
+	if not test_connection() then client_connect() end
+	if redis_client:del(bare_jid) > 0 then
+		module:log("debug", "remove [%s]=%s", bare_jid, server_id);
+	end
+	if redis_client:srem(host, server_id..":"..node) > 0 then
+		module:log("debug", "remove [%s]=>%s:%s", host, server_id, node);
+	end
+end
+
 function module.load()
 	redis_config = module:get_option("redis_sessions_auth", redis_config);
 end

plugins/mod_sipwise_redis_sessions.lua

@@ -12,7 +12,8 @@ local array = require "util.array";
 local redis = require 'redis';
 local redis_config = {
 	port = 6739, host = "127.0.0.1",
-	server_id = "0", redis_db = "2"
+	server_id = "0", redis_db = "2",
+	clean_local_sessions = true,
 };
 local redis_client;
 
@@ -48,17 +49,23 @@ local function resource_bind(event)
 	redis_client:sadd(bare_jid, redis_config.server_id..":"..resource);
 end
 
-local function resource_unbind(event)
-	local session, _ = event.session, event.error;
-	local node, domain, resource = jid.split(session.full_jid);
-	local full_jid, bare_jid = session.full_jid, node.."@"..domain;
+local function remove_resource(full_jid)
+	local node, domain, resource = jid.split(full_jid);
+	local bare_jid = node.."@"..domain;
 
-	module:log("debug", "resource-unbind from %s", session.host);
-	module:log("debug", "remove [%s]=%s", full_jid, redis_config.server_id);
 	if not test_connection() then client_connect() end
 	redis_client:del(full_jid);
-	module:log("debug", "remove [%s]=>%s:%s", bare_jid, redis_config.server_id, resource);
+	module:log("debug", "remove [%s]=%s", full_jid, redis_config.server_id);
 	redis_client:srem(bare_jid, redis_config.server_id..":"..resource);
+	module:log("debug", "remove [%s]=>%s:%s", bare_jid,
+		redis_config.server_id, resource);
+end
+
+local function resource_unbind(event)
+	local session, _ = event.session, event.error;
+
+	module:log("debug", "resource-unbind from %s", session.host);
+	remove_resource(session.full_jid);
 end
 
 local function split_key(key)
@@ -66,6 +73,23 @@ local function split_key(key)
 	return t[1], t[2];
 end
 
+local function clean_local_sessions(host)
+	if not test_connection() then client_connect() end
+	local l = redis_client:keys('*@'..host);
+	for _, bare_jid in ipairs(l) do
+		local resources = redis_client:smembers(bare_jid);
+		module:log("debug", "clean previous sessions for %s on %s",
+			bare_jid, redis_config.server_id);
+		for _, resource in ipairs(resources) do
+			local server, old_resource = split_key(resource);
+			if server == redis_config.server_id then
+				module:log("debug", "remove old resource %s", old_resource);
+				remove_resource(bare_jid..'/'..old_resource);
+			end
+		end
+	end
+end
+
 function redis_sessions.get_hosts(j)
 	local node, domain = jid.split(j);
 	local bare_jid = node.."@"..domain;
@@ -85,12 +109,32 @@ function redis_sessions.get_hosts(j)
 	return res;
 end
 
+function redis_sessions.clean_host(j, server_id)
+	local bare_jid = jid.bare(j);
+	module:log("debug", "clean jid %s from %s", bare_jid, server_id);
+	if not test_connection() then client_connect() end
+	local l = redis_client:smembers(bare_jid);
+	for _,v in pairs(l) do
+		local h, _ = split_key(v);
+		if h == server_id then
+			redis_client:srem(bare_jid, v);
+			redis_client:del(v);
+			module:log("debug", "removed %s from %s", v, bare_jid);
+		end
+	end
+end
+
 function module.load()
 	redis_config = module:get_option("redis_sessions_auth", redis_config);
 end
 
 function module.add_host(module)
-	module:hook("resource-bind", resource_bind, 200);
-	module:hook("resource-unbind", resource_unbind, 200);
-	module:log("debug", "hooked at %s", module:get_host());
+	local host = module:get_host();
+
+	if module:get_host_type() ~= "component" then
+		module:hook("resource-bind", resource_bind, 200);
+		module:hook("resource-unbind", resource_unbind, 200);
+		clean_local_sessions(host);
+		module:log("debug", "hooked at %s", host);
+	end
 end