External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-03-01 16:58:46 -03:30
Code Issues Packages Projects Releases Wiki Activity

Allow wsrelay to fail without FATAL (#15191)

Browse Source 
We have not identify the root cause of wsrelay failure but attempt to make wsrelay restart itself resulted in postgres and redis connection leak. We were not able to fully identify where the redis connection leak comes from so reverting back to failing and removing startsecs 30 will prevent wsrelay to FATAL
This commit is contained in:
Hao Liu
2024-05-20 19:34:12 -04:00
committed by GitHub
parent 7de350dc3e
commit fc9064e27f
3 changed files with 58 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
awx/main/management/commands/run_wsrelay.py
View File

@@ -101,8 +101,9 @@ class Command(BaseCommand):
migrating = bool(executor.migration_plan(executor.loader.graph.leaf_nodes())) migrating = bool(executor.migration_plan(executor.loader.graph.leaf_nodes()))
connection.close() # Because of async nature, main loop will use new connection, so close this connection.close() # Because of async nature, main loop will use new connection, so close this
except Exception as exc: except Exception as exc:
logger.warning(f'Error on startup of run_wsrelay (error: {exc}), retry in 10s...') time.sleep(10) # Prevent supervisor from restarting the service too quickly and the service to enter FATAL state
time.sleep(10) # sleeping before logging because logging rely on setting which require database connection...
logger.warning(f'Error on startup of run_wsrelay (error: {exc}), slept for 10s...')
return return
# In containerized deployments, migrations happen in the task container, # In containerized deployments, migrations happen in the task container,
@@ -121,13 +122,14 @@ class Command(BaseCommand):
return return
try: try:
my_hostname = Instance.objects.my_hostname() my_hostname = Instance.objects.my_hostname() # This relies on settings.CLUSTER_HOST_ID which requires database connection
logger.info('Active instance with hostname {} is registered.'.format(my_hostname)) logger.info('Active instance with hostname {} is registered.'.format(my_hostname))
except RuntimeError as e: except RuntimeError as e:
# the CLUSTER_HOST_ID in the task, and web instance must match and # the CLUSTER_HOST_ID in the task, and web instance must match and
# ensure network connectivity between the task and web instance # ensure network connectivity between the task and web instance
logger.info('Unable to return currently active instance: {}, retry in 5s...'.format(e)) time.sleep(10) # Prevent supervisor from restarting the service too quickly and the service to enter FATAL state
time.sleep(5) # sleeping before logging because logging rely on setting which require database connection...
logger.warning(f"Unable to return currently active instance: {e}, slept for 10s before return.")
return return
if options.get('status'): if options.get('status'):
@@ -166,12 +168,14 @@ class Command(BaseCommand):
WebsocketsMetricsServer().start() WebsocketsMetricsServer().start()
while True: try:
try: logger.info('Starting Websocket Relayer...')
asyncio.run(WebSocketRelayManager().run()) websocket_relay_manager = WebSocketRelayManager()
except KeyboardInterrupt: asyncio.run(websocket_relay_manager.run())
logger.info('Shutting down Websocket Relayer') except KeyboardInterrupt:
break logger.info('Terminating Websocket Relayer')
except Exception as e: except BaseException as e: # BaseException is used to catch all exceptions including asyncio.CancelledError
logger.exception('Error in Websocket Relayer, exception: {}. Restarting in 10 seconds'.format(e)) time.sleep(10) # Prevent supervisor from restarting the service too quickly and the service to enter FATAL state
time.sleep(10) # sleeping before logging because logging rely on setting which require database connection...
logger.warning(f"Encounter error while running Websocket Relayer {e}, slept for 10s...")
return

110
awx/main/wsrelay.py
View File

@@ -285,8 +285,6 @@ class WebSocketRelayManager(object):
except asyncio.CancelledError: except asyncio.CancelledError:
# Handle the case where the task was already cancelled by the time we got here. # Handle the case where the task was already cancelled by the time we got here.
pass pass
except Exception as e:
logger.warning(f"Failed to cancel relay connection for {hostname}: {e}")
del self.relay_connections[hostname] del self.relay_connections[hostname]
@@ -297,8 +295,6 @@ class WebSocketRelayManager(object):
self.stats_mgr.delete_remote_host_stats(hostname) self.stats_mgr.delete_remote_host_stats(hostname)
except KeyError: except KeyError:
pass pass
except Exception as e:
logger.warning(f"Failed to delete stats for {hostname}: {e}")
async def run(self): async def run(self):
event_loop = asyncio.get_running_loop() event_loop = asyncio.get_running_loop()
@@ -306,7 +302,6 @@ class WebSocketRelayManager(object):
self.stats_mgr = RelayWebsocketStatsManager(event_loop, self.local_hostname) self.stats_mgr = RelayWebsocketStatsManager(event_loop, self.local_hostname)
self.stats_mgr.start() self.stats_mgr.start()
# Set up a pg_notify consumer for allowing web nodes to "provision" and "deprovision" themselves gracefully.
database_conf = deepcopy(settings.DATABASES['default']) database_conf = deepcopy(settings.DATABASES['default'])
database_conf['OPTIONS'] = deepcopy(database_conf.get('OPTIONS', {})) database_conf['OPTIONS'] = deepcopy(database_conf.get('OPTIONS', {}))
@@ -318,79 +313,54 @@ class WebSocketRelayManager(object):
if 'PASSWORD' in database_conf: if 'PASSWORD' in database_conf:
database_conf['OPTIONS']['password'] = database_conf.pop('PASSWORD') database_conf['OPTIONS']['password'] = database_conf.pop('PASSWORD')
task = None async_conn = await psycopg.AsyncConnection.connect(
dbname=database_conf['NAME'],
host=database_conf['HOST'],
user=database_conf['USER'],
port=database_conf['PORT'],
**database_conf.get("OPTIONS", {}),
)
# Managing the async_conn here so that we can close it if we need to restart the connection await async_conn.set_autocommit(True)
async_conn = None on_ws_heartbeat_task = event_loop.create_task(self.on_ws_heartbeat(async_conn))
# Establishes a websocket connection to /websocket/relay on all API servers # Establishes a websocket connection to /websocket/relay on all API servers
try: while True:
while True: if on_ws_heartbeat_task.done():
if not task or task.done(): raise Exception("on_ws_heartbeat_task has exited")
try:
# Try to close the connection if it's open
if async_conn:
try:
await async_conn.close()
except Exception as e:
logger.warning(f"Failed to close connection to database for pg_notify: {e}")
# and re-establish the connection future_remote_hosts = self.known_hosts.keys()
async_conn = await psycopg.AsyncConnection.connect( current_remote_hosts = self.relay_connections.keys()
dbname=database_conf['NAME'], deleted_remote_hosts = set(current_remote_hosts) - set(future_remote_hosts)
host=database_conf['HOST'], new_remote_hosts = set(future_remote_hosts) - set(current_remote_hosts)
user=database_conf['USER'],
port=database_conf['PORT'],
**database_conf.get("OPTIONS", {}),
)
await async_conn.set_autocommit(True)
# before creating the task that uses the connection # This loop handles if we get an advertisement from a host we already know about but
task = event_loop.create_task(self.on_ws_heartbeat(async_conn), name="on_ws_heartbeat") # the advertisement has a different IP than we are currently connected to.
logger.info("Creating `on_ws_heartbeat` task in event loop.") for hostname, address in self.known_hosts.items():
if hostname not in self.relay_connections:
# We've picked up a new hostname that we don't know about yet.
continue
except Exception as e: if address != self.relay_connections[hostname].remote_host:
logger.warning(f"Failed to connect to database for pg_notify: {e}") deleted_remote_hosts.add(hostname)
new_remote_hosts.add(hostname)
future_remote_hosts = self.known_hosts.keys() # Delete any hosts with closed connections
current_remote_hosts = self.relay_connections.keys() for hostname, relay_conn in self.relay_connections.items():
deleted_remote_hosts = set(current_remote_hosts) - set(future_remote_hosts) if not relay_conn.connected:
new_remote_hosts = set(future_remote_hosts) - set(current_remote_hosts) deleted_remote_hosts.add(hostname)
# This loop handles if we get an advertisement from a host we already know about but if deleted_remote_hosts:
# the advertisement has a different IP than we are currently connected to. logger.info(f"Removing {deleted_remote_hosts} from websocket broadcast list")
for hostname, address in self.known_hosts.items(): await asyncio.gather(*[self.cleanup_offline_host(h) for h in deleted_remote_hosts])
if hostname not in self.relay_connections:
# We've picked up a new hostname that we don't know about yet.
continue
if address != self.relay_connections[hostname].remote_host: if new_remote_hosts:
deleted_remote_hosts.add(hostname) logger.info(f"Adding {new_remote_hosts} to websocket broadcast list")
new_remote_hosts.add(hostname)
# Delete any hosts with closed connections for h in new_remote_hosts:
for hostname, relay_conn in self.relay_connections.items(): stats = self.stats_mgr.new_remote_host_stats(h)
if not relay_conn.connected: relay_connection = WebsocketRelayConnection(name=self.local_hostname, stats=stats, remote_host=self.known_hosts[h])
deleted_remote_hosts.add(hostname) relay_connection.start()
self.relay_connections[h] = relay_connection
if deleted_remote_hosts: await asyncio.sleep(settings.BROADCAST_WEBSOCKET_NEW_INSTANCE_POLL_RATE_SECONDS)
logger.info(f"Removing {deleted_remote_hosts} from websocket broadcast list")
await asyncio.gather(*[self.cleanup_offline_host(h) for h in deleted_remote_hosts])
if new_remote_hosts:
logger.info(f"Adding {new_remote_hosts} to websocket broadcast list")
for h in new_remote_hosts:
stats = self.stats_mgr.new_remote_host_stats(h)
relay_connection = WebsocketRelayConnection(name=self.local_hostname, stats=stats, remote_host=self.known_hosts[h])
relay_connection.start()
self.relay_connections[h] = relay_connection
await asyncio.sleep(settings.BROADCAST_WEBSOCKET_NEW_INSTANCE_POLL_RATE_SECONDS)
finally:
if async_conn:
logger.info("Shutting down db connection for wsrelay.")
try:
await async_conn.close()
except Exception as e:
logger.info(f"Failed to close connection to database for pg_notify: {e}")

1
tools/ansible/roles/dockerfile/templates/supervisor_task.conf.j2
View File

@@ -31,7 +31,6 @@ command = awx-manage run_wsrelay
directory = /var/lib/awx directory = /var/lib/awx
{% endif %} {% endif %}
autorestart = true autorestart = true
startsecs = 30
stopasgroup=true stopasgroup=true
killasgroup=true killasgroup=true
stdout_logfile=/dev/stdout stdout_logfile=/dev/stdout