Merge pull request #21372 from jrafanie/gracefully_handle_logical_rep…

…lication_with_non_superuser_pg_role

Gracefully handle non-superuser setting up replication

(cherry picked from commit 198317d)

app/models/miq_region.rb

@@ -144,24 +144,13 @@ def self.global_replication_type?
   end
 
   def self.replication_type
-    if remote_replication_type?
-      :remote
-    elsif global_replication_type?
-      :global
-    else
-      :none
-    end
+    MiqPglogical.new.replication_type
   end
 
   def self.replication_type=(desired_type)
-    current_type = replication_type
-    return desired_type if desired_type == current_type
-
-    MiqPglogical.new.destroy_provider   if current_type == :remote
-    PglogicalSubscription.delete_all    if current_type == :global
-    MiqPglogical.new.configure_provider if desired_type == :remote
-    # Do nothing to add a global
-    desired_type
+    return desired_type if desired_type == replication_type
+
+    MiqPglogical.new.replication_type = desired_type
   end
 
   def ems_clouds

app/models/pglogical_subscription.rb

@@ -169,7 +169,7 @@ def self.remote_region_attributes(subscription_name)
   def self.subscriptions
     with_connection_error_handling do
       pglogical.subscriptions(connection.current_database)
-    end
+    end || []
   end
   private_class_method :subscriptions
 
@@ -241,6 +241,9 @@ def find_password
   end
 
   def create_subscription
+    # Don't even start into this method if logical replication is not supported (not superuser)
+    return unless logical_replication_supported?
+
     MiqRegion.destroy_region(connection, remote_region_number)
 
     # new_subscription_name also queries the remote, so we fetch it early to avoid a nested remote query
@@ -302,6 +305,6 @@ def with_remote_pglogical_client(connect_timeout = 0)
   def subscription_attributes
     self.class.with_connection_error_handling do
       pglogical.subscriptions.find { |s| s["subscription_name"] == id }
-    end
+    end || {}
   end
 end

lib/extensions/ar_migration.rb

@@ -95,6 +95,8 @@ def restart_subscription
 
 module ArPglogicalMigration
   def migrate(direction)
+    return super unless PglogicalSubscription.logical_replication_supported?
+
     ArPglogicalMigrationHelper.subscriptions.each do |s|
       ArPglogicalMigrationHelper::RemoteRegionMigrationWatcher.new(s, version.to_s).wait_for_remote_region_migration
     end

lib/miq_pglogical.rb

@@ -65,6 +65,28 @@ def replication_lag
     self.class.with_connection_error_handling { pglogical.lag_bytes }
   end
 
+  def replication_type
+    if provider?
+      :remote
+    elsif subscriber?
+      :global
+    else
+      :none
+    end
+  end
+
+  def replication_type=(desired_type)
+    return unless PglogicalSubscription.logical_replication_supported?
+
+    current_type = replication_type
+    destroy_provider if current_type == :remote
+    PglogicalSubscription.delete_all if current_type == :global
+    configure_provider if desired_type == :remote
+
+    # Do nothing to add a global
+    desired_type
+  end
+
   def replication_wal_retained
     self.class.with_connection_error_handling { pglogical.wal_retained_bytes }
   end

lib/miq_pglogical/connection_handling.rb

@@ -2,10 +2,37 @@ class MiqPglogical
   module ConnectionHandling
     extend ActiveSupport::Concern
 
+    included do
+      delegate :logical_replication_supported?, :to => :class
+    end
+
     module ClassMethods
+      def logical_replication_supported?
+        return @logical_replication_supported if defined?(@logical_replication_supported)
+
+        is_superuser = ActiveRecord::Base.connection.select_value("SELECT usesuper FROM pg_user WHERE usename = CURRENT_USER;", "SQL")
+        unless is_superuser
+          warn_bookends = ["\e[33m", "\e[0m"]
+          msg = "WARNING: Current user is NOT a superuser, logical replication will not function."
+          if $stderr.tty?
+            warn(warn_bookends.join(msg).to_s)
+          else
+            warn msg
+          end
+          _log.warn msg
+        end
+
+        @logical_replication_supported = is_superuser
+      end
+
       def with_connection_error_handling
         retry_attempted ||= false
-        yield
+        if logical_replication_supported?
+          yield
+        else
+          # Silently do no harm since logical replication is not supported.
+          nil
+        end
       rescue PG::ConnectionBad
         raise if retry_attempted
 

lib/task_helpers/development/replication.rb

@@ -34,7 +34,7 @@ def setup
           setup_global(GLOBAL)
 
           # TODO: We have the technology to watch for this and report when it's all good or bad
-          puts "Local replication is setup... try checking for users in all regions: psql -U root development_replication_99 -c \"select id from users;\""
+          puts "Local replication is setup... try checking for users in all regions: psql -U #{PG_USER} #{database(GLOBAL)} -c \"SELECT id FROM users;\""
         ensure
           restore
         end
@@ -44,6 +44,11 @@ def teardown
             teardown_global_subscription_for_region(r)
             teardown_remote_publication(r)
           end
+
+          regions = REMOTES + [GLOBAL]
+          regions.each do |r|
+            run_command("dropdb -U #{PG_USER} #{database(r)}", raise_on_error: false)
+          end
         end
 
         def setup_remote(region)
@@ -88,12 +93,12 @@ def command_environment(region)
 
         def configure_global_region(region)
           run_command("bin/rails r 'TaskHelpers::Development::Replication.configure_global_region_script'", env: command_environment(region))
-          run_command("psql -U #{PG_USER} #{database(region)} -c 'select * from pg_subscription;'")
+          run_command("psql -U #{PG_USER} #{database(region)} -c 'SELECT * FROM pg_subscription;'", raise_on_error: false)
         end
 
         def configure_remote_region(region)
           run_command("bin/rails r 'MiqRegion.replication_type = :remote'", env: command_environment(region))
-          run_command("psql -U #{PG_USER} #{database(region)} -c 'select * from pg_publication;'")
+          run_command("psql -U #{PG_USER} #{database(region)} -c 'SELECT * FROM pg_publication;'")
         end
 
         def create_region(region)
@@ -119,11 +124,11 @@ def run_command(command, raise_on_error: true, env: {})
         end
 
         def teardown_global_subscription_for_region(region)
-          run_command("psql -U #{PG_USER} #{database(GLOBAL)} -c 'drop subscription region_#{region}_subscription;'", raise_on_error: false)
+          run_command("psql -U #{PG_USER} #{database(GLOBAL)} -c 'DROP SUBSCRIPTION region_#{region}_subscription;'", raise_on_error: false)
         end
 
         def teardown_remote_publication(region)
-          run_command("psql -U #{PG_USER} #{database(region)} -c 'drop publication miq;'", raise_on_error: false)
+          run_command("psql -U #{PG_USER} #{database(region)} -c 'DROP PUBLICATION miq;'", raise_on_error: false)
         end
       end
     end

spec/lib/miq_pglogical_spec.rb

@@ -62,6 +62,78 @@
     end
   end
 
+  describe "#replication_type" do
+    it "returns :global when configured as a pglogical subscriber" do
+      allow(subject).to receive(:provider?).and_return(false)
+      allow(subject).to receive(:subscriber?).and_return(true)
+      expect(subject.replication_type).to eq(:global)
+    end
+
+    it "returns :remote when configured as a pglogical provider" do
+      allow(subject).to receive(:provider?).and_return(true)
+      allow(subject).to receive(:subscriber?).and_return(false)
+      expect(subject.replication_type).to eq(:remote)
+    end
+
+    it "returns :remote when configured as both a provider and a subscriber since subscriptions are shared across all databases of a cluster" do
+      allow(subject).to receive(:provider?).and_return(true)
+      allow(subject).to receive(:subscriber?).and_return(true)
+      expect(subject.replication_type).to eq(:remote)
+    end
+
+    it "returns :none if pglogical is not configured" do
+      allow(subject).to receive(:provider?).and_return(false)
+      allow(subject).to receive(:subscriber?).and_return(false)
+      expect(subject.replication_type).to eq(:none)
+    end
+  end
+
+  describe "#replication_type=" do
+    it "returns the replication_type, even when unchanged" do
+      allow(subject).to receive(:provider?).and_return(true)
+      allow(subject).to receive(:subscriber?).and_return(false)
+      expect(subject).to receive(:destroy_provider)
+      expect(subject).to receive(:configure_provider)
+      expect(subject.replication_type = :remote).to eq :remote
+    end
+
+    it "destroys the provider when transition is :remote -> :none" do
+      allow(subject).to receive(:provider?).and_return(true)
+      allow(subject).to receive(:subscriber?).and_return(false)
+      expect(subject).to receive(:destroy_provider)
+      expect(subject.replication_type = :none).to eq :none
+    end
+
+    it "deletes all subscriptions when transition is :global -> :none" do
+      allow(subject).to receive(:provider?).and_return(false)
+      allow(subject).to receive(:subscriber?).and_return(true)
+      expect(PglogicalSubscription).to receive(:delete_all)
+      expect(subject.replication_type = :none).to eq :none
+    end
+
+    it "creates a new provider when transition is :none -> :remote" do
+      allow(subject).to receive(:provider?).and_return(false)
+      allow(subject).to receive(:subscriber?).and_return(false)
+      expect(subject).to receive(:configure_provider)
+      expect(subject.replication_type = :remote).to eq :remote
+    end
+
+    it "deletes all subscriptions and creates a new provider when transition is :global -> :remote" do
+      allow(subject).to receive(:provider?).and_return(false)
+      allow(subject).to receive(:subscriber?).and_return(true)
+      expect(PglogicalSubscription).to receive(:delete_all)
+      expect(subject).to receive(:configure_provider)
+      expect(subject.replication_type = :remote).to eq :remote
+    end
+
+    it "destroys the provider when transition is :remote -> :global" do
+      allow(subject).to receive(:provider?).and_return(true)
+      allow(subject).to receive(:subscriber?).and_return(false)
+      expect(subject).to receive(:destroy_provider)
+      expect(subject.replication_type = :global).to eq :global
+    end
+  end
+
   describe ".save_global_region" do
     let(:subscription) { double }
     it "sets replication type for this region to 'global'" do

spec/models/miq_region_spec.rb

@@ -73,90 +73,6 @@
     end
   end
 
-  describe ".replication_type" do
-    it "returns :global when configured as a pglogical subscriber" do
-      pgl = double(:provider? => false, :subscriber? => true, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(described_class.replication_type).to eq(:global)
-    end
-
-    it "returns :remote when configured as a pglogical provider" do
-      pgl = double(:provider? => true, :subscriber? => false, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(described_class.replication_type).to eq(:remote)
-    end
-
-    it "returns :remote when configured as both a provider and a subscriber since subscriptions are shared across all databases of a cluster" do
-      pgl = double(:provider? => true, :subscriber? => true, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(described_class.replication_type).to eq(:remote)
-    end
-
-    it "returns :none if pglogical is not configured" do
-      pgl = double(:provider? => false, :subscriber? => false, :node? => false)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(described_class.replication_type).to eq(:none)
-    end
-  end
-
-  describe ".replication_type=" do
-    it "returns the replication_type, even when unchanged" do
-      pgl = double(:provider? => true, :subscriber? => false, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-      expect(described_class.replication_type = :remote).to eq :remote
-    end
-
-    it "destroys the provider when transition is :remote -> :none" do
-      pgl = double(:provider? => true, :subscriber? => false, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(pgl).to receive(:destroy_provider)
-
-      expect(described_class.replication_type = :none).to eq :none
-    end
-
-    it "deletes all subscriptions when transition is :global -> :none" do
-      pgl = double(:provider? => false, :subscriber? => true, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(PglogicalSubscription).to receive(:delete_all)
-
-      expect(described_class.replication_type = :none).to eq :none
-    end
-
-    it "creates a new provider when transition is :none -> :remote" do
-      pgl = double(:provider? => false, :subscriber? => false, :node? => false)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(pgl).to receive(:configure_provider)
-
-      expect(described_class.replication_type = :remote).to eq :remote
-    end
-
-    it "deletes all subscriptions and creates a new provider when transition is :global -> :remote" do
-      pgl = double(:provider? => false, :subscriber? => true, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(PglogicalSubscription).to receive(:delete_all)
-      expect(pgl).to receive(:configure_provider)
-
-      expect(described_class.replication_type = :remote).to eq :remote
-    end
-
-    it "destroys the provider when transition is :remote -> :global" do
-      pgl = double(:provider? => true, :subscriber? => false, :node? => true)
-      allow(MiqPglogical).to receive(:new).and_return(pgl)
-
-      expect(pgl).to receive(:destroy_provider)
-
-      expect(described_class.replication_type = :global).to eq :global
-    end
-  end
-
   describe "#api_system_auth_token" do
     it "generates the token correctly" do
       user = "admin"