Physical log based replication

This is a replication scheme that relies on physical logs generated by Berkeley DB to replicate changes to other nodes asynchronously. The replicant nodes (a.k.a. physreps) pull the physical logs from the source machine and apply it locally.

It is important to note that physreps are applying the logs out-of-band and, thus, should not be considered part of the source cluster.

In order to enable replication, replicate_from must be added to the physrep’s lrl file. This line can either take a valid Comdb2 cluster tier, a hostname or a comma-separated list of hostnames (without any space).

replicate_from source_dbname @source_host/source_tier

Physical replicants started with the above configuration will start replicating directly off of the specified source host or one of the nodes in the cluster tier.

Setup tiered replication topology

Physical replicants pull logs from the source by executing a SELECT query. Thus any physrep added to the system would add to the overall cost incurred by the source host/cluster.

In order to avoid having source support all physreps directly, one could setup tiered replication, in which some physical replicants could become the source for other physical replicants, thus keeping some of the load off of the top-level source host/cluster. A physical replicant’s ‘tier’ is it’s distance from the source cluster.


                            S       |                    -- tier 0 
                           / \      |
                          /   \     | log-flow
                         R1   R2    |                    -- tier 1
                        / \     \   |
                       /   \     \  v
                      R3   R4    R5                      -- tier 2

Note: The source cluster nodes are always considered at tier 0.

Setting up tiered replication topology requires 2 base tables to maintain the current state of replication as well as the replication topology.

These tables automatically get updated to reflect the changes as replicants join or leave the system and thus are not designed to be manually modified under normal circumstances. In order to keep the load evenly spread, these table are consulted to ensure a certain fanout physrep_fanout is maintained across all the nodes. The LSN file and offset in the comdb2_physreps table is used by all the nodes to pause log-deletion.

Algorithm

On start, a physical replicant executes sys.physrep.register_replicant() against the physrep_metadb, which in turn, responds with a list of potential nodes which can be used as the source of physical logs. The metadb creates this list by doing a graph traversal of nodes (listed in comdb2_physreps) connected by edges (listed in comdb2_physrep_connections) starting at the source db as the root-node/tier 0. The physical replicant then chooses a node from this list and tries to connect to it. On successful connection, the replicant executes sys.physrep.update_registry() against the physrep_metadb, confirming that the replicant is now connected to a node.

Upon successful registration, the physical replicants execute SELECT .. FROM .. comdb2_transaction_logs to retrieve (PULL) log records, which is then applied locally. The replicants also periodically execute sys.physrep.keepalive() against physrep_metadb to inform about their current LSN. This information is used by the nodes to control log-deletion.

Cross-tier replication

In certain setups, where a TCP connection is not permitted from a lower replication tier (say development) to a higher replication tier (say production), the PULL strategy disussed above will not work.

So, in order to enable physical replicants, running in development tier, replicate from source nodes running in production tier, we force them to wait for a connection from a node in production tier, and when a connection arrives, the replicants use that to execute the (reverse) query to read logs from comdb2_transaction_logs.

                      ,------- S <-------,
                      |(1) connect       |
                      |                  |                         (production)
----------------------|------------------|-------------------------------------
                      |                  |                        (development)
                      |                  |(2) execute `SELECT .. FROM comdb2_transaction_logs`
                      `------> R1 -------'

Now, the question is, how would the nodes in production know which node(s) in development to connect to? They get this information from physrep_metadb's comdb2_physrep_sources table. This table can be updated with all the source->target information, based on which a replicant (in development) waits for a connection (sys.physrep.should_wait_for_con()) and the source nodes (in production) can connect to the physrep nodes by getting a list from the same table (sys.physrep.should_wait_for_con()). The comdb2_physrep_sources table allows the user to specify the source and target db’s by either hostname, tier, or cluster. A database will attempt to establish a reverse connection with a target if it’s source_dbname matches the source_dbname, and the source_host field matches the sources hostname, cluster-name, or tier. When a source sees that a target is specified by tier, it will attempt to establish a reverse-connection to all of the machines listed under comdb2db for that tier. If a target is listed by cluster-name, the source will attempt to establish a reverse-connection to all of the hosts listed in that cluster, as represented by the bbcpu.lst.

NOTE: In cross-tier replication, the replication metadata tables must be hosted by a separate database running in the a lower (development) tier.

Alternate Metadbs

Physrep setup supports configuring multiple alternate metadbs in addition to its primary physrep_metadb. The intention is to allow prod databases to connect only to prod metadbs. A beta physical replicant which replicates from a prod instance can serve as an intermediary source for lower tiered physreps by listing an alternate-metadb which is accessible by the lower tiered physreps. Lower tiered physreps would then replicate from the beta instance using this alternate metadb as their primary metadb.

Key takeaways:

  • A physical replicant registers (register_replicant) only against it’s primary metadb.
  • A source database will periodically update its physrep-metadb and all alternate metadbs with its current LSN and first-logfile via the ‘physrep_keepalive’ function.
  • A source database uses the ‘comdb2_physrep_sources’ table in its primary metadb and all alternate metadbs to determine whether it should initiate a reverse-connection to a child database.

Physical replication metadata tables

comdb2_physreps

CREATE TABLE comdb2_physreps(dbname CSTRING(60),
                             host CSTRING(120),
                             file INT,
                             offset INT,
                             firstfile INT,
                             last_keepalive DATETIME,
                             state CSTRING(60),
                             UNIQUE (dbname, host))

This could be thought of as the registry of all the physical replicants nodes in the system (both sources and replicants).

The firstfile column records each node’s oldest available transaction log file (populated by keepalive_v2), while file records its newest. Together they let register_replicant prefer sources whose [firstfile, file] range actually covers a registering replicant’s LSN. The firstfile column is optional: if it is absent, keepalive_v2 simply does not report it and register_replicant skips range-filtering — replicants still confirm coverage at connect time via physrep_verify_source_range. Adding the column is a pure optimization (fewer wasted connection attempts), not a requirement.

Recommended rollout ordering. physrep_verify_source_range ships on (the connect-time coverage check has no compatibility concerns and is the primary fix, active as soon as the binary is deployed). physrep_keepalive_v2 ships off so a new binary is a behavior-neutral drop-in for the registry path:

  1. Deploy the new binary across the fleet (sources, replicants, metadbs). Connect-time source verification is active immediately.
  2. ALTER TABLE comdb2_physreps ADD COLUMN firstfile INT on the metadb cluster(s).
  3. Enable physrep_keepalive_v2 fleet-wide so nodes begin reporting firstfile; register_replicant then also range-filters candidates during registration.

Enabling physrep_keepalive_v2 before step 1 completes or step 2 is done risks a node invoking the v2 keepalive SP on an older metadb that cannot handle a missing firstfile column.

  • Physical replicant states:
    • Pending : The node has requested to become a physical replicant (registration in-progress)
    • Active : The node is a physical replicant
    • Inactive : The node might not be actively replicating; not allowed to become a replication source

comdb2_physrep_connections

CREATE TABLE comdb2_physrep_connections(dbname CSTRING(60),
                                        host CSTRING(120),
                                        source_dbname CSTRING(60),
                                        source_host CSTRING(120),
                                        UNIQUE (dbname, host, source_dbname, source_host), 
                                        FOREIGN KEY(dbname, host) REFERENCES comdb2_physreps(dbname, host) ON DELETE CASCADE)

comdb2_physrep_sources

CREATE TABLE comdb2_physrep_sources(dbname CSTRING(60),
                                    host CSTRING(120),
                                    source_dbname CSTRING(60),
                                    source_host CSTRING(120),
                                    UNIQUE (dbname, host, source_dbname, source_host))

Tunables

  • blocking_physrep: The SELECT .. FROM comdb2_transaction_logs query executed by physical replicants blocks for the next log record. (Default: false)
  • physrep_debug: Print extended physrep trace. (Default: off)
  • physrep_exit_on_invalid_logstream: Exit physreps on invalid logstream. (Default: off)
  • physrep_fanout: Maximum number of physical replicants that a node can service (Default: 8)
  • physrep_hung_replicant_check_freq_sec: Check for hung physical replicant this often. (Default: 10)
  • physrep_hung_replicant_threshold: Report if the physical replicant has been inactive for this duration. (Default: 60)
  • physrep_i_am_metadb: Marks the node as ‘physical replcation metadb’. (Default: off)
  • physrep_keepalive_freq_sec: Periodically send lsn to source node after this interval. (Default: 10)
  • physrep_keepalive_v2: Use version 2 of keepalive, which also reports the oldest (first) log file. Ships off for safe gradual rollout; enable fleet-wide only after the metadb comdb2_physreps table has a firstfile column (older metadb binaries cannot tolerate its absence). (Default: off)
  • physrep_max_candidates: Maximum number of candidates that should be returned to a new physical replicant during registration. (Default: 6)
  • physrep_no_source_alarm_threshold: Consecutive no-viable-source registration cycles (a cycle in which every reachable candidate was probed and none retained logs covering this replicant’s LSN) before a physrep latches gbl_physrep_no_viable_source and logs PHYSREP NO VIABLE SOURCE. The latched state is exposed as the physrep_no_viable_source metric in comdb2_metrics and is cleared once the replicant connects to a viable source. (Default: 10)
  • physrep_metadb_host: List of physical replication metadb cluster hosts.
  • physrep_metadb_name: Physical replication metadb cluster name.
  • physrep_reconnect_penalty: Physrep wait seconds before retry to the same node. (Default: 5)
  • physrep_register_interval: Interval for physical replicant re-registration. (Default: 3600)
  • physrep_shuffle_host_list: Shuffle the host list returned by register_replicant() before connecting to the hosts. (Default: off)
  • physrep_source_dbname: Physical replication source cluster dbname.
  • physrep_source_host: List of physical replication source cluster hosts.
  • physrep_verify_source_range: Before committing to a candidate source, verify (by querying the source directly) that its live log range covers the replicant’s LSN; skip it if not. Schema-independent, needs neither the firstfile column nor a uniform fleet, and has no compatibility concerns (purely local, fails safe, and uses a query older sources already support). (Default: on)
  • revsql_allow_command_execution : Allow processing and execution of command * over the reverse connection that has come in as part of the request. This is mostly intended for testing. (Default: off)
  • revsql_cdb2_debug: Print extended reversql-sql cdb2 related trace. (Default: off)
  • revsql_connect_freq_sec: This node will attempt to reverse connect to the remote host at this frequency. (Default: 5 secs)
  • revsql_debug: Print extended reversql-sql trace. (Default: off)
  • revsql_host_refresh_freq_sec: The frequency at which the reverse connection host list will be refreshed. (Default: 60 secs)
  • revsql_force_rte: Force rte-mode for all reverse connections. (Default: on)
  • connect_remote_rte: Force rte-mode for both fdb and revsql connections. (Default: off)

lrl configurations

  • replicate_from dbname @host/tier: This line sets the source host/cluster. It is required for all physical replicants.
  • replicate_wait : Tells the physical replicant to wait for this many seconds before applying the log records.
  • physrep_metadb: If set, all the nodes will connect to this database (as against source host/cluster mentioned via replicate_from) for replication metadata tables
  • alternate_metadb : If set, parent node will use this in addition to its normal metadb in updating stats, and determining whether it should establish a reverse-connection.
  • physrep_fanout_override : This is set on the metadb, and allows per-database overrides of the 'physrep_fanout' tunable. The 'physrep_fanout_override' message-trap allows this to be set dynamically. The 'physrep_fanout_dump' message-trap prints the current overrides.
  • physrep_ignore : All the log records that belong to any of these tables are ignored by physical replicants
  • nonames: This configuration forces system database file names to not carry the database name. This setting is required for physical-log based replication to work properly.
  • elect_highest_committed_gen: Bias election by the highest generation in the logfile. This setting is required for physical-log based replication to work properly.
  • autoanalyze: Since physical replicants run in read-only mode, autoanalyze would not have any effect on the state of ‘stat’ tables. Thus, it is recommended to keep it turned off.

Stored procedures

  • sys.cmd.start_replication: Start replication threads
  • sys.cmd.stop_replication: Stop replication threads

Note: Based on the current lrls (and existence of replication matadata tables) the replication threads are created automatically on start.

There are additional stored procedures that nodes participating in the physical replication execute to perform various tasks from setting up to maintining the state of physical replication. These procedures should be considered internal and not be executed manually.

  • sys.physrep.register_replicant: A request to register a new physical replicant. This is the first step of registration.
  • sys.physrep.update_registry: As the second step of registration, the physrep executes this to confirm that it is successfully connected to a node.
  • sys.physrep.keepalive: Is executed by replicants periodically to report their current LSNs.
  • sys.physrep.keepalive_v2: Like keepalive, but also reports the oldest (first) log file into comdb2_physreps.firstfile when that column exists. Tolerates its absence.
  • sys.physrep.reset_nodes: Is executed to reset states of nodes in replication metadb.
  • sys.physrep.topology: Returns a list of all the nodes in the system alongside their tiers.
  • sys.physrep.get_reverse_hosts: Returns pairs of dbname/host against which the source node must periodically send reversesql requests.
  • sys.physrep.should_wait_for_con: This is used by physreps to check whether they should wait for a ‘reversesql’ connection from a source host.

Log-triggers / Log-queue-triggers

  • log-triggers provide a way for a physical replicant to redirect writes to queues to either a file, or to a Kafka topic. It provides a solution for products which previously utilized consumers under local-replication. As consuming directly from a queue requires the ability to write to the database (which is not possible for physical replicants) log-triggers are being developed as an alternate notification mechanism. As log-triggers are under active development, we expect it to change as we receive feedback from customers.

The log-trigger sub-system introduces two new lrl-directives:

  • qdump <queue-name> <file-name> <maxsize>: All write-traffic directed to the specified ‘queue-name’ will be written to ‘file-name’. The maxsize parameter places a limit on the amount of memory we allow to remain queued in memory. Specifying 0 for maxsize allows this to grow without bound.
  • qkafka <queue-name> <kafka-topic> <maxsize>: All write-traffic directed to the specified ‘queue-name’ will be written to the specified ‘kafka-topic’. As with the qdump parameter, the maxsize argument places a limit on the amount of memory we allow in the outgoing queue, with 0 allowing it to grow without bound. In a clustered-phyrep deployment, qkafka will only write to kafka from the current master node.