Second attempt which removes any kind proxying.
This commit is contained in:
parent
46589a1a0c
commit
0058d09e35
|
@ -1,13 +1,7 @@
|
||||||
development:
|
development:
|
||||||
prepared_statements: false
|
prepared_statements: false
|
||||||
adapter: postgresql_fallback
|
adapter: postgresql
|
||||||
host: 172.17.0.2
|
|
||||||
port: 6432
|
|
||||||
database: discourse_development
|
database: discourse_development
|
||||||
username: tgxworld
|
|
||||||
password: test
|
|
||||||
replica_host: 172.17.0.3
|
|
||||||
replica_port: 6432
|
|
||||||
min_messages: warning
|
min_messages: warning
|
||||||
pool: 5
|
pool: 5
|
||||||
timeout: 5000
|
timeout: 5000
|
||||||
|
|
|
@ -14,7 +14,7 @@ class TaskObserver
|
||||||
logger.info { "PG connection heartbeat: Master connection is not active.".freeze }
|
logger.info { "PG connection heartbeat: Master connection is not active.".freeze }
|
||||||
else
|
else
|
||||||
logger.error { "PG connection heartbeat failed with error: \"#{ex}\"" }
|
logger.error { "PG connection heartbeat failed with error: \"#{ex}\"" }
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
@ -28,15 +28,21 @@ end
|
||||||
module ActiveRecord
|
module ActiveRecord
|
||||||
module ConnectionHandling
|
module ConnectionHandling
|
||||||
def postgresql_fallback_connection(config)
|
def postgresql_fallback_connection(config)
|
||||||
master_connection = postgresql_connection(config)
|
begin
|
||||||
|
connection = postgresql_connection(config)
|
||||||
|
rescue PG::ConnectionBad => e
|
||||||
|
connection = postgresql_connection(config.dup.merge({
|
||||||
|
"host" => config["replica_host"], "port" => config["replica_port"]
|
||||||
|
}))
|
||||||
|
|
||||||
replica_connection = postgresql_connection(config.dup.merge({
|
verify_replica(connection)
|
||||||
host: config[:replica_host], port: config[:replica_port]
|
|
||||||
}))
|
|
||||||
verify_replica(replica_connection)
|
|
||||||
|
|
||||||
klass = ConnectionAdapters::PostgreSQLFallbackAdapter.proxy_pass(master_connection.class)
|
Discourse.enable_readonly_mode if !Discourse.readonly_mode?
|
||||||
klass.new(master_connection, replica_connection, logger, config)
|
|
||||||
|
start_connection_heartbeart(connection, config)
|
||||||
|
end
|
||||||
|
|
||||||
|
connection
|
||||||
end
|
end
|
||||||
|
|
||||||
private
|
private
|
||||||
|
@ -45,92 +51,24 @@ module ActiveRecord
|
||||||
value = connection.raw_connection.exec("SELECT pg_is_in_recovery()").values[0][0]
|
value = connection.raw_connection.exec("SELECT pg_is_in_recovery()").values[0][0]
|
||||||
raise "Replica database server is not in recovery mode." if value == 'f'
|
raise "Replica database server is not in recovery mode." if value == 'f'
|
||||||
end
|
end
|
||||||
end
|
|
||||||
|
|
||||||
module ConnectionAdapters
|
def interval
|
||||||
class PostgreSQLFallbackAdapter < AbstractAdapter
|
5
|
||||||
ADAPTER_NAME = "PostgreSQLFallback".freeze
|
end
|
||||||
MAX_FAILURE = 5
|
|
||||||
HEARTBEAT_INTERVAL = 5
|
|
||||||
|
|
||||||
attr_reader :main_connection
|
def start_connection_heartbeart(existing_connection, config)
|
||||||
|
timer_task = Concurrent::TimerTask.new(execution_interval: interval) do |task|
|
||||||
|
connection = postgresql_connection(config)
|
||||||
|
|
||||||
def self.all_methods(klass)
|
if connection.active?
|
||||||
methods = []
|
existing_connection.disconnect!
|
||||||
|
Discourse.disable_readonly_mode if Discourse.readonly_mode?
|
||||||
(klass.ancestors - AbstractAdapter.ancestors).each do |_klass|
|
task.shutdown
|
||||||
%w(public protected private).map do |level|
|
|
||||||
methods << _klass.send("#{level}_instance_methods", false)
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
methods.flatten.uniq.sort
|
|
||||||
end
|
|
||||||
|
|
||||||
def self.proxy_pass(klass)
|
|
||||||
Class.new(self) do
|
|
||||||
(self.all_methods(klass) - self.all_methods(self)).each do |method|
|
|
||||||
self.class_eval <<-EOF
|
|
||||||
def #{method}(*args, &block)
|
|
||||||
proxy_method(:#{method}, *args, &block)
|
|
||||||
end
|
|
||||||
EOF
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
def initialize(master_connection, replica_connection, logger, config)
|
timer_task.add_observer(TaskObserver.new)
|
||||||
super(nil, logger, config)
|
timer_task.execute
|
||||||
|
|
||||||
@master_connection = master_connection
|
|
||||||
@main_connection = @master_connection
|
|
||||||
@replica_connection = replica_connection
|
|
||||||
@failure_count = 0
|
|
||||||
load!
|
|
||||||
end
|
|
||||||
|
|
||||||
def proxy_method(method, *args, &block)
|
|
||||||
@main_connection.send(method, *args, &block)
|
|
||||||
rescue ActiveRecord::StatementInvalid => e
|
|
||||||
if e.message.include?("PG::UnableToSend") && @main_connection == @master_connection
|
|
||||||
@failure_count += 1
|
|
||||||
|
|
||||||
if @failure_count == MAX_FAILURE
|
|
||||||
Discourse.enable_readonly_mode if !Discourse.readonly_mode?
|
|
||||||
@main_connection = @replica_connection
|
|
||||||
load!
|
|
||||||
connection_heartbeart(@master_connection)
|
|
||||||
@failure_count = 0
|
|
||||||
else
|
|
||||||
proxy_method(method, *args, &block)
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
raise e
|
|
||||||
end
|
|
||||||
|
|
||||||
private
|
|
||||||
|
|
||||||
def load!
|
|
||||||
@visitor = @main_connection.visitor
|
|
||||||
@connection = @main_connection.raw_connection
|
|
||||||
end
|
|
||||||
|
|
||||||
def connection_heartbeart(connection, interval = HEARTBEAT_INTERVAL)
|
|
||||||
timer_task = Concurrent::TimerTask.new(execution_interval: interval) do |task|
|
|
||||||
connection.reconnect!
|
|
||||||
|
|
||||||
if connection.active?
|
|
||||||
@main_connection = connection
|
|
||||||
load!
|
|
||||||
Discourse.disable_readonly_mode if Discourse.readonly_mode?
|
|
||||||
task.shutdown
|
|
||||||
end
|
|
||||||
end
|
|
||||||
|
|
||||||
timer_task.add_observer(TaskObserver.new)
|
|
||||||
timer_task.execute
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -1,55 +1,67 @@
|
||||||
require 'rails_helper'
|
require 'rails_helper'
|
||||||
require_dependency 'active_record/connection_adapters/postgresql_fallback_adapter'
|
require_dependency 'active_record/connection_adapters/postgresql_fallback_adapter'
|
||||||
|
|
||||||
describe ActiveRecord::ConnectionAdapters::PostgreSQLFallbackAdapter do
|
describe ActiveRecord::ConnectionHandling do
|
||||||
let(:master_connection) { ActiveRecord::Base.connection }
|
let(:config) do
|
||||||
let(:replica_connection) { master_connection.dup }
|
ActiveRecord::Base.configurations["test"].merge({
|
||||||
let(:adapter) { described_class.new(master_connection, replica_connection, nil, nil) }
|
"adapter" => "postgresql_fallback",
|
||||||
|
"replica_host" => "localhost",
|
||||||
before :each do
|
"replica_port" => "6432"
|
||||||
ActiveRecord::Base.clear_all_connections!
|
})
|
||||||
end
|
end
|
||||||
|
|
||||||
describe "proxy_method" do
|
after do
|
||||||
context "when master connection is not active" do
|
ActiveRecord::Base.clear_all_connections!
|
||||||
|
Discourse.disable_readonly_mode
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "#postgresql_fallback_connection" do
|
||||||
|
it 'should return a PostgreSQL adapter' do
|
||||||
|
expect(ActiveRecord::Base.postgresql_fallback_connection(config))
|
||||||
|
.to be_an_instance_of(ActiveRecord::ConnectionAdapters::PostgreSQLAdapter)
|
||||||
|
end
|
||||||
|
|
||||||
|
context 'when master server is down' do
|
||||||
before do
|
before do
|
||||||
replica_connection.stubs(:send)
|
@replica_connection = mock('replica_connection')
|
||||||
master_connection.stubs(:send).raises(ActiveRecord::StatementInvalid.new('PG::UnableToSend'))
|
|
||||||
master_connection.stubs(:reconnect!)
|
|
||||||
master_connection.stubs(:active?).returns(false)
|
|
||||||
|
|
||||||
@old_const = described_class::HEARTBEAT_INTERVAL
|
ActiveRecord::Base.expects(:postgresql_connection).with(config).raises(PG::ConnectionBad)
|
||||||
described_class.const_set("HEARTBEAT_INTERVAL", 0.1)
|
|
||||||
|
ActiveRecord::Base.expects(:postgresql_connection).with(config.merge({
|
||||||
|
"host" => "localhost", "port" => "6432"
|
||||||
|
})).returns(@replica_connection)
|
||||||
|
|
||||||
|
ActiveRecord::Base.expects(:verify_replica).with(@replica_connection)
|
||||||
|
|
||||||
|
@replica_connection.expects(:disconnect!)
|
||||||
|
|
||||||
|
ActiveRecord::Base.stubs(:interval).returns(0.1)
|
||||||
|
|
||||||
|
Concurrent::TimerTask.any_instance.expects(:shutdown)
|
||||||
end
|
end
|
||||||
|
|
||||||
after do
|
it 'should failover to a replica server' do
|
||||||
Discourse.disable_readonly_mode
|
ActiveRecord::Base.postgresql_fallback_connection(config)
|
||||||
described_class.const_set("HEARTBEAT_INTERVAL", @old_const)
|
|
||||||
end
|
|
||||||
|
|
||||||
it "should set site to readonly mode and carry out failover and switch back procedures" do
|
|
||||||
expect(adapter.main_connection).to eq(master_connection)
|
|
||||||
adapter.proxy_method('some method')
|
|
||||||
expect(Discourse.readonly_mode?).to eq(true)
|
expect(Discourse.readonly_mode?).to eq(true)
|
||||||
expect(adapter.main_connection).to eq(replica_connection)
|
|
||||||
|
|
||||||
master_connection.stubs(:active?).returns(true)
|
ActiveRecord::Base.unstub(:postgresql_connection)
|
||||||
sleep 0.15
|
sleep 0.15
|
||||||
|
|
||||||
expect(Discourse.readonly_mode?).to eq(false)
|
expect(Discourse.readonly_mode?).to eq(false)
|
||||||
expect(adapter.main_connection).to eq(master_connection)
|
|
||||||
|
expect(ActiveRecord::Base.connection)
|
||||||
|
.to be_an_instance_of(ActiveRecord::ConnectionAdapters::PostgreSQLAdapter)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'should raise errors not related to the database connection' do
|
context 'when both master and replica server is down' do
|
||||||
master_connection.stubs(:send).raises(StandardError.new)
|
it 'should raise the right error' do
|
||||||
expect { adapter.proxy_method('some method') }.to raise_error(StandardError)
|
ActiveRecord::Base.expects(:postgresql_connection).raises(PG::ConnectionBad).twice
|
||||||
end
|
|
||||||
|
|
||||||
it 'should proxy methods successfully' do
|
expect { ActiveRecord::Base.postgresql_fallback_connection(config) }
|
||||||
expect(adapter.proxy_method(:execute, 'SELECT 1').values[0][0]).to eq("1")
|
.to raise_error(PG::ConnectionBad)
|
||||||
expect(adapter.proxy_method(:active?)).to eq(true)
|
end
|
||||||
expect(adapter.proxy_method(:raw_connection)).to eq(master_connection.raw_connection)
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
Loading…
Reference in New Issue