Add show_subscription_status to show info about local subscriptions

PJMODOS · PJMODOS · commit ad05a52c5cc1 · 2015-12-06T12:05:02.000+01:00
diff --git a/README.md b/README.md
@@ -181,6 +181,14 @@ Nodes can be added and removed dynamically using the SQL interfaces.
   - `subscription_name` - name of the existing subscription
   - `relation` - name of existing table, optionally qualified
 
+- `pglogical.show_subscription_status(subscription_name name)`
+  Shows status and basic information about subscription.
+
+  Parameters:
+  - `subscription_name` - optional name of the existing subscription, when no
+    name was provided, the function will show status for all subscriptions on
+    local node
+
 - `pglogical.show_subscription_table(subscription_name name,
   relation regclass)`
   Shows synchronization status of a table.
diff --git a/expected/init.out b/expected/init.out
@@ -59,6 +59,12 @@ SELECT sync_kind, sync_subid, sync_nspname, sync_relname, sync_status FROM pglog
  f         | 3848008564 |              |              | r
 (1 row)
 
+SELECT * FROM pglogical.show_subscription_status();
+ subscription_name |   status    | provider_node |         provider_dsn         | replication_sets | forward_origins 
+-------------------+-------------+---------------+------------------------------+------------------+-----------------
+ test_subscription | replicating | test_provider | dbname=regression user=super | {default}        | 
+(1 row)
+
 -- Make sure we see the slot and active connection
 \c regression
 SELECT plugin, slot_type, database, active FROM pg_replication_slots;
diff --git a/pglogical--1.0.sql b/pglogical--1.0.sql
@@ -66,6 +66,12 @@ RETURNS boolean STRICT VOLATILE LANGUAGE c AS 'MODULE_PATHNAME', 'pglogical_alte
 CREATE FUNCTION pglogical.alter_subscription_remove_replication_set(subscription_name name, replication_set name)
 RETURNS boolean STRICT VOLATILE LANGUAGE c AS 'MODULE_PATHNAME', 'pglogical_alter_subscription_remove_replication_set';
 
+CREATE FUNCTION pglogical.show_subscription_status(subscription_name name DEFAULT NULL,
+	OUT subscription_name text, OUT status text, OUT provider_node text,
+	OUT provider_dsn text, OUT replication_sets text[],
+	OUT forward_origins text[])
+RETURNS SETOF record STABLE LANGUAGE c AS 'MODULE_PATHNAME', 'pglogical_show_subscription_status';
+
 CREATE TABLE pglogical.replication_set (
 	set_id oid NOT NULL PRIMARY KEY,
 	set_nodeid oid NOT NULL,
diff --git a/pglogical_functions.c b/pglogical_functions.c
@@ -80,6 +80,7 @@ PG_FUNCTION_INFO_V1(pglogical_alter_subscription_synchronize);
 PG_FUNCTION_INFO_V1(pglogical_alter_subscription_resynchronize_table);
 
 PG_FUNCTION_INFO_V1(pglogical_show_subscription_table);
+PG_FUNCTION_INFO_V1(pglogical_show_subscription_status);
 
 /* Replication set manipulation. */
 PG_FUNCTION_INFO_V1(pglogical_create_replication_set);
@@ -699,6 +700,119 @@ pglogical_show_subscription_table(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+/*
+ * Show info about subscribtion.
+ */
+Datum
+pglogical_show_subscription_status(PG_FUNCTION_ARGS)
+{
+	List			   *subscriptions;
+	ListCell		   *lc;
+	ReturnSetInfo	   *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+	TupleDesc			tupdesc;
+	Tuplestorestate	   *tupstore;
+	PGLogicalLocalNode *node;
+	MemoryContext		per_query_ctx;
+	MemoryContext		oldcontext;
+
+	/* check to see if caller supports us returning a tuplestore */
+	if (rsinfo == NULL || !IsA(rsinfo, ReturnSetInfo))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("set-valued function called in context that cannot accept a set")));
+	if (!(rsinfo->allowedModes & SFRM_Materialize))
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("materialize mode required, but it is not " \
+						"allowed in this context")));
+
+	node = get_local_node(true);
+	if (!node)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("current database is not configured as pglogical node"),
+				 errhint("create pglogical node first")));
+
+	if (PG_ARGISNULL(0))
+	{
+		subscriptions = get_node_subscriptions(node->node->id, false);
+	}
+	else
+	{
+		PGLogicalSubscription  *sub;
+		sub = get_subscription_by_name(NameStr(*PG_GETARG_NAME(0)), false);
+		subscriptions = list_make1(sub);
+	}
+
+	/* Switch into long-lived context to construct returned data structures */
+	per_query_ctx = rsinfo->econtext->ecxt_per_query_memory;
+	oldcontext = MemoryContextSwitchTo(per_query_ctx);
+
+	/* Build a tuple descriptor for our result type */
+	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+		elog(ERROR, "return type must be a row type");
+
+	tupstore = tuplestore_begin_heap(true, false, work_mem);
+	rsinfo->returnMode = SFRM_Materialize;
+	rsinfo->setResult = tupstore;
+	rsinfo->setDesc = tupdesc;
+
+	MemoryContextSwitchTo(oldcontext);
+
+	foreach (lc, subscriptions)
+	{
+		PGLogicalSubscription  *sub = lfirst(lc);
+		PGLogicalWorker		   *apply;
+		Datum	values[6];
+		bool	nulls[6];
+		char   *status;
+
+		memset(values, 0, sizeof(values));
+		memset(nulls, 0, sizeof(nulls));
+
+		LWLockAcquire(PGLogicalCtx->lock, LW_EXCLUSIVE);
+		apply = pglogical_apply_find(MyDatabaseId, sub->id);
+		if (pglogical_worker_running(apply))
+		{
+			PGLogicalSyncStatus	   *sync;
+			sync = get_subscription_sync_status(sub->id, true);
+
+			if (!sync)
+				status = "unknown";
+			else if (sync->status == SYNC_STATUS_READY)
+				status = "replicating";
+			else
+				status = "initializing";
+		}
+		else if (!sub->enabled)
+			status = "disabled";
+		else
+			status = "down";
+		LWLockRelease(PGLogicalCtx->lock);
+
+		values[0] = CStringGetTextDatum(sub->name);
+		values[1] = CStringGetTextDatum(status);
+		values[2] = CStringGetTextDatum(sub->origin->name);
+		values[3] = CStringGetTextDatum(sub->origin_if->dsn);
+		if (sub->replication_sets)
+			values[4] =
+				PointerGetDatum(strlist_to_textarray(sub->replication_sets));
+		else
+			nulls[4] = true;
+		if (sub->forward_origins)
+			values[5] =
+				PointerGetDatum(strlist_to_textarray(sub->forward_origins));
+		else
+			nulls[5] = true;
+
+		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+	}
+
+	tuplestore_donestoring(tupstore);
+
+	PG_RETURN_VOID();
+}
+
 /*
  * Create new replication set.
  */
diff --git a/pglogical_sync.c b/pglogical_sync.c
@@ -470,7 +470,7 @@ pglogical_sync_subscription(PGLogicalSubscription *sub)
 
 	StartTransactionCommand();
 	oldctx = MemoryContextSwitchTo(myctx);
-	sync = get_subscription_sync_status(sub->id);
+	sync = get_subscription_sync_status(sub->id, false);
 	MemoryContextSwitchTo(oldctx);
 	CommitTransactionCommand();
 
@@ -631,7 +631,7 @@ pglogical_sync_table(PGLogicalSubscription *sub, RangeVar *table)
 	StartTransactionCommand();
 
 	/* Sanity check. */
-	sync = get_subscription_sync_status(sub->id);
+	sync = get_subscription_sync_status(sub->id, false);
 	if (sync->status != SYNC_STATUS_READY)
 	{
 		elog(ERROR,
@@ -925,7 +925,7 @@ syncstatus_fromtuple(HeapTuple tuple, TupleDesc desc)
 
 /* Get the sync status for a subscription. */
 PGLogicalSyncStatus *
-get_subscription_sync_status(Oid subid)
+get_subscription_sync_status(Oid subid, bool missing_ok)
 {
 	PGLogicalSyncStatus	   *sync;
 	RangeVar	   *rv;
@@ -953,7 +953,16 @@ get_subscription_sync_status(Oid subid)
 	}
 
 	if (!HeapTupleIsValid(tuple))
+	{
+		if (missing_ok)
+		{
+			systable_endscan(scan);
+			heap_close(rel, RowExclusiveLock);
+			return NULL;
+		}
+
 		elog(ERROR, "subscription %u status not found", subid);
+	}
 
 	sync = syncstatus_fromtuple(tuple, tupDesc);
 
diff --git a/pglogical_sync.h b/pglogical_sync.h
@@ -55,7 +55,8 @@ extern void pglogical_sync_table(PGLogicalSubscription *sub, RangeVar *table);
 extern void create_local_sync_status(PGLogicalSyncStatus *sync);
 extern void drop_subscription_sync_status(Oid subid);
 
-extern PGLogicalSyncStatus *get_subscription_sync_status(Oid subid);
+extern PGLogicalSyncStatus *get_subscription_sync_status(Oid subid,
+														 bool missing_ok);
 extern void set_subscription_sync_status(Oid subid, char status);
 
 extern void drop_table_sync_status(const char *nspname, const char *relname);
diff --git a/sql/init.sql b/sql/init.sql
@@ -52,6 +52,8 @@ END;$$;
 
 SELECT sync_kind, sync_subid, sync_nspname, sync_relname, sync_status FROM pglogical.local_sync_status ORDER BY 2,3,4;
 
+SELECT * FROM pglogical.show_subscription_status();
+
 -- Make sure we see the slot and active connection
 \c regression
 SELECT plugin, slot_type, database, active FROM pg_replication_slots;
