Sane logging.

Now we store node id in shmem and prepend [SHND node_id] to all log messages if
'%z' if specified in log line prefix. Besides, we add [SHMN] to any
shardman-related log message -- via macro in C and unfortunately manually in
plpgsql.

Author: arssher

init.sql

@@ -18,8 +18,7 @@ BEGIN
 -- Yes, malicious user might have another extension containing 'pg_shardman'...
 -- Probably better just call no-op func from the library
 	IF strpos(current_setting('shared_preload_libraries'), 'pg_shardman') = 0 THEN
-		RAISE EXCEPTION 'pg_shardman must be loaded via shared_preload_libraries. Refusing to
-						proceed.';
+		RAISE EXCEPTION 'pg_shardman must be loaded via shared_preload_libraries. Refusing to proceed.';
 	END IF;
 END
 $$;
@@ -173,7 +172,7 @@ DECLARE
 	lord_connstring text;
 	lord_id int;
 BEGIN
-	raise INFO 'Booting lord';
+	RAISE INFO '[SHMN] Booting lord';
 	PERFORM shardman.create_meta_pub();
 
 	lord_id := shardman.my_id();
@@ -232,12 +231,12 @@ CREATE FUNCTION drop_repslot(slot_name text, with_fire bool DEFAULT false)
 DECLARE
 	slot_exists bool;
 BEGIN
-	RAISE DEBUG '[SHARDMAN] Dropping repslot %', slot_name;
+	RAISE DEBUG '[SHMN] Dropping repslot %', slot_name;
 	EXECUTE format('SELECT EXISTS (SELECT * FROM pg_replication_slots
 				   WHERE slot_name = %L)', slot_name) INTO slot_exists;
 	IF slot_exists THEN
 		IF with_fire THEN -- kill walsender twice
-			RAISE DEBUG '[SHARDMAN] Killing repslot % with fire', slot_name;
+			RAISE DEBUG '[SHMN] Killing repslot % with fire', slot_name;
 			PERFORM shardman.terminate_repslot_walsender(slot_name);
 			PERFORM pg_sleep(1);
 			PERFORM shardman.terminate_repslot_walsender(slot_name);
@@ -268,8 +267,7 @@ CREATE FUNCTION eliminate_sub(subname name, drop_sub bool DEFAULT true)
 DECLARE
 	sub_exists bool;
 BEGIN
-	RAISE DEBUG '[SHARDMAN %] eliminating sub %, drop_sub %',
-		shardman.my_id(), subname, drop_sub;
+	RAISE DEBUG '[SHMN] eliminating sub %, drop_sub %', subname, drop_sub;
 	EXECUTE format('SELECT EXISTS (SELECT 1 FROM pg_subscription WHERE subname
 				   = %L)', subname) INTO sub_exists;
 	IF sub_exists THEN
@@ -300,7 +298,7 @@ DECLARE
 	sub record;
 	rs record;
 BEGIN
-	RAISE DEBUG '[SHARDMAN %] pg_shardman is dropping, cleaning up', shardman.my_id();
+	RAISE DEBUG '[SHMN] pg_shardman is dropping, cleaning up';
 
 	FOR pub IN SELECT pubname FROM pg_publication WHERE pubname LIKE 'shardman_%' LOOP
 		EXECUTE format('DROP PUBLICATION %I', pub.pubname);

membership.sql

@@ -36,7 +36,7 @@ CREATE UNIQUE INDEX unique_node_connstr ON shardman.nodes (connstring)
 CREATE FUNCTION rm_node_worker_side() RETURNS TRIGGER AS $$
 BEGIN
 	IF OLD.id = (SELECT shardman.my_id()) THEN
-		RAISE DEBUG '[SHARDMAN] rm_node_worker_side called';
+		RAISE DEBUG '[SHMN] rm_node_worker_side called';
 		PERFORM shardman.pg_shardman_cleanup(false);
 	END IF;
 	RETURN NULL;
@@ -105,7 +105,7 @@ DECLARE
 	connstr text := connstring FROM shardman.nodes WHERE id = shardman.my_id();
 BEGIN
 	IF connstr IS NULL THEN
-		RAISE EXCEPTION 'Node not in cluster, can''t get its connstring';
+		RAISE EXCEPTION '[SHMN] Node not in cluster, can''t get its connstring';
 	END IF;
 	RETURN connstr;
 END $$ LANGUAGE plpgsql;
@@ -118,7 +118,7 @@ DECLARE
 				worker_status IS NOT NULL;
 BEGIN
 	IF connstr IS NULL THEN
-		RAISE EXCEPTION 'Worker node with id % not found', node_id;
+		RAISE EXCEPTION '[SHMN] Worker node with id % not found', node_id;
 	END IF;
 	RETURN connstr;
 END $$ LANGUAGE plpgsql STRICT;

shard.sql

@@ -95,8 +95,8 @@ CREATE TABLE partitions (
 -- it.
 CREATE FUNCTION new_primary() RETURNS TRIGGER AS $$
 BEGIN
-	RAISE DEBUG '[SHARDMAN] new_primary trigger called on node % for part %, owner %',
-		shardman.my_id(), NEW.part_name, NEW.owner;
+	RAISE DEBUG '[SHMN] new_primary trigger called for part %, owner %',
+		NEW.part_name, NEW.owner;
 	IF NEW.owner != shardman.my_id() THEN
 		PERFORM shardman.replace_usual_part_with_foreign(NEW);
 	END IF;
@@ -190,8 +190,8 @@ DECLARE
 	src_next_lname text;
 BEGIN
 	ASSERT NEW.owner != OLD.owner, 'part_moved handles only moved parts';
-	RAISE DEBUG '[SHARDMAN %] part_moved trigger called for part %, owner % -> %',
-		me, NEW.part_name, OLD.owner, NEW.owner;
+	RAISE DEBUG '[SHMN] part_moved trigger called for part %, owner % -> %',
+		NEW.part_name, OLD.owner, NEW.owner;
 	ASSERT NEW.nxt = OLD.nxt OR (NEW.nxt IS NULL AND OLD.nxt IS NULL),
 		'both part and replica must not be moved in one update';
 	ASSERT NEW.prv = OLD.prv OR (NEW.prv IS NULL AND OLD.prv IS NULL),
@@ -223,8 +223,7 @@ BEGIN
 		-- Drop old table anyway;
 		EXECUTE format('DROP TABLE IF EXISTS %I', NEW.part_name);
 	ELSEIF me = NEW.owner THEN -- dst node
-		RAISE DEBUG '[SHARDMAN %] part_moved trigger on dst node',
-			me;
+		RAISE DEBUG '[SHMN] part_moved trigger on dst node';
 		-- Drop subscription used for copy
 		PERFORM shardman.eliminate_sub(cp_logname);
 		-- If primary part was moved, replace moved table with foreign one
@@ -371,19 +370,17 @@ DECLARE
 	me int := shardman.my_id();
 	my_part shardman.partitions;
 BEGIN
-	RAISE DEBUG '[SHARDMAN %] update_fdw_server called for part %, owner %',
-		shardman.my_id(), part.part_name, part.owner;
+	RAISE DEBUG '[SHMN] update_fdw_server called for part %, owner %',
+		part.part_name, part.owner;
 
 	SELECT * FROM shardman.partitions WHERE part_name = part.part_name AND
 											owner = me INTO my_part;
 	IF my_part.part_name IS NOT NULL THEN -- we are holding the part
 		IF my_part.prv IS NULL THEN
-			RAISE DEBUG '[SHARDMAN %] we are holding primary for part %, not updating fdw server for it',
-			me, part.part_name;
+			RAISE DEBUG '[SHMN] we are holding primary for part %, not updating fdw server for it', part.part_name;
 			RETURN;
 		ELSE
-			RAISE DEBUG '[SHARDMAN %] we are holding replica for part %, updating fdw server for it',
-			me, part.part_name;
+			RAISE DEBUG '[SHMN] we are holding replica for part %, updating fdw server for it', part.part_name;
 		END IF;
 	END IF;
 
@@ -446,8 +443,7 @@ BEGIN
 	-- and dangerous: what if table was created and dropped before this
 	-- change reached us? We might also use it with local table (create
 	-- foreign server pointing to it, etc), but that's just ugly.
-	RAISE DEBUG '[SHARDMAN] my id: %, creating ft %',
-		shardman.my_id(), part.part_name;
+	RAISE DEBUG '[SHMN] creating ft %', part.part_name;
 	EXECUTE format('CREATE FOREIGN TABLE %I %s SERVER %I OPTIONS (table_name %L)',
 				   fdw_part_name,
 				   (SELECT
@@ -551,7 +547,7 @@ END
 $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION go_away() RETURNS TRIGGER AS $$
 BEGIN
-	RAISE EXCEPTION 'The "%" table is read only.', TG_TABLE_NAME
+	RAISE EXCEPTION '[SHMN] The "%" table is read only.', TG_TABLE_NAME
 		USING HINT = 'Probably table copy is in progress';
 END;
 $$ LANGUAGE plpgsql;
@@ -567,7 +563,8 @@ END $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION readonly_replica_on(relation regclass)
 	RETURNS void AS $$
 BEGIN
-	RAISE DEBUG '[SHARDMAN] table % made read-only for all but apply workers', relation;
+	RAISE DEBUG '[SHMN] table % made read-only for all but apply workers',
+		relation;
 	PERFORM shardman.readonly_replica_off(relation);
 	PERFORM shardman.create_modification_triggers(
 		relation, 'shardman_readonly_replica', 'shardman.ror_go_away()');
@@ -580,7 +577,7 @@ END $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION ror_go_away() RETURNS TRIGGER AS $$
 BEGIN
 	IF NOT shardman.inside_apply_worker() THEN
-		RAISE EXCEPTION 'The "%" table is read only for non-apply workers', TG_TABLE_NAME
+		RAISE EXCEPTION '[SHMN] The "%" table is read only for non-apply workers', TG_TABLE_NAME
 		USING HINT =
 		'If you see this, most probably node with primary part has failed and' ||
 		' you need to promote replica. Promotion is not yet implemented, sorry :(';
@@ -685,7 +682,7 @@ DECLARE
 	newval text := shardman.ensure_sync_standby_c(standby);
 BEGIN
 	IF newval IS NOT NULL THEN
-		RAISE DEBUG '[SHARDMAN] Adding standby %, new value is %', standby, newval;
+		RAISE DEBUG '[SHMN] Adding standby %, new value is %', standby, newval;
 		PERFORM shardman.set_sync_standbys(newval);
 	END IF;
 END $$ LANGUAGE plpgsql STRICT;
@@ -699,7 +696,7 @@ DECLARE
 	newval text := shardman.remove_sync_standby_c(standby);
 BEGIN
 	IF newval IS NOT NULL THEN
-		RAISE DEBUG '[SHARDMAN] Removing standby %, new value is %', standby, newval;
+		RAISE DEBUG '[SHMN] Removing standby %, new value is %', standby, newval;
 		PERFORM shardman.set_sync_standbys(newval);
 	END IF;
 END $$ LANGUAGE plpgsql STRICT;
@@ -710,7 +707,7 @@ CREATE FUNCTION set_sync_standbys(standby text) RETURNS void AS $$
 BEGIN
 	PERFORM pg_reload_conf();
 	PERFORM shardman.set_sync_standbys_c(standby);
-	RAISE DEBUG '[SHARDMAN] sync_standbys set to %', standby;
+	RAISE DEBUG '[SHMN] sync_standbys set to %', standby;
 END $$ LANGUAGE plpgsql STRICT;
 CREATE FUNCTION set_sync_standbys_c(standby text) RETURNS void
 	AS 'pg_shardman' LANGUAGE C STRICT;

src/include/pg_shardman.h

@@ -12,10 +12,18 @@
 #include <signal.h>
 
 #include "miscadmin.h"
+#include "storage/lwlock.h"
 #include "libpq-fe.h"
 
-#define shmn_elog(level,fmt,...) elog(level, "[SHARDMAN] " fmt, ## __VA_ARGS__)
+/*
+ * By convention, we prefix shardman-related messages with [SHMN].
+ * Unfortunately, currently it is impossible to have variadic polymorhic
+ * functions accepting args with different types in plpgsql, so in sql funcs
+ * we just type it manually.
+ */
+#define shmn_elog(level,fmt,...) elog(level, "[SHMN] " fmt, ## __VA_ARGS__)
 
+/* Commonly used SPI stuff */
 #define SPI_PROLOG do { \
 	StartTransactionCommand(); \
 	SPI_connect(); \
@@ -63,7 +71,16 @@ extern char *shardman_shardlord_connstring;
 extern int shardman_cmd_retry_naptime;
 extern int shardman_poll_interval;
 
-extern int32 shardman_my_node_id;
+/*
+ * State in shared memory. We hold exclusive lock while modifying this, and
+ * shard lock while reading.
+ */
+typedef struct ShmnSharedState
+{
+	int32 my_id;
+	LWLock *lock;
+} ShmnSharedState;
+extern ShmnSharedState *snss;
 #define SHMN_INVALID_NODE_ID -1
 
 typedef struct Cmd
@@ -87,6 +104,9 @@ typedef struct RepCount
 } RepCount;
 
 extern void _PG_init(void);
+extern void _PG_fini(void);
+extern int32 my_id(void);
+extern void set_my_id(int32 new_id);
 extern void shardlord_main(Datum main_arg);
 extern bool signal_pending(void);
 extern void check_for_sigterm(void);

src/include/shardman_hooks.h

@@ -5,8 +5,12 @@
 #ifndef SHARDMAN_HOOKS_H
 #define SHARDMAN_HOOKS_H
 
-extern emit_log_hook_type log_hook_next;
+#include "storage/ipc.h"
 
-extern void shardman_log_hook(ErrorData *edata);
+extern emit_log_hook_type old_log_hook;
+extern shmem_startup_hook_type old_shmem_startup_hook;
+
+extern void shardman_log(ErrorData *edata);
+extern void shardman_shmem_startup(void);
 
 #endif							/* SHARDMAN_HOOKS_H */

src/pg_shardman.c

@@ -68,8 +68,10 @@ int shardman_poll_interval;
  * cleanup after receiving SIGTERM.
  */
 static PGconn *conn = NULL;
-/* This node id. */
 int32 shardman_my_node_id = -1;
+/* Link to shared memory state */
+ShmnSharedState *snss = NULL;
+
 
 /*
  * Entrypoint of the module. Define variables and register background worker.
@@ -86,9 +88,19 @@ _PG_init()
 						errhint("Add pg_shardman to shared_preload_libraries.")));
 	}
 
+	/*
+	 * Request additional shared resources.  (These are no-ops if we're not in
+	 * the postmaster process.)  We'll allocate or attach to the shared
+	 * resources in shardman_shmem_startup().
+	 */
+	RequestAddinShmemSpace(sizeof(ShmnSharedState));
+	RequestNamedLWLockTranche("pg_shardman", 1);
+
 	/* remember & set hooks */
-	log_hook_next = emit_log_hook;
-	emit_log_hook = shardman_log_hook;
+	old_log_hook = emit_log_hook;
+	emit_log_hook = shardman_log;
+	old_shmem_startup_hook = shmem_startup_hook;
+	shmem_startup_hook = shardman_shmem_startup;
 
 	DefineCustomBoolVariable("shardman.shardlord",
 							 "This node is the shardlord?",
@@ -165,6 +177,42 @@ _PG_init()
 	/* TODO: clean up publications if we were shardlord before */
 }
 
+/*
+ * Module unload callback
+ */
+void
+_PG_fini(void)
+{
+	/* Uninstall hooks. */
+	emit_log_hook = old_log_hook;
+	shmem_startup_hook = old_shmem_startup_hook;
+}
+
+/* Get this node id stored in shmem */
+int32
+my_id(void)
+{
+	int32 id;
+
+	/* If MyProc is NULL, shmem is not yet inited or we are bootstrapping */
+	if (MyProc == NULL)
+		return SHMN_INVALID_NODE_ID;
+
+	LWLockAcquire(snss->lock, LW_SHARED);
+	id = snss->my_id;
+	LWLockRelease(snss->lock);
+	return id;
+}
+
+/* Get this node id stored in shmem */
+void
+set_my_id(int32 new_id)
+{
+	LWLockAcquire(snss->lock, LW_EXCLUSIVE);
+	snss->my_id = new_id;
+	LWLockRelease(snss->lock);
+}
+
 /*
  * shardlord bgw starts here
  */
@@ -412,8 +460,8 @@ pg_shardman_installed_local(void)
 	if (get_extension_oid("pg_shardman", true) == InvalidOid)
 	{
 		installed = false;
-		shmn_elog(WARNING, "pg_shardman library is preloaded on shardlord, but"
-				  " extenstion is not created");
+		shmn_elog(WARNING,
+				  "Terminating shardlord: pg_shardman lib is preloaded, but ext is not created");
 	}
 	PopActiveSnapshot();
 	CommitTransactionCommand();

src/shard.c

@@ -232,7 +232,7 @@ rebalance(Cmd *cmd)
 	exec_tasks(tasks, num_parts);
 	SHMN_CHECK_FOR_INTERRUPTS_CMD(cmd);
 
-	shmn_elog(INFO, "Relation %s rebalanced:", relation);
+	shmn_elog(INFO, "Relation %s rebalanced", relation);
 	update_cmd_status(cmd->id, "done");
 }