view plugins/mod_pep_simple.lua @ 11523:5f15ab7c6ae5

Statistics: Rewrite statistics backends to use OpenMetrics The metric subsystem of Prosody has had some shortcomings from the perspective of the current state-of-the-art in metric observability. The OpenMetrics standard [0] is a formalization of the data model (and serialization format) of the well-known and widely-used Prometheus [1] software stack. The previous stats subsystem of Prosody did not map well to that format (see e.g. [2] and [3]); the key reason is that it was trying to do too much math on its own ([2]) while lacking first-class support for "families" of metrics ([3]) and structured metric metadata (despite the `extra` argument to metrics, there was no standard way of representing common things like "tags" or "labels"). Even though OpenMetrics has grown from the Prometheus world of monitoring, it maps well to other popular monitoring stacks such as: - InfluxDB (labels can be mapped to tags and fields as necessary) - Carbon/Graphite (labels can be attached to the metric name with dot-separation) - StatsD (see graphite when assuming that graphite is used as backend, which is the default) The util.statsd module has been ported to use the OpenMetrics model as a proof of concept. An implementation which exposes the util.statistics backend data as Prometheus metrics is ready for publishing in prosody-modules (most likely as mod_openmetrics_prometheus to avoid breaking existing 0.11 deployments). At the same time, the previous measure()-based API had one major advantage: It is really simple and easy to use without requiring lots of knowledge about OpenMetrics or similar concepts. For that reason as well as compatibility with existing code, it is preserved and may even be extended in the future. However, code relying on the `stats-updated` event as well as `get_stats` from `statsmanager` will break because the data model has changed completely; in case of `stats-updated`, the code will simply not run (as the event was renamed in order to avoid conflicts); the `get_stats` function has been removed completely (so it will cause a traceback when it is attempted to be used). Note that the measure_*_event methods have been removed from the module API. I was unable to find any uses or documentation and thus deemed they should not be ported. Re-implementation is possible when necessary. [0]: https://openmetrics.io/ [1]: https://prometheus.io/ [2]: #959 [3]: #960
author Jonas Schäfer <jonas@wielicki.name>
date Sun, 18 Apr 2021 11:47:41 +0200
parents f73947a9bd8c
children 39ae08180c81
line wrap: on
line source

-- Prosody IM
-- Copyright (C) 2008-2010 Matthew Wild
-- Copyright (C) 2008-2010 Waqas Hussain
--
-- This project is MIT/X11 licensed. Please see the
-- COPYING file in the source package for more information.
--


local jid_bare = require "util.jid".bare;
local jid_split = require "util.jid".split;
local st = require "util.stanza";
local is_contact_subscribed = require "core.rostermanager".is_contact_subscribed;
local pairs = pairs;
local next = next;
local type = type;
local unpack = table.unpack or unpack; -- luacheck: ignore 113
local calculate_hash = require "util.caps".calculate_hash;
local core_post_stanza = prosody.core_post_stanza;
local bare_sessions = prosody.bare_sessions;

local xmlns_pubsub = "http://jabber.org/protocol/pubsub";

-- Used as canonical 'empty table'
local NULL = {};
-- data[user_bare_jid][node] = item_stanza
local data = {};
--- recipients[user_bare_jid][contact_full_jid][subscribed_node] = true
local recipients = {};
-- hash_map[hash][subscribed_nodes] = true
local hash_map = {};

module.save = function()
	return { data = data, recipients = recipients, hash_map = hash_map };
end
module.restore = function(state)
	data = state.data or {};
	recipients = state.recipients or {};
	hash_map = state.hash_map or {};
end

local function subscription_presence(user_bare, recipient)
	local recipient_bare = jid_bare(recipient);
	if (recipient_bare == user_bare) then return true end
	local username, host = jid_split(user_bare);
	return is_contact_subscribed(username, host, recipient_bare);
end

module:hook("pep-publish-item", function (event)
	local session, bare, node, id, item = event.session, event.user, event.node, event.id, event.item;
	item.attr.xmlns = nil;
	local disable = #item.tags ~= 1 or #item.tags[1] == 0;
	if #item.tags == 0 then item.name = "retract"; end
	local stanza = st.message({from=bare, type='headline'})
		:tag('event', {xmlns='http://jabber.org/protocol/pubsub#event'})
			:tag('items', {node=node})
				:add_child(item)
			:up()
		:up();

	-- store for the future
	local user_data = data[bare];
	if disable then
		if user_data then
			user_data[node] = nil;
			if not next(user_data) then data[bare] = nil; end
		end
	else
		if not user_data then user_data = {}; data[bare] = user_data; end
		user_data[node] = {id, item};
	end

	-- broadcast
	for recipient, notify in pairs(recipients[bare] or NULL) do
		if notify[node] then
			stanza.attr.to = recipient;
			core_post_stanza(session, stanza);
		end
	end
end);

local function publish_all(user, recipient, session)
	local d = data[user];
	local notify = recipients[user] and recipients[user][recipient];
	if d and notify then
		for node in pairs(notify) do
			if d[node] then
				-- luacheck: ignore id
				local id, item = unpack(d[node]);
				session.send(st.message({from=user, to=recipient, type='headline'})
					:tag('event', {xmlns='http://jabber.org/protocol/pubsub#event'})
						:tag('items', {node=node})
							:add_child(item)
						:up()
					:up());
			end
		end
	end
end

local function get_caps_hash_from_presence(stanza, current)
	local t = stanza.attr.type;
	if not t then
		for _, child in pairs(stanza.tags) do
			if child.name == "c" and child.attr.xmlns == "http://jabber.org/protocol/caps" then
				local attr = child.attr;
				if attr.hash then -- new caps
					if attr.hash == 'sha-1' and attr.node and attr.ver then return attr.ver, attr.node.."#"..attr.ver; end
				else -- legacy caps
					if attr.node and attr.ver then return attr.node.."#"..attr.ver.."#"..(attr.ext or ""), attr.node.."#"..attr.ver; end
				end
				return; -- bad caps format
			end
		end
	elseif t == "unavailable" or t == "error" then
		return;
	end
	return current; -- no caps, could mean caps optimization, so return current
end

module:hook("presence/bare", function(event)
	-- inbound presence to bare JID received
	local origin, stanza = event.origin, event.stanza;
	local user = stanza.attr.to or (origin.username..'@'..origin.host);
	local t = stanza.attr.type;
	local self = not stanza.attr.to;

	-- Only cache subscriptions if user is online
	if not bare_sessions[user] then return; end

	if not t then -- available presence
		if self or subscription_presence(user, stanza.attr.from) then
			local recipient = stanza.attr.from;
			local current = recipients[user] and recipients[user][recipient];
			local hash = get_caps_hash_from_presence(stanza, current);
			if current == hash or (current and current == hash_map[hash]) then return; end
			if not hash then
				if recipients[user] then recipients[user][recipient] = nil; end
			else
				recipients[user] = recipients[user] or {};
				if hash_map[hash] then
					recipients[user][recipient] = hash_map[hash];
					publish_all(user, recipient, origin);
				else
					recipients[user][recipient] = hash;
					local from_bare = origin.type == "c2s" and origin.username.."@"..origin.host;
					if self or origin.type ~= "c2s" or (recipients[from_bare] and recipients[from_bare][origin.full_jid]) ~= hash then
						-- COMPAT from ~= stanza.attr.to because OneTeam and Asterisk 1.8 can't deal with missing from attribute
						origin.send(
							st.stanza("iq", {from=user, to=stanza.attr.from, id="disco", type="get"})
								:query("http://jabber.org/protocol/disco#info")
						);
					end
				end
			end
		end
	elseif t == "unavailable" then
		if recipients[user] then recipients[user][stanza.attr.from] = nil; end
	elseif not self and t == "unsubscribe" then
		local from = jid_bare(stanza.attr.from);
		local subscriptions = recipients[user];
		if subscriptions then
			for subscriber in pairs(subscriptions) do
				if jid_bare(subscriber) == from then
					recipients[user][subscriber] = nil;
				end
			end
		end
	end
end, 10);

module:hook("iq/bare/http://jabber.org/protocol/pubsub:pubsub", function(event)
	local session, stanza = event.origin, event.stanza;
	local payload = stanza.tags[1];

	if stanza.attr.type == 'set' and (not stanza.attr.to or jid_bare(stanza.attr.from) == stanza.attr.to) then
		payload = payload.tags[1]; -- <publish node='http://jabber.org/protocol/tune'>
		if payload and (payload.name == 'publish' or payload.name == 'retract') and payload.attr.node then
			local node = payload.attr.node;
			payload = payload.tags[1];
			if payload and payload.name == "item" then -- <item>
				local id = payload.attr.id or "1";
				payload.attr.id = id;
				session.send(st.reply(stanza));
				module:fire_event("pep-publish-item", {
					node = node, user = jid_bare(session.full_jid), actor = session.jid,
					id = id, session = session, item = st.clone(payload);
				});
				return true;
			else
				module:log("debug", "Payload is missing the <item>", node);
			end
		else
			module:log("debug", "Unhandled payload: %s", payload and payload:top_tag() or "(no payload)");
		end
	elseif stanza.attr.type == 'get' then
		local user = stanza.attr.to and jid_bare(stanza.attr.to) or session.username..'@'..session.host;
		if subscription_presence(user, stanza.attr.from) then
			local user_data = data[user];
			local node, requested_id;
			payload = payload.tags[1];
			if payload and payload.name == 'items' then
				node = payload.attr.node;
				local item = payload.tags[1];
				if item and item.name == "item" then
					requested_id = item.attr.id;
				end
			end
			if node and user_data and user_data[node] then -- Send the last item
				local id, item = unpack(user_data[node]);
				if not requested_id or id == requested_id then
					local reply_stanza = st.reply(stanza)
						:tag('pubsub', {xmlns='http://jabber.org/protocol/pubsub'})
							:tag('items', {node=node})
								:add_child(item)
							:up()
						:up();
					session.send(reply_stanza);
					return true;
				else -- requested item doesn't exist
					local reply_stanza = st.reply(stanza)
						:tag('pubsub', {xmlns='http://jabber.org/protocol/pubsub'})
							:tag('items', {node=node})
						:up();
					session.send(reply_stanza);
					return true;
				end
			elseif node then -- node doesn't exist
				session.send(st.error_reply(stanza, 'cancel', 'item-not-found'));
				module:log("debug", "Item '%s' not found", node)
				return true;
			else --invalid request
				session.send(st.error_reply(stanza, 'modify', 'bad-request'));
				module:log("debug", "Invalid request: %s", payload);
				return true;
			end
		else --no presence subscription
			session.send(st.error_reply(stanza, 'auth', 'not-authorized')
				:tag('presence-subscription-required', {xmlns='http://jabber.org/protocol/pubsub#errors'}));
			module:log("debug", "Unauthorized request: %s", payload);
			return true;
		end
	end
end);

module:hook("iq-result/bare/disco", function(event)
	local session, stanza = event.origin, event.stanza;
	if stanza.attr.type == "result" then
		local disco = stanza.tags[1];
		if disco and disco.name == "query" and disco.attr.xmlns == "http://jabber.org/protocol/disco#info" then
			-- Process disco response
			local self = not stanza.attr.to;
			local user = stanza.attr.to or (session.username..'@'..session.host);
			local contact = stanza.attr.from;
			local current = recipients[user] and recipients[user][contact];
			if type(current) ~= "string" then return; end -- check if waiting for recipient's response
			local ver = current;
			if not string.find(current, "#") then
				ver = calculate_hash(disco.tags); -- calculate hash
			end
			local notify = {};
			for _, feature in pairs(disco.tags) do
				if feature.name == "feature" and feature.attr.var then
					local nfeature = feature.attr.var:match("^(.*)%+notify$");
					if nfeature then notify[nfeature] = true; end
				end
			end
			hash_map[ver] = notify; -- update hash map
			if self then
				for jid, item in pairs(session.roster) do -- for all interested contacts
					if item.subscription == "both" or item.subscription == "from" then
						if not recipients[jid] then recipients[jid] = {}; end
						recipients[jid][contact] = notify;
						publish_all(jid, contact, session);
					end
				end
			end
			recipients[user][contact] = notify; -- set recipient's data to calculated data
			-- send messages to recipient
			publish_all(user, contact, session);
		end
	end
end);

module:hook("account-disco-info", function(event)
	local reply = event.reply;
	reply:tag('identity', {category='pubsub', type='pep'}):up();
	reply:tag('feature', {var=xmlns_pubsub}):up();
	local features = {
		"access-presence",
		"auto-create",
		"auto-subscribe",
		"filtered-notifications",
		"item-ids",
		"last-published",
		"presence-notifications",
		"presence-subscribe",
		"publish",
		"retract-items",
		"retrieve-items",
	};
	for _, feature in ipairs(features) do
		reply:tag('feature', {var=xmlns_pubsub.."#"..feature}):up();
	end
end);

module:hook("account-disco-items", function(event)
	local reply = event.reply;
	local bare = reply.attr.to;
	local user_data = data[bare];

	if user_data then
		for node, _ in pairs(user_data) do
			reply:tag('item', {jid=bare, node=node}):up();
		end
	end
end);

module:hook("account-disco-info-node", function (event)
	local stanza, node = event.stanza, event.node;
	local user = stanza.attr.to;
	local user_data = data[user];
	if user_data and user_data[node] then
		event.exists = true;
		event.reply:tag('identity', {category='pubsub', type='leaf'}):up();
	end
end);

module:hook("resource-unbind", function (event)
	local user_bare_jid = event.session.username.."@"..event.session.host;
	if not bare_sessions[user_bare_jid] then -- User went offline
		-- We don't need this info cached anymore, clear it.
		recipients[user_bare_jid] = nil;
	end
end);