Use new replication gem

Gemfile

@@ -50,7 +50,7 @@ gem "log_decorator",                  "~>0.1",         :require => false
 gem "manageiq-api-client",            "~>0.3.3",       :require => false
 gem "manageiq-messaging",             "~>0.1.4",       :require => false
 gem "manageiq-password",              "~>0.3",         :require => false
-gem "manageiq-postgres_ha_admin",     "~>3.0",         :require => false
+gem "manageiq-postgres_ha_admin",     "~>3.1",         :require => false
 gem "memoist",                        "~>0.15.0",      :require => false
 gem "mime-types",                     "~>3.0",         :path => File.expand_path("mime-types-redirector", __dir__)
 gem "more_core_extensions",           "~>3.7"
@@ -162,7 +162,7 @@ group :automate, :seed, :manageiq_default do
 end
 
 group :replication, :manageiq_default do
-  gem "pg-pglogical",                   "~>2.1.2",       :require => false
+  gem "pg-logical_replication", "~>1.0", :require => false
 end
 
 group :rest_api, :manageiq_default do

app/models/miq_region.rb

@@ -150,10 +150,6 @@ def self.global_replication_type?
     MiqPglogical.new.subscriber?
   end
 
-  def self.replication_enabled?
-    MiqPglogical.new.node?
-  end
-
   def self.replication_type
     if global_replication_type?
       :global

app/models/pglogical_subscription.rb

@@ -1,9 +1,5 @@
-# This model wraps a pglogical stored proc (pglogical.show_subscription_status)
-# This is exposed to us through the PostgreSQLAdapter#pglogical object's #subscriptions method
-# This model then exposes select values returned from that method
 require 'pg/dsn_parser'
-require 'pg/pglogical'
-require 'pg/pglogical/active_record_extension'
+require 'pg/logical_replication'
 
 class PglogicalSubscription < ActsAsArModel
   set_columns_hash(
@@ -65,12 +61,8 @@ def self.save_all!(subscription_list)
   end
 
   def delete(reload_failover_monitor_service = true)
-    pglogical.subscription_drop(id, true)
+    safe_delete
     MiqRegion.destroy_region(connection, provider_region)
-    if self.class.count == 0
-      pglogical.node_drop(MiqPglogical.local_node_name, true)
-      pglogical.disable
-    end
     EvmDatabase.restart_failover_monitor_service_queue if reload_failover_monitor_service
   end
 
@@ -81,16 +73,16 @@ def self.delete_all(list = nil)
   end
 
   def disable
-    pglogical.subscription_disable(id).check
+    pglogical.disable_subscription(id).check
   end
 
   def enable
-    pglogical.subscription_enable(id).check
+    pglogical.enable_subscription(id).check
   end
 
   def self.pglogical(refresh = false)
     @pglogical = nil if refresh
-    @pglogical ||= connection.pglogical
+    @pglogical ||= PG::LogicalReplication::Client.new(connection.raw_connection)
   end
 
   def pglogical(refresh = false)
@@ -108,32 +100,52 @@ def validate(new_connection_params = {})
   end
 
   def backlog
-    connection.xlog_location_diff(remote_node_lsn, remote_replication_lsn)
+    connection.xlog_location_diff(remote_region_lsn, subscription_status["remote_replication_lsn"])
   rescue PG::Error => e
     _log.error(e.message)
     nil
   end
 
+  def sync_tables
+    pglogical.sync_subscription(id)
+  end
+
   # translate the output from the pglogical stored proc to our object columns
   def self.subscription_to_columns(sub)
     cols = sub.symbolize_keys
 
     # delete the things we dont care about
+    cols.delete(:database_name)
+    cols.delete(:owner)
     cols.delete(:slot_name)
-    cols.delete(:replication_sets)
-    cols.delete(:forward_origins)
+    cols.delete(:publications)
     cols.delete(:remote_replication_lsn)
     cols.delete(:local_replication_lsn)
 
-    cols[:id] = cols.delete(:subscription_name)
+    cols[:id]     = cols.delete(:subscription_name)
+    cols[:status] = subscription_status(cols.delete(:worker_count), cols.delete(:enabled))
 
     # create the individual dsn columns
-    cols.merge!(dsn_attributes(cols.delete(:provider_dsn)))
+    cols.merge!(dsn_attributes(cols.delete(:subscription_dsn)))
 
-    cols.merge!(provider_node_attributes(cols.delete(:provider_node)))
+    cols.merge!(remote_region_attributes(cols[:id]))
   end
   private_class_method :subscription_to_columns
 
+  def self.subscription_status(workers, enabled)
+    return "disabled" unless enabled
+
+    case workers
+    when 0
+      "down"
+    when 1
+      "replicating"
+    else
+      "initializing"
+    end
+  end
+  private_class_method :subscription_status
+
   def self.dsn_attributes(dsn)
     attrs = PG::DSNParser.parse(dsn)
     attrs.select! { |k, _v| [:dbname, :host, :user, :port].include?(k) }
@@ -143,17 +155,17 @@ def self.dsn_attributes(dsn)
   end
   private_class_method :dsn_attributes
 
-  def self.provider_node_attributes(node_name)
+  def self.remote_region_attributes(subscription_name)
     attrs = {}
-    attrs[:provider_region] = MiqPglogical.node_name_to_region(node_name)
+    attrs[:provider_region] = subscription_name.split("_")[1].to_i
     region = MiqRegion.find_by_region(attrs[:provider_region])
     attrs[:provider_region_name] = region.description if region
     attrs
   end
-  private_class_method :provider_node_attributes
+  private_class_method :remote_region_attributes
 
   def self.subscriptions
-    pglogical.enabled? ? pglogical.subscriptions : []
+    pglogical.subscriptions(connection.current_database)
   end
   private_class_method :subscriptions
 
@@ -178,6 +190,15 @@ def self.find_id(to_find)
 
   private
 
+  def safe_delete
+    pglogical.drop_subscription(id, true)
+  rescue PG::InternalError => e
+    raise unless e.message =~ /could not connect to publisher/ || e.message =~ /replication slot .* does not exist/
+    disable
+    pglogical.alter_subscription_options(id, "slot_name" => "NONE")
+    pglogical.drop_subscription(id, true)
+  end
+
   def remote_region_number
     with_remote_connection do |_conn|
       return MiqRegionRemote.region_number_from_sequence
@@ -188,43 +209,23 @@ def new_subscription_name
     "region_#{remote_region_number}_subscription"
   end
 
-  def ensure_node_created
-    return if MiqPglogical.new.node?
-
-    pglogical.enable
-    node_dsn = PG::Connection.parse_connect_args(connection.raw_connection.conninfo_hash.delete_blanks)
-    pglogical.node_create(MiqPglogical.local_node_name, node_dsn).check
-  end
-
-  def with_subscription_disabled
-    disable
-    yield
-  ensure
-    enable
-  end
-
   def update_subscription
-    with_subscription_disabled do
-      provider_node_name = MiqPglogical.region_to_node_name(provider_region)
-      find_password if password.nil?
-      pglogical.node_dsn_update(provider_node_name, dsn)
-    end
+    find_password if password.nil?
+    pglogical.set_subscription_conninfo(id, conn_info_hash)
     self
   end
 
   # sets this instance's password field to the one in the subscription dsn in the database
   def find_password
     return password if password.present?
-    s = pglogical.subscription_show_status(id).symbolize_keys
-    dsn_hash = PG::DSNParser.parse(s.delete(:provider_dsn))
+    s = subscription_status.symbolize_keys
+    dsn_hash = PG::DSNParser.parse(s.delete(:subscription_dsn))
     self.password = dsn_hash[:password]
   end
 
   def create_subscription
-    ensure_node_created
     MiqRegion.destroy_region(connection, remote_region_number)
-    pglogical.subscription_create(new_subscription_name, dsn, [MiqPglogical::REPLICATION_SET_NAME],
-                                  false).check
+    pglogical.create_subscription(new_subscription_name, conn_info_hash, [MiqPglogical::PUBLICATION_NAME]).check
     self
   end
 
@@ -234,26 +235,21 @@ def assert_different_region!
     end
   end
 
-  def dsn
-    conf = {
+  def conn_info_hash
+    {
       :dbname   => dbname,
       :host     => host,
       :user     => user,
       :password => decrypted_password,
       :port     => port
     }.delete_blanks
-    PG::Connection.parse_connect_args(conf)
   end
 
   def decrypted_password
     ManageIQ::Password.try_decrypt(password)
   end
 
-  def remote_replication_lsn
-    pglogical.subscription_show_status(id)["remote_replication_lsn"]
-  end
-
-  def remote_node_lsn
+  def remote_region_lsn
     with_remote_connection(&:xlog_location)
   end
 
@@ -263,4 +259,8 @@ def with_remote_connection
       yield conn
     end
   end
+
+  def subscription_status
+    pglogical.subscriptions.find { |s| s["subscription_name"] == id }
+  end
 end

config/default_replication_exclude_tables.yml → config/replication_exclude_tables.yml

@@ -85,7 +85,6 @@
 - server_roles
 - sessions
 - vim_performance_states
-- vim_performance_tag_values
 - vmdb_database_metrics
 - vmdb_databases
 - vmdb_indexes

lib/evm_database.rb

@@ -147,7 +147,7 @@ def self.run_failover_monitor(monitor = nil)
     monitor ||= ManageIQ::PostgresHaAdmin::FailoverMonitor.new(Rails.root.join("config", "ha_admin.yml"))
 
     configure_rails_handler(monitor)
-    configure_pglogical_handlers(monitor)
+    configure_logical_replication_handlers(monitor)
 
     _log.info("Starting database failover monitor")
     monitor.monitor_loop
@@ -177,12 +177,12 @@ def self.configure_rails_handler(monitor)
   end
   private_class_method :configure_rails_handler
 
-  def self.configure_pglogical_handlers(monitor)
+  def self.configure_logical_replication_handlers(monitor)
     return unless MiqServer.my_server.has_active_role?("database_operations")
 
     local_db_conninfo = ActiveRecord::Base.connection.raw_connection.conninfo_hash.delete_blanks
     PglogicalSubscription.all.each do |s|
-      handler = ManageIQ::PostgresHaAdmin::PglogicalConfigHandler.new(:subscription => s.id, :conn_info => local_db_conninfo)
+      handler = ManageIQ::PostgresHaAdmin::LogicalReplicationConfigHandler.new(:subscription => s.id, :conn_info => local_db_conninfo)
       _log.info("Configuring database failover for replication subscription #{s.id} ")
 
       handler.after_failover do |new_conn_info|
@@ -193,5 +193,5 @@ def self.configure_pglogical_handlers(monitor)
       monitor.add_handler(handler)
     end
   end
-  private_class_method :configure_pglogical_handlers
+  private_class_method :configure_logical_replication_handlers
 end

lib/evm_database_ops.rb

@@ -186,13 +186,8 @@ def self.restore(db_opts, connect_opts = {})
     end
 
     MiqRegion.replication_type = :none
-    60.times do
-      break if VmdbDatabaseConnection.where("application_name LIKE 'pglogical manager%'").count.zero?
-      _log.info("Waiting for pglogical connections to close...")
-      sleep 5
-    end
 
-    connection_count = backup_type == :basebackup ? VmdbDatabaseConnection.unscoped.count : VmdbDatabaseConnection.count
+    connection_count = backup_type == :basebackup ? VmdbDatabaseConnection.unscoped.where(:backend_type => "client backend").count : VmdbDatabaseConnection.count
     if connection_count > 1
       message = "Database restore failed. #{connection_count - 1} connections remain to the database."
       _log.error(message)

lib/extensions/ar_migration.rb

@@ -69,9 +69,11 @@ def wait_message
     end
 
     def restart_subscription
-      c = SubscriptionHelper.establish_connection.connection
-      c.pglogical.subscription_disable(subscription.id)
-      c.pglogical.subscription_enable(subscription.id)
+      c = SubscriptionHelper.establish_connection.connection.raw_connection
+      rep_client = PG::LogicalReplication::Client.new(c)
+
+      rep_client.disable_subscription(subscription.id)
+      rep_client.enable_subscription(subscription.id)
     ensure
       SubscriptionHelper.remove_connection
     end