Skip to content

Fix gpactivatestandby after clean shutdown. #1889

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
woblerr wants to merge 2 commits into
base: main
Choose a base branch
from
+305 −80
Open

Fix gpactivatestandby after clean shutdown. #1889

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
81688c6
Fix gpactivatestandby after clean shutdown
woblerr Aug 12, 2026
8840626
Add ASF license header to test.
woblerr Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
164 changes: 84 additions & 80 deletions gpMgmt/bin/gpactivatestandby
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,15 +17,16 @@ import os
import sys
import signal
import glob
import math
import time
import shutil
import tempfile
from datetime import datetime, timedelta

# import GPDB modules
try:
import pg as pygresql
from gppylib.commands import unix, gp, pg
from gppylib.commands.base import ExecutionError
from gppylib.db import dbconn
from gppylib.gpparseopts import OptParser, OptChecker, OptionGroup, SUPPRESS_HELP
from gppylib.gplog import get_default_logger, setup_tool_logging, enable_verbose_logging
Expand All @@ -39,6 +40,7 @@ EXECNAME = os.path.split(__file__)[-1]

# Threshold values
LOG_TIME_THRESHOLD_MINS = 120
STANDBY_ACTIVATION_TIMEOUT = 600

STANDBY_SIGNAL_FILE = "standby.signal"
POSTGRESQL_AUTO_CONF_FILE = "postgresql.auto.conf"
Expand Down Expand Up @@ -255,7 +257,24 @@ def check_standby_running(options):


#-------------------------------------------------------------------------
def check_or_start_standby(options):
def get_remaining_time(deadline, stage):
remaining_time = deadline - time.monotonic()
if remaining_time <= 0:
raise GpActivateStandbyException('Timed out waiting for %s.' % stage)
return remaining_time


#-------------------------------------------------------------------------
def wait_for_postmaster(options, deadline):
logger.info('Waiting for standby postmaster to start...')

while gp.get_postmaster_pid_locally(options.coordinator_data_dir) <= 0:
remaining_time = get_remaining_time(deadline, 'standby postmaster to start')
time.sleep(min(1, remaining_time))


#-------------------------------------------------------------------------
def check_or_start_standby(options, deadline):
"""
Check if standby postmaster is running. We need the process
to activate it, but there could be some cases where user wants
Expand All @@ -276,33 +295,19 @@ def check_or_start_standby(options):
logger.error('Use -f option to bring the system anyway')
raise GpActivateStandbyException('postmaster is not running')
else:
# XXX: We don't have enough knowledge to bring up the standby
# but there could be a situation where user needs to activate
# it anyway. We'll restart with the full options after
# promoting the standby. This is nothing but a bailout and
# this could lose some of the latest changes from the primary.
fd, trigger_file = tempfile.mkstemp(dir=options.coordinator_data_dir)

cmd = gp.GpConfigHelper("add promote trigger file",
options.coordinator_data_dir,
'promote_trigger_file',
value="'" + trigger_file + "'")
cmd.run(validateAfter=True)

with open(trigger_file, 'w') as f:
f.write('');
f.close();

start_coordinator(options)
start_coordinator(options, deadline)

return False

#-------------------------------------------------------------------------
def start_coordinator(options):
"""Starts the coordinator."""
def start_coordinator(options, deadline):
"""Starts the standby coordinator without accessing the catalog."""

logger.info('Starting standby coordinator database in utility mode...')
gp.NewGpStart.local('Start CBDB', coordinatorOnly=True, coordinatorDirectory=options.coordinator_data_dir)
logger.info('Starting standby coordinator database...')
gp.GpStandbyStart.local('Start standby coordinator',
options.coordinator_data_dir,
int(os.getenv('PGPORT')))
wait_for_postmaster(options, deadline)

#-------------------------------------------------------------------------
def stop_coordinator():
Expand All @@ -312,15 +317,20 @@ def stop_coordinator():
gp.GpStop.local('Stop CBDB', coordinatorOnly=True, fast=True)

#-------------------------------------------------------------------------
def promote_standby(coordinator_data_dir):
def promote_standby(coordinator_data_dir, deadline):
"""Promote standby"""

logger.info('Promoting standby...')
# Keeping the timeout here consistent with
# MIRROR_PROMOTION_TIMEOUT which is defined as 10 mins
promotion_timeout = int(math.ceil(
get_remaining_time(deadline, 'standby promotion')))
cmd = gp.Command('pg_ctl promote',
'pg_ctl promote -D %s -t 600' % coordinator_data_dir)
cmd.run(validateAfter=True)
'pg_ctl promote -D %s -t %s' % (coordinator_data_dir, promotion_timeout))
try:
cmd.run(validateAfter=True)
except ExecutionError:
# Replace pg_ctl's generic timeout error with an activation-stage error.
get_remaining_time(deadline, 'standby promotion')
raise
logger.info('Standby coordinator is promoted')

# After promotion run CHECKPOINT to force the new TimeLineID to be
Expand All @@ -332,77 +342,71 @@ def promote_standby(coordinator_data_dir):
# promoted coordinator's control file but the coordinator's
# postmaster will actually not be ready yet to accept database
# connections for a small period of time. Use the same
# MIRROR_PROMOTION_TIMEOUT of 10 minutes here as well.
# standby activation deadline here as well.
logger.debug('forcing CHECKPOINT to reflect new TimeLineID...')
for i in range(600):
while True:
conn = None
try:
dburl = dbconn.DbURL()
remaining_time = get_remaining_time(
deadline, 'promoted coordinator to accept database connections')
connection_timeout = max(1, min(2, int(math.ceil(remaining_time))))
dburl = dbconn.DbURL(timeout=connection_timeout, retries=1)
conn = dbconn.connect(dburl, utility=True, logConn=False)
dbconn.execSQL(conn, 'CHECKPOINT')
conn.close()
return True
except pygresql.InternalError as e:
return
except (pygresql.InternalError, dbconn.ConnectionError):
pass
time.sleep(1)
finally:
if conn is not None:
conn.close()

return False
remaining_time = get_remaining_time(
deadline, 'promoted coordinator to accept database connections')
time.sleep(min(1, remaining_time))

#-------------------------------------------------------------------------
# Main
#-------------------------------------------------------------------------

# setup logging
logger = get_default_logger()
setup_tool_logging(EXECNAME, unix.getLocalHostname(), unix.getUserName())

# parse args and options
(options_, args_) = parseargs()

# if we got a new log dir, we can now set it up.
if options_.logfile:
setup_tool_logging(EXECNAME, unix.getLocalHostname(), unix.getUserName(), logdir=options_.logfile)
def main():
setup_tool_logging(EXECNAME, unix.getLocalHostname(), unix.getUserName())

try:
warnings_generated_ = print_summary(options_)
options, _ = parseargs()

if options.logfile:
setup_tool_logging(EXECNAME, unix.getLocalHostname(), unix.getUserName(), logdir=options.logfile)

# disable keyboard interrupt to prevent users from canceling
# out of the process at a very bad time. If there is a partial
# update to the gp_configuration catalog and the user cancels
# you get stuck where you can't go forward and you can't go
# backwards.
signal.signal(signal.SIGINT, signal.SIG_IGN)
try:
warnings_generated = print_summary(options)

requires_restart = not check_or_start_standby(options_)
# Disable keyboard interrupt to prevent users from canceling out of
# the process while the activation and catalog update are in progress.
signal.signal(signal.SIGINT, signal.SIG_IGN)

# promote standby, only if the standby is running in recovery
if not requires_restart:
res = promote_standby(options_.coordinator_data_dir)
if not res:
raise GpActivateStandbyException('Timed out waiting for promoted coordinator to accept database connections.')
deadline = time.monotonic() + STANDBY_ACTIVATION_TIMEOUT
requires_restart = not check_or_start_standby(options, deadline)

# now we can access the catalog. promote action has already updated
# catalog, so array.coordinator is the old (promoted) standby at this point.
array_ = get_config()
promote_standby(options.coordinator_data_dir, deadline)

# If we forced to start utility coordinator, this is the time to restart
# cluster so that the new coordinator becomes dispatch mode.
if requires_restart:
cmd = gp.GpStop.local('CBDB restart', restart=True, datadir=options_.coordinator_data_dir)
# Promotion has updated the catalog, so array.coordinator is the
# former standby, which is now the coordinator.
array = get_config()

if requires_restart:
gp.GpStop.local('CBDB restart', restart=True,
datadir=options.coordinator_data_dir)

signal.signal(signal.SIGINT, signal.default_int_handler)

print_results(array, unix.getLocalHostname(), options)
return 1 if warnings_generated else 0
except Exception as e:
logger.fatal('Error activating standby coordinator: %s' % str(e))
return 2

# At this point, cancel isn't all that bad so re-enable
# keyboard interrupt.
signal.signal(signal.SIGINT, signal.default_int_handler)

print_results(array_, unix.getLocalHostname(), options_)

if warnings_generated_:
sys.exit(1)
else:
sys.exit(0)

except Exception as e:
logger.fatal('Error activating standby coordinator: %s' % str(e))
sys.exit(2)

sys.exit(0)
if __name__ == '__main__':
sys.exit(main())
140 changes: 140 additions & 0 deletions gpMgmt/bin/gppylib/test/unit/test_unit_gpactivatestandby.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,140 @@
#!/usr/bin/env python3
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to you under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import imp
import os

from mock import Mock, call, patch

from gppylib.test.unit.gp_unittest import GpTestCase, run_tests


class GpActivateStandbyTestCase(GpTestCase):
def setUp(self):
gpactivatestandby_file = os.path.abspath(
os.path.dirname(__file__) + '/../../../gpactivatestandby')
self.subject = imp.load_source('gpactivatestandby', gpactivatestandby_file)
self.subject.logger = Mock(spec=['info', 'debug', 'fatal'])
self.options = Mock(coordinator_data_dir='/data/coordinator', force=True,
logfile=None)

@patch('gpactivatestandby.time.monotonic', return_value=4.5)
def test_get_remaining_time_uses_existing_deadline(self, mock_monotonic):
self.assertEqual(5.5, self.subject.get_remaining_time(10, 'promotion'))

@patch('gpactivatestandby.time.monotonic', return_value=10)
def test_get_remaining_time_reports_timed_out_stage(self, mock_monotonic):
with self.assertRaisesRegex(
self.subject.GpActivateStandbyException,
'Timed out waiting for standby postmaster to start'):
self.subject.get_remaining_time(10, 'standby postmaster to start')

@patch('gpactivatestandby.time.sleep')
@patch('gpactivatestandby.time.monotonic', return_value=1)
@patch('gpactivatestandby.gp.get_postmaster_pid_locally', side_effect=[-1, 1234])
def test_wait_for_postmaster_retries_with_existing_deadline(
self, mock_get_pid, mock_monotonic, mock_sleep):
self.subject.wait_for_postmaster(self.options, 10)

self.assertEqual(2, mock_get_pid.call_count)
mock_sleep.assert_called_once_with(1)

@patch.dict(os.environ, {'PGPORT': '5432'})
@patch('gpactivatestandby.wait_for_postmaster')
@patch('gpactivatestandby.gp.GpStandbyStart.local')
def test_start_coordinator_does_not_use_gpstart(
self, mock_standby_start, mock_wait_for_postmaster):
self.subject.start_coordinator(self.options, 700)

mock_standby_start.assert_called_once_with(
'Start standby coordinator', '/data/coordinator', 5432)
mock_wait_for_postmaster.assert_called_once_with(self.options, 700)

@patch('gpactivatestandby.time.sleep')
@patch('gpactivatestandby.time.monotonic', side_effect=[100, 100, 101, 101])
@patch('gpactivatestandby.dbconn.execSQL')
@patch('gpactivatestandby.dbconn.connect')
@patch('gpactivatestandby.dbconn.DbURL')
@patch('gpactivatestandby.gp.Command')
def test_promote_standby_reuses_deadline_for_connection_retry(
self, mock_command, mock_dburl, mock_connect, mock_exec_sql,
mock_monotonic, mock_sleep):
conn = Mock()
mock_connect.side_effect = [
self.subject.pygresql.InternalError('starting up'),
conn,
]

self.subject.promote_standby('/data/coordinator', 600)

mock_command.assert_called_once_with(
'pg_ctl promote',
'pg_ctl promote -D /data/coordinator -t 500')
mock_command.return_value.run.assert_called_once_with(validateAfter=True)
self.assertEqual(2, mock_connect.call_count)
self.assertEqual([
call(timeout=2, retries=1),
call(timeout=2, retries=1),
], mock_dburl.call_args_list)
mock_exec_sql.assert_called_once_with(conn, 'CHECKPOINT')
conn.close.assert_called_once_with()
mock_sleep.assert_called_once_with(1)

@patch('gpactivatestandby.time.monotonic', side_effect=[100, 600])
@patch('gpactivatestandby.gp.Command')
def test_promote_standby_reports_stage_when_pg_ctl_exhausts_deadline(
self, mock_command, mock_monotonic):
mock_command.return_value.run.side_effect = self.subject.ExecutionError(
'pg_ctl timed out', mock_command.return_value)

with self.assertRaisesRegex(
self.subject.GpActivateStandbyException,
'Timed out waiting for standby promotion'):
self.subject.promote_standby('/data/coordinator', 600)

@patch('gpactivatestandby.print_results')
@patch('gpactivatestandby.get_config')
@patch('gpactivatestandby.promote_standby')
@patch('gpactivatestandby.check_or_start_standby', return_value=False)
@patch('gpactivatestandby.print_summary', return_value=False)
@patch('gpactivatestandby.parseargs')
@patch('gpactivatestandby.setup_tool_logging')
@patch('gpactivatestandby.signal.signal')
@patch('gpactivatestandby.time.monotonic', return_value=100)
@patch('gpactivatestandby.gp.GpStop.local')
def test_main_shares_one_deadline_across_start_and_promotion(
self, mock_gpstop, mock_monotonic, mock_signal, mock_setup_logging,
mock_parseargs, mock_print_summary, mock_check_or_start,
mock_promote, mock_get_config, mock_print_results):
mock_parseargs.return_value = (self.options, [])
array = Mock()
mock_get_config.return_value = array

self.assertEqual(0, self.subject.main())

deadline = 100 + self.subject.STANDBY_ACTIVATION_TIMEOUT
self.assertEqual([
call(self.options, deadline),
], mock_check_or_start.call_args_list)
mock_promote.assert_called_once_with('/data/coordinator', deadline)
mock_gpstop.assert_called_once_with(
'CBDB restart', restart=True, datadir='/data/coordinator')
mock_print_results.assert_called_once_with(
array, self.subject.unix.getLocalHostname(), self.options)


if __name__ == '__main__':
run_tests()
Loading
Loading