Add a #backlog method to PglogicalSubscription objects

Gemfile

@@ -74,7 +74,7 @@ gem "open4",                          "~>1.3.0",       :require => false
 gem "outfielding-jqplot-rails",       "= 1.0.8"
 gem "ovirt-engine-sdk",               "~>4.0.6",       :require => false # Required by the oVirt provider
 gem "ovirt_metrics",                  "~>1.4.0",       :require => false
-gem "pg-pglogical",                   "~>1.0.0",       :require => false
+gem "pg-pglogical",                   "~>1.1.0",       :require => false
 gem "puma",                           "~>3.3.0"
 gem "query_relation",                 "~>0.1.0",       :require => false
 gem "rails",                          "~>5.0.1"

app/models/pglogical_subscription.rb

@@ -101,6 +101,10 @@ def validate(new_connection_params = {})
                                                  connection_hash['dbname'])
   end
 
+  def backlog
+    connection.xlog_location_diff(remote_node_lsn, remote_replication_lsn)
+  end
+
   # translate the output from the pglogical stored proc to our object columns
   def self.subscription_to_columns(sub)
     cols = sub.symbolize_keys
@@ -109,6 +113,8 @@ def self.subscription_to_columns(sub)
     cols.delete(:slot_name)
     cols.delete(:replication_sets)
     cols.delete(:forward_origins)
+    cols.delete(:remote_replication_lsn)
+    cols.delete(:local_replication_lsn)
 
     cols[:id] = cols.delete(:subscription_name)
 
@@ -164,7 +170,7 @@ def self.find_id(to_find)
   private
 
   def remote_region_number
-    MiqRegionRemote.with_remote_connection(host, port || 5432, user, decrypted_password, dbname, "postgresql") do |_conn|
+    with_remote_connection do |_conn|
       return MiqRegionRemote.region_number_from_sequence
     end
   end
@@ -217,7 +223,7 @@ def assert_valid_schemas!
     local_errors = EvmDatabase.check_schema
     raise local_errors if local_errors
     find_password if password.nil?
-    MiqRegionRemote.with_remote_connection(host, port || 5432, user, decrypted_password, dbname, "postgresql") do |conn|
+    with_remote_connection do |conn|
       remote_errors = EvmDatabase.check_schema(conn)
       raise remote_errors if remote_errors
     end
@@ -237,4 +243,19 @@ def dsn
   def decrypted_password
     MiqPassword.try_decrypt(password)
   end
+
+  def remote_replication_lsn
+    pglogical.subscription_show_status(id)["remote_replication_lsn"]
+  end
+
+  def remote_node_lsn
+    with_remote_connection(&:xlog_location)
+  end
+
+  def with_remote_connection
+    find_password
+    MiqRegionRemote.with_remote_connection(host, port || 5432, user, decrypted_password, dbname, "postgresql") do |conn|
+      yield conn
+    end
+  end
 end

lib/extensions/ar_adapter/ar_dba/postgresql.rb

@@ -11,6 +11,14 @@ def spid
     select_value("SELECT pg_backend_pid()").to_i
   end
 
+  def xlog_location
+    select_value("SELECT pg_current_xlog_insert_location()")
+  end
+
+  def xlog_location_diff(lsn1, lsn2)
+    select_value("SELECT pg_xlog_location_diff(#{quote(lsn1)}, #{quote(lsn2)})").to_i
+  end
+
   def client_connections
     data = select(<<-SQL, "Client Connections").to_a
                   SELECT client_addr   AS client_address

spec/lib/extensions/ar_dba_spec.rb

@@ -1,6 +1,18 @@
 describe "ar_dba extension" do
   let(:connection) { ApplicationRecord.connection }
 
+  describe "#xlog_location" do
+    it "returns a valid lsn" do
+      expect(connection.xlog_location).to match(%r{\h+/\h+})
+    end
+  end
+
+  describe "#xlog_location_diff" do
+    it "returns the correct xlog difference" do
+      expect(connection.xlog_location_diff("18/72F84A48", "18/72F615B8")). to eq(144_528)
+    end
+  end
+
   describe "#primary_key_index" do
     it "returns nil when there is no primary key" do
       table_name = "no_pk_test"

spec/models/pglogical_subscription_spec.rb

@@ -4,22 +4,26 @@
   let(:subscriptions) do
     [
       {
-        "subscription_name" => "region_#{remote_region1}_subscription",
-        "status"            => "replicating",
-        "provider_node"     => "region_#{remote_region1}",
-        "provider_dsn"      => "dbname = 'vmdb\\'s_test' host='example.com' user='root' port='' password='p=as\\' s\\''",
-        "slot_name"         => "pgl_vmdb_test_region_#{remote_region1}_subscripdb71d61",
-        "replication_sets"  => ["miq"],
-        "forward_origins"   => ["all"]
+        "subscription_name"      => "region_#{remote_region1}_subscription",
+        "status"                 => "replicating",
+        "provider_node"          => "region_#{remote_region1}",
+        "provider_dsn"           => "dbname = 'vmdb\\'s_test' host='example.com' user='root' port='' password='p=as\\' s\\''",
+        "slot_name"              => "pgl_vmdb_test_region_#{remote_region1}_subscripdb71d61",
+        "replication_sets"       => ["miq"],
+        "forward_origins"        => ["all"],
+        "remote_replication_lsn" => "0/420D9A0",
+        "local_replication_lsn"  => "18/72DE8268"
       },
       {
-        "subscription_name" => "region_#{remote_region2}_subscription",
-        "status"            => "disabled",
-        "provider_node"     => "region_#{remote_region2}",
-        "provider_dsn"      => "dbname = vmdb_test2 host=test.example.com user = postgres port=5432 fallback_application_name='bin/rails'",
-        "slot_name"         => "pgl_vmdb_test_region_#{remote_region2}_subscripdb71d61",
-        "replication_sets"  => ["miq"],
-        "forward_origins"   => ["all"]
+        "subscription_name"      => "region_#{remote_region2}_subscription",
+        "status"                 => "disabled",
+        "provider_node"          => "region_#{remote_region2}",
+        "provider_dsn"           => "dbname = vmdb_test2 host=test.example.com user = postgres port=5432 fallback_application_name='bin/rails'",
+        "slot_name"              => "pgl_vmdb_test_region_#{remote_region2}_subscripdb71d61",
+        "replication_sets"       => ["miq"],
+        "forward_origins"        => ["all"],
+        "remote_replication_lsn" => "1/53E9A8",
+        "local_replication_lsn"  => "20/72FF8369"
       }
     ]
   end
@@ -401,4 +405,21 @@ def with_an_invalid_remote_schema
       sub.validate
     end
   end
+
+  describe "#backlog" do
+    let(:remote_connection) { double(:remote_connection) }
+
+    before do
+      allow(pglogical).to receive(:enabled?).and_return(true)
+      allow(pglogical).to receive(:subscriptions).and_return([subscriptions.first])
+      allow(pglogical).to receive(:subscription_show_status).and_return(subscriptions.first)
+    end
+
+    it "returns the correct value" do
+      expect(MiqRegionRemote).to receive(:with_remote_connection).and_yield(remote_connection)
+      expect(remote_connection).to receive(:xlog_location).and_return("0/42108F8")
+
+      expect(described_class.first.backlog).to eq(12_120)
+    end
+  end
 end