Squashed 'contrib/pg_shardman/' changes from c350eaf..4c97bed

4c97bed Add monitor function
07ff8f9 Merge branch 'broadcast' of https://git.postgrespro.ru/a.sher/pg_shardman into broadcast
e7cfaa5 Add monitor_deadlocks function

git-subtree-dir: contrib/pg_shardman
git-subtree-split: 4c97bed

Author: arssher

pg_shardman--0.0.2.sql

@@ -2007,41 +2007,104 @@ BEGIN
 	LOOP
 		poll := format('%s%s:SELECT shardman.serialize_lock_graph();', poll, node_id);
 	END LOOP;
-	SELECT shardman.broadcast(poll) INTO graph;
+	SELECT shardman.broadcast(poll, ignore_errors => true) INTO graph;
 
 	RETURN graph;
 END;
 $$ LANGUAGE plpgsql;
 
 
--- Detect distributed deadlock and return set of process involed in deadlock. If there is no deadlock then this view ias empty.
---
--- This query is based on the algorithm described by Knuth for detecting a cycle in a linked list. In one column, keep track of the children,
--- the children's children, the children's children's children, etc. In another column, keep track of the grandchildren, the grandchildren's grandchildren,
--- the grandchildren's grandchildren's grandchildren, etc.
---
--- For the initial selection, the distance between Child and Grandchild columns is 1. Every selection from union all increases the depth of Child by 1, and that of Grandchild by 2.
--- The distance between them increases by 1.
---
--- If there is any loop, since the distance only increases by 1 each time, at some point after Child is in the loop, the distance will be a multiple of the cycle length.
--- When that happens, the Child and the Grandchild columns are the same. Use that as an additional condition to stop the recursion, and detect it in the rest of your code as an error.
-CREATE VIEW deadlock AS
-	WITH RECURSIVE LinkTable AS (SELECT wait AS Parent, hold AS Child FROM shardman.deserialize_lock_graph(shardman.global_lock_graph())),
+-- Detect distributed deadlock and returns path in the lock graph forming deadlock loop
+CREATE FUNCTION detect_deadlock(lock_graph text) RETURNS shardman.process[] AS $$
+	WITH RECURSIVE LinkTable AS (SELECT wait AS Parent, hold AS Child FROM shardman.deserialize_lock_graph(lock_graph)),
 	cte AS (
-    	SELECT lt1.Parent, lt1.Child, lt2.Child AS Grandchild
-    	FROM LinkTable lt1
-    	INNER JOIN LinkTable lt2 on lt2.Parent = lt1.Child
-    	UNION ALL
-    	SELECT cte.Parent, lt1.Child, lt3.Child AS Grandchild
-    	FROM cte
-    	INNER JOIN LinkTable lt1 ON lt1.Parent = cte.Child
-    	INNER JOIN LinkTable lt2 ON lt2.Parent = cte.Grandchild
-    	INNER JOIN LinkTable lt3 ON lt3.Parent = lt2.Child
-    	WHERE cte.Child <> cte.Grandchild
+		SELECT Child, Parent, ARRAY[Child] AS AllParents, false AS Loop
+  	  	FROM LinkTable
+	  	UNION ALL
+	  	SELECT c.Child, c.Parent, p.AllParents||c.Child, c.Child=ANY(p.AllParents)
+	 	FROM LinkTable c JOIN cte p	ON c.Parent = p.Child AND NOT p.Loop
 	)
-	SELECT DISTINCT Parent
-	FROM cte
-	WHERE Child = Grandchild;
+	SELECT AllParents FROM cte WHERE Loop;
+$$ LANGUAGE sql;
+
+-- Monitor cluster for presence of distributed deadlocks and node failures.
+-- Tries to cancel queries causing deadlock and exclude unavailable nodes from the cluser.
+CREATE FUNCTION monitor(deadlock_check_timeout_sec int = 5, rm_node_timeout_sec int = 60) RETURNS void AS $$
+DECLARE
+	prev_deadlock_path shardman.process[];
+	deadlock_path shardman.process[];
+	victim shardman.process;
+	loop_begin int;
+	loop_end int;
+	prev_loop_begin int;
+	prev_loop_end int;
+	sep int;
+	resp text;
+	error_begin int;
+	error_end int;
+	error_msg text;
+	error_node_id int;
+	failed_node_id int;
+	failure_timestamp timestamp with time zone;
+BEGIN
+	IF shardman.redirect_to_shardlord(format('monitor(%s, %s)', deadlock_check_timeout_sec, rm_node_timeout_sec))
+	THEN
+		RETURN;
+	END IF;
+
+	RAISE NOTICE 'Start cluster monitor...';
+
+	LOOP
+		resp := shardman.global_lock_graph();
+		error_begin := position('<error>' IN resp);
+		IF error_begin<>0
+		THEN
+			error_end := position('</error>' IN resp);
+			sep := position(':' IN resp);
+			error_node_id := substring(resp FROM error_begin+7 FOR sep-error_begin-7)::int;
+			error_msg := substring(resp FROM sep+1 FOR error_end-sep-1);
+			IF error_node_id = failed_node_id
+			THEN
+				IF clock_timestamp() > failure_timestamp + rm_node_timeout_sec * interval '1 sec'
+				THEN
+					RAISE NOTICE 'Remove node % because of % timeout expiration', failed_node_id, rm_node_timeout_sec;
+					PERFORM shardman.broadcast(format('0:SELECT shardman.rm_node(%s, force=>true);', failed_node_id));
+					failed_node_id := null;
+				END IF;
+			ELSE
+				RAISE NOTICE 'Node % reports error message %', error_node_id, error_msg;
+				failed_node_id := error_node_id;
+				failure_timestamp := clock_timestamp();
+			END IF;
+			prev_deadlock_path := null;
+		ELSE
+			failed_node_id := null;
+			deadlock_path := shardman.detect_deadlock(resp);
+			loop_end := array_upper(deadlock_path, 1);
+			loop_begin := array_position(deadlock_path, deadlock_path[loop_end]);
+			-- Check if old and new lock graph contain the same subgraph.
+			-- Because we can not make consistent distributed snapshot,
+			-- collected global local graph can contain "false" loops.
+			-- So we report deadlock only if detected loop persists during
+			-- deadlock detection period.
+			IF prev_deadlock_path IS NOT NULL
+			   AND loop_end - loop_begin = prev_loop_end - prev_loop_begin
+			   AND deadlock_path[loop_begin:loop_end] = prev_deadlock_path[prev_loop_begin:prev_loop_end]
+			THEN
+				-- Try to cancel random node in loop
+				victim := deadlock_path[loop_begin + ((loop_end - loop_begin)*random())::integer];
+				RAISE NOTICE 'Detect deadlock: cancel process % at node %', victim.pid, victim.node;
+				PERFORM shardman.broadcast(format('%s:SELECT pg_cancel_backend(%s);',
+					victim.node, victim.pid));
+			END IF;
+			prev_deadlock_path := deadlock_path;
+			prev_loop_begin := loop_begin;
+			prev_loop_end := loop_end;
+		END IF;
+		PERFORM pg_sleep(deadlock_check_timeout_sec);
+	END LOOP;
+END;
+$$ LANGUAGE plpgsql;
 
 
 -- View for monitoring logical replication lag.

pg_shardman.c

@@ -122,10 +122,17 @@ wait_command_completion(PGconn* conn)
 	return  true;
 }
 
+typedef struct
+{
+	PGconn* con;
+	char*   sql;
+	int     node;
+} Channel;
+
 Datum
 broadcast(PG_FUNCTION_ARGS)
 {
-	char *sql_full = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	char* sql_full = text_to_cstring(PG_GETARG_TEXT_PP(0));
 	char* cmd = pstrdup(sql_full);
 	bool  ignore_errors = PG_GETARG_BOOL(1);
 	bool  two_phase = PG_GETARG_BOOL(2);
@@ -143,17 +150,18 @@ broadcast(PG_FUNCTION_ARGS)
 	int   n_cmds = 0;
 	int   i;
 	int n_cons = 1024;
-	PGconn** conn;
+	Channel* chan;
+	PGconn* con;
 	StringInfoData resp;
 
-	char* errmsg = NULL;
+	char const* errmsg = "";
 
 	elog(DEBUG1, "Broadcast commmand '%s'",  cmd);
 
 	initStringInfo(&resp);
 
 	SPI_connect();
-	conn = (PGconn**) palloc(sizeof(PGconn*) * n_cons);
+	chan = (Channel*) palloc(sizeof(Channel) * n_cons);
 
 	while ((sep = strchr(cmd, *cmd == '{' ? '}' : ';')) != NULL)
 	{
@@ -163,7 +171,7 @@ broadcast(PG_FUNCTION_ARGS)
 			cmd += 1;
 		rc = sscanf(cmd, "%d:%n", &node_id, &n);
 		if (rc != 1) {
-			elog(ERROR, "SHARDMAN: Invalid command string: %s", cmd);
+			elog(ERROR, "SHARDMAN: Invalid command string: '%s' in '%s'", cmd, sql_full);
 		}
 		sql = cmd + n;
 		cmd = sep + 1;
@@ -191,21 +199,26 @@ broadcast(PG_FUNCTION_ARGS)
 		}
 		if (n_cmds >= n_cons)
 		{
-			conn = (PGconn**) repalloc(conn, sizeof(PGconn*) * (n_cons *= 2));
+			chan = (Channel*) repalloc(chan, sizeof(Channel) * (n_cons *= 2));
 		}
 
-		conn[n_cmds] = PQconnectdb(conn_str);
-		if (PQstatus(conn[n_cmds++]) != CONNECTION_OK)
+		con = PQconnectdb(conn_str);
+		chan[n_cmds].con = con;
+		chan[n_cmds].node = node_id;
+		chan[n_cmds].sql = sql;
+		n_cmds += 1;
+
+		if (PQstatus(con) != CONNECTION_OK)
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s%d:Connection failure: %s.",
-								  errmsg ? errmsg : "", node_id,
-								  PQerrorMessage(conn[n_cmds - 1]));
+				errmsg = psprintf("%s<error>%d:Connection failure: %s</error>",
+								  errmsg, node_id, PQerrorMessage(con));
+				chan[n_cmds-1].sql = NULL;
 				continue;
 			}
 			errmsg = psprintf("Failed to connect to node %d: %s", node_id,
-							  PQerrorMessage(conn[n_cmds-1]));
+							  PQerrorMessage(con));
 			goto cleanup;
 		}
 		if (!sync_commit_on)
@@ -221,18 +234,18 @@ broadcast(PG_FUNCTION_ARGS)
 			}
 		}
 		elog(DEBUG1, "Send command '%s' to node %d", sql, node_id);
-		if (!PQsendQuery(conn[n_cmds - 1], sql)
-			|| (sequential && !wait_command_completion(conn[n_cmds - 1])))
+		if (!PQsendQuery(con, sql)
+			|| (sequential && !wait_command_completion(con)))
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s%d:Failed to send query '%s': %s'.",
-								  errmsg ? errmsg : "", node_id, sql,
-								  PQerrorMessage(conn[n_cmds-1]));
+				errmsg = psprintf("%s<error>%d:Failed to send query '%s': %s</error>",
+								  errmsg, node_id, sql, PQerrorMessage(con));
+				chan[n_cmds-1].sql = NULL;
 				continue;
 			}
 			errmsg = psprintf("Failed to send query '%s' to node %d: %s'", sql,
-							  node_id, PQerrorMessage(conn[n_cmds-1]));
+							  node_id, PQerrorMessage(con));
 			goto cleanup;
 		}
 	}
@@ -248,7 +261,15 @@ broadcast(PG_FUNCTION_ARGS)
 		PGresult* res = NULL;
 		ExecStatusType status;
 
-		while ((next_res = PQgetResult(conn[i])) != NULL)
+		con = chan[i].con;
+
+		if (chan[i].sql == NULL)
+		{
+			/* Ignore commands which were not sent */
+			continue;
+		}
+
+		while ((next_res = PQgetResult(con)) != NULL)
 		{
 			if (res != NULL)
 			{
@@ -260,10 +281,12 @@ broadcast(PG_FUNCTION_ARGS)
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s%d:Failed to received response for '%s': %s.", errmsg ? errmsg : "", node_id, sql_full, PQerrorMessage(conn[i]));
+				errmsg = psprintf("%s<error>%d:Failed to received response for '%s': %s</error>",
+								  errmsg, chan[i].node, chan[i].sql, PQerrorMessage(con));
 				continue;
 			}
-			errmsg = psprintf("Failed to receive response for query %s from node %d: %s", sql_full, node_id, PQerrorMessage(conn[i]));
+			errmsg = psprintf("Failed to receive response for query %s from node %d: %s",
+							  chan[i].sql, chan[i].node, PQerrorMessage(con));
 			goto cleanup;
 		}
 
@@ -273,11 +296,13 @@ broadcast(PG_FUNCTION_ARGS)
 		{
 			if (ignore_errors)
 			{
-				errmsg = psprintf("%s%d:Command %s failed: %s.", errmsg ? errmsg : "", node_id, sql_full, PQerrorMessage(conn[i]));
+				errmsg = psprintf("%s<error>%d:Command %s failed: %s</error>",
+								  errmsg, chan[i].node, chan[i].sql, PQerrorMessage(con));
 				PQclear(res);
 				continue;
 			}
-			errmsg = psprintf("Command %s failed at node %d: %s", sql_full, node_id, PQerrorMessage(conn[i]));
+			errmsg = psprintf("Command %s failed at node %d: %s",
+							  chan[i].sql, chan[i].node, PQerrorMessage(con));
 			PQclear(res);
 			goto cleanup;
 		}
@@ -292,11 +317,13 @@ broadcast(PG_FUNCTION_ARGS)
 				if (ignore_errors)
 				{
 					appendStringInfoString(&resp, "?");
-					elog(WARNING, "SHARDMAN: Query '%s' doesn't return single tuple at node %d", sql_full, node_id);
+					elog(WARNING, "SHARDMAN: Query '%s' doesn't return single tuple at node %d",
+						 chan[i].sql, chan[i].node);
 				}
 				else
 				{
-					errmsg = psprintf("Query '%s' doesn't return single tuple at node %d", sql_full, node_id);
+					errmsg = psprintf("Query '%s' doesn't return single tuple at node %d",
+									  chan[i].sql, chan[i].node);
 					PQclear(res);
 					goto cleanup;
 				}
@@ -315,37 +342,39 @@ broadcast(PG_FUNCTION_ARGS)
   cleanup:
 	for (i = 0; i < n_cmds; i++)
 	{
+		con = chan[i].con;
 		if (two_phase)
 		{
-			if (errmsg)
+			if (*errmsg)
 			{
-				res = PQexec(conn[i], "ROLLBACK PREPARED 'shardlord'");
+				res = PQexec(con, "ROLLBACK PREPARED 'shardlord'");
 				if (PQresultStatus(res) != PGRES_COMMAND_OK)
 				{
 					elog(WARNING, "SHARDMAN: Rollback of 2PC failed at node %d: %s",
-						 node_id, PQerrorMessage(conn[i]));
+						 chan[i].node, PQerrorMessage(con));
 				}
 				PQclear(res);
 			}
 			else
 			{
-				res = PQexec(conn[i], "COMMIT PREPARED 'shardlord'");
+				res = PQexec(con, "COMMIT PREPARED 'shardlord'");
 				if (PQresultStatus(res) != PGRES_COMMAND_OK)
 				{
 					elog(WARNING, "SHARDMAN: Commit of 2PC failed at node %d: %s",
-						 node_id, PQerrorMessage(conn[i]));
+						 chan[i].node, PQerrorMessage(con));
 				}
 				PQclear(res);
 			}
 		}
-		PQfinish(conn[i]);
+		PQfinish(con);
 	}
 
-	if (errmsg)
+	if (*errmsg)
 	{
 		if (ignore_errors)
 		{
-			appendStringInfo(&resp, "Error:%s", errmsg);
+			resetStringInfo(&resp);
+			appendStringInfoString(&resp, errmsg);
 			elog(WARNING, "SHARDMAN: %s", errmsg);
 		}
 		else
@@ -354,7 +383,7 @@ broadcast(PG_FUNCTION_ARGS)
 		}
 	}
 
-	pfree(conn);
+	pfree(chan);
 	SPI_finish();
 
 	PG_RETURN_TEXT_P(cstring_to_text(resp.data));
@@ -366,7 +395,7 @@ broadcast(PG_FUNCTION_ARGS)
  * defaults, everything. Parameter is not REGCLASS because pg_dump can't
  * handle oids anyway. Connstring must be proper libpq connstring, it is feed
  * to pg_dump.
- * TODO: actually we should have much more control on what is dumped, so we
+ * TODO: actually we should have muchmore control on what is dumped, so we
  * need to copy-paste parts of messy pg_dump or collect the needed data
  * manually walking over catalogs.
  */