pg-stream-gcp

A PostgreSQL transaction log handler for the pg-stream-core library to enable logical replication streaming output to Google Cloud Platform (GCP) PubSub.

The pg-stream CLI utility is the simplest way to use the functionality in pg-stream-gcp without writing any code.

Usage

import * as pg from '@jmorecroft67/pg-stream-core';
import * as gcp from '@jmorecroft67/pg-stream-gcp';
import { pipe } from 'fp-ts/function';
import * as RTE from 'fp-ts/ReaderTaskEither';
import * as TE from 'fp-ts/TaskEither';
import { createLogger } from 'bunyan';

export const logger = createLogger({
  name: 'test',
  level: 'debug'
});

const handler = gcp.pubSubHandler({
  projectId: 'gcp-project-id',
  topicName: 'gcp-topic-name',
  noOrdering: false, // preserve message ordering per table
  includeAll: false, // only send table-related messages
  retryOptions: {
    // retry on failures
    delay: 1000, // 1 second initial wait
    exponentialBackoff: {
      // exponentially increase wait time
      capDelay: 30000 // max 30 second wait
    },
    maxRetries: 5 // abort after 5 retries
  }
});

const options: pg.RecvlogicalOptions = {
  host: 'localhost',
  port: 5432,
  database: 'postgres',
  username: 'postgres',
  password: 'topsecret',
  publicationName: 'testpub',
  slotName: 'testslot',
  xlogs: {
    handler,
    concurrency: 5
  },
  createTempSlotIfNone: true,
  createPublicationIfNone: true
};

TE.bracket(
  pg.open({ options, logger }),
  (conn) =>
    pipe(
      { conn, logger },
      pipe(
        pg.start(options, true),
        RTE.chain(() => pg.recvlogical(options))
      )
    ),
  (conn) => pg.close({ conn, logger })
)();

We can demonstrate this program by running it, then manipulating some data in our database, which we can do by running the test query program in the pg-stream-core README.

$ $(npm bin)/ts-node src/test-gcp.ts | bunyan &
$ $(npm bin)/ts-node src/test-query.ts | bunyan

The output of the first program is then:

[2022-05-23T01:23:50.149Z]  WARN: test/8825 on Jesss-MBP: SSL is not enabled on the Postgres server but connecting anyway.
[2022-05-23T01:23:50.149Z]  INFO: test/8825 on Jesss-MBP: Connection established. (host=localhost, port=5432)
[2022-05-23T01:23:50.184Z] DEBUG: test/8825 on Jesss-MBP: Startup sequence completed.
[2022-05-23T01:23:50.185Z] DEBUG: test/8825 on Jesss-MBP: Sending SQL query.
    sql: SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = 'testslot'
[2022-05-23T01:23:50.188Z]  INFO: test/8825 on Jesss-MBP: Command complete. (tag="SELECT 0")
[2022-05-23T01:23:50.188Z]  INFO: test/8825 on Jesss-MBP: Query complete. (result=[])
[2022-05-23T01:23:50.189Z] DEBUG: test/8825 on Jesss-MBP: Sending SQL query.
    sql: SELECT * FROM pg_publication WHERE pubname = 'testpub'
[2022-05-23T01:23:50.191Z]  INFO: test/8825 on Jesss-MBP: Command complete. (tag="SELECT 1")
[2022-05-23T01:23:50.191Z]  INFO: test/8825 on Jesss-MBP: Query complete.
    result: [{"oid":"16582","pubname":"testpub","pubowner":"10","puballtables":true,"pubinsert":true,"pubupdate":true,"pubdelete":true,"pubtruncate":true,"pubviaroot":false}]
[2022-05-23T01:23:50.191Z] DEBUG: test/8825 on Jesss-MBP: Sending SQL query.
    sql: CREATE_REPLICATION_SLOT testslot TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT
[2022-05-23T01:23:50.197Z]  INFO: test/8825 on Jesss-MBP: Command complete. (tag=CREATE_REPLICATION_SLOT)
[2022-05-23T01:23:50.197Z]  INFO: test/8825 on Jesss-MBP: Query complete.
    result: [{"slot_name":"testslot","consistent_point":"0/1E4C450","snapshot_name":null,"output_plugin":"pgoutput"}]
[2022-05-23T01:24:14.128Z] DEBUG: test/8825 on Jesss-MBP: Sent message(s) to topic. (projectId=gcp-project-id, topicName=gcp-topic-name)
    acks: [
      "1"
    ]
[2022-05-23T01:24:14.128Z] DEBUG: test/8825 on Jesss-MBP: Sent message(s) to topic. (projectId=gcp-project-id, topicName=gcp-topic-name)
    acks: [
      "2"
    ]
[2022-05-23T01:24:14.128Z] DEBUG: test/8825 on Jesss-MBP: Sent message(s) to topic. (projectId=gcp-project-id, topicName=gcp-topic-name)
    acks: [
      "3"
    ]
[2022-05-23T01:24:14.128Z] DEBUG: test/8825 on Jesss-MBP: Sent message(s) to topic. (projectId=gcp-project-id, topicName=gcp-topic-name)
    acks: [
      "4"
    ]
[2022-05-23T01:24:14.128Z] DEBUG: test/8825 on Jesss-MBP: Sent message(s) to topic. (projectId=gcp-project-id, topicName=gcp-topic-name)
    acks: [
      "5"
    ]

kill -SIGINT 8825

[2022-05-23T01:24:26.145Z]  INFO: test/8825 on Jesss-MBP: Logical streaming complete.
[2022-05-23T01:24:26.146Z]  INFO: test/8825 on Jesss-MBP: Connection closed.

This is running against the GCP PubSub emulator, which is returning the acknowledgement IDs of "1", "2", etc. We can see from these logs that the five messages were sent in entirely separate batches, which makes sense since the number of database events (5) matched our concurrency setting.