diff --git a/awx/main/analytics/broadcast_websocket.py b/awx/main/analytics/broadcast_websocket.py index df1582c9b9..7e63d31c76 100644 --- a/awx/main/analytics/broadcast_websocket.py +++ b/awx/main/analytics/broadcast_websocket.py @@ -65,7 +65,7 @@ class FixedSlidingWindow: return sum(self.buckets.values()) or 0 -class BroadcastWebsocketStatsManager: +class RelayWebsocketStatsManager: def __init__(self, event_loop, local_hostname): self._local_hostname = local_hostname @@ -74,7 +74,7 @@ class BroadcastWebsocketStatsManager: self._redis_key = BROADCAST_WEBSOCKET_REDIS_KEY_NAME def new_remote_host_stats(self, remote_hostname): - self._stats[remote_hostname] = BroadcastWebsocketStats(self._local_hostname, remote_hostname) + self._stats[remote_hostname] = RelayWebsocketStats(self._local_hostname, remote_hostname) return self._stats[remote_hostname] def delete_remote_host_stats(self, remote_hostname): @@ -107,7 +107,7 @@ class BroadcastWebsocketStatsManager: return parser.text_string_to_metric_families(stats_str.decode('UTF-8')) -class BroadcastWebsocketStats: +class RelayWebsocketStats: def __init__(self, local_hostname, remote_hostname): self._local_hostname = local_hostname self._remote_hostname = remote_hostname diff --git a/awx/main/management/commands/run_wsrelay.py b/awx/main/management/commands/run_wsrelay.py index 105c2ae199..ce778b8356 100644 --- a/awx/main/management/commands/run_wsrelay.py +++ b/awx/main/management/commands/run_wsrelay.py @@ -13,7 +13,7 @@ from django.db import connection from django.db.migrations.executor import MigrationExecutor from awx.main.analytics.broadcast_websocket import ( - BroadcastWebsocketStatsManager, + RelayWebsocketStatsManager, safe_name, ) from awx.main.wsrelay import WebSocketRelayManager @@ -130,9 +130,9 @@ class Command(BaseCommand): if options.get('status'): try: - stats_all = BroadcastWebsocketStatsManager.get_stats_sync() + stats_all = RelayWebsocketStatsManager.get_stats_sync() except redis.exceptions.ConnectionError as e: - print(f"Unable to get Broadcast Websocket Status. Failed to connect to redis {e}") + print(f"Unable to get Relay Websocket Status. Failed to connect to redis {e}") return data = {} @@ -151,13 +151,13 @@ class Command(BaseCommand): host_stats = Command.get_connection_status(hostnames, data) lines = Command._format_lines(host_stats) - print(f'Broadcast websocket connection status from "{my_hostname}" to:') + print(f'Relay websocket connection status from "{my_hostname}" to:') print('\n'.join(lines)) host_stats = Command.get_connection_stats(hostnames, data) lines = Command._format_lines(host_stats) - print(f'\nBroadcast websocket connection stats from "{my_hostname}" to:') + print(f'\nRelay websocket connection stats from "{my_hostname}" to:') print('\n'.join(lines)) return @@ -166,4 +166,4 @@ class Command(BaseCommand): websocket_relay_manager = WebSocketRelayManager() asyncio.run(websocket_relay_manager.run()) except KeyboardInterrupt: - logger.debug('Terminating Websocket Broadcaster') + logger.debug('Terminating Websocket Relayer') diff --git a/awx/main/wsrelay.py b/awx/main/wsrelay.py index 4dd60ca834..39f7d73d03 100644 --- a/awx/main/wsrelay.py +++ b/awx/main/wsrelay.py @@ -13,8 +13,8 @@ from django.conf import settings from django.apps import apps from awx.main.analytics.broadcast_websocket import ( - BroadcastWebsocketStats, - BroadcastWebsocketStatsManager, + RelayWebsocketStats, + RelayWebsocketStatsManager, ) import awx.main.analytics.subsystem_metrics as s_metrics @@ -50,19 +50,17 @@ class WebsocketRelayConnection: def __init__( self, name, - stats: BroadcastWebsocketStats, + stats: RelayWebsocketStats, remote_host: str, remote_port: int = settings.BROADCAST_WEBSOCKET_PORT, protocol: str = settings.BROADCAST_WEBSOCKET_PROTOCOL, verify_ssl: bool = settings.BROADCAST_WEBSOCKET_VERIFY_CERT, - endpoint: str = 'relay', ): self.name = name self.event_loop = asyncio.get_event_loop() self.stats = stats self.remote_host = remote_host self.remote_port = remote_port - self.endpoint = endpoint self.protocol = protocol self.verify_ssl = verify_ssl self.channel_layer = None @@ -91,7 +89,7 @@ class WebsocketRelayConnection: logger.warning(f"Connection from {self.name} to {self.remote_host} cancelled") raise - uri = f"{self.protocol}://{self.remote_host}:{self.remote_port}/websocket/{self.endpoint}/" + uri = f"{self.protocol}://{self.remote_host}:{self.remote_port}/websocket/relay/" timeout = aiohttp.ClientTimeout(total=10) secret_val = WebsocketSecretAuthHelper.construct_secret() @@ -145,6 +143,10 @@ class WebsocketRelayConnection: logger.warning(logmsg) continue + from remote_pdb import RemotePdb + + RemotePdb('127.0.0.1', 4444).set_trace() + if payload.get("type") == "consumer.subscribe": for group in payload['groups']: name = f"{self.remote_host}-{group}" @@ -197,7 +199,7 @@ class WebSocketRelayManager(object): self.relay_connections = dict() self.local_hostname = get_local_host() self.event_loop = asyncio.get_event_loop() - self.stats_mgr = BroadcastWebsocketStatsManager(self.event_loop, self.local_hostname) + self.stats_mgr = RelayWebsocketStatsManager(self.event_loop, self.local_hostname) async def run(self): self.stats_mgr.start() diff --git a/docs/websockets.md b/docs/websockets.md index 3c4959358c..84e1226594 100644 --- a/docs/websockets.md +++ b/docs/websockets.md @@ -10,11 +10,11 @@ To communicate between our different services we use websockets. Every AWX node Inside AWX we use the `emit_channel_notification` function which places messages onto the queue. The messages are given an explicit event group and event type which we later use in our wire protocol to control message delivery to the client. -### Broadcast Backplane +### Relay Backplane Previously, AWX leveraged RabbitMQ to deliver Ansible events that emanated from one AWX node to all other AWX nodes so that any client listening and subscribed to the Websockets could get events from any running playbook. We are since moved off of RabbitMQ and onto a per-node local Redis instance. To maintain the requirement that any Websocket connection can receive events from any playbook running on any AWX node we still need to deliver every event to every AWX node. AWX does this via a fully connected Websocket backplane. -#### Broadcast Backplane Token +#### Relay Backplane Token AWX node(s) connect to every other node via the Websocket backplane. The backplane websockets initiate from the `wsrelay` process and connect to other nodes via the same nginx process that serves webpage websocket connections and marshalls incoming web/API requests. If you have configured AWX to run with an ssl terminated connection in front of nginx then you likely will have nginx configured to handle http traffic and thus the websocket connection will flow unencrypted over http. If you have nginx configured with ssl enabled, then the websocket traffic will flow encrypted. diff --git a/tools/ansible/roles/dockerfile/templates/supervisor.conf.j2 b/tools/ansible/roles/dockerfile/templates/supervisor.conf.j2 index 4435923949..4e8a52e9f7 100644 --- a/tools/ansible/roles/dockerfile/templates/supervisor.conf.j2 +++ b/tools/ansible/roles/dockerfile/templates/supervisor.conf.j2 @@ -58,23 +58,6 @@ stdout_logfile_maxbytes=0 stderr_logfile=/dev/stderr stderr_logfile_maxbytes=0 -[program:wsrelay] -{% if kube_dev | bool %} -command = make wsrelay -directory = /awx_devel -{% else %} -command = awx-manage run_wsrelay -directory = /var/lib/awx -{% endif %} -autorestart = true -startsecs = 30 -stopasgroup=true -killasgroup=true -stdout_logfile=/dev/stdout -stdout_logfile_maxbytes=0 -stderr_logfile=/dev/stderr -stderr_logfile_maxbytes=0 - [program:awx-rsyslogd] command = rsyslogd -n -i /var/run/awx-rsyslog/rsyslog.pid -f /var/lib/awx/rsyslog/rsyslog.conf autorestart = true diff --git a/tools/ansible/roles/dockerfile/templates/supervisor_task.conf.j2 b/tools/ansible/roles/dockerfile/templates/supervisor_task.conf.j2 index a2f2bd5298..9abd6b49b4 100644 --- a/tools/ansible/roles/dockerfile/templates/supervisor_task.conf.j2 +++ b/tools/ansible/roles/dockerfile/templates/supervisor_task.conf.j2 @@ -22,6 +22,23 @@ stdout_logfile_maxbytes=0 stderr_logfile=/dev/stderr stderr_logfile_maxbytes=0 +[program:wsrelay] +{% if kube_dev | bool %} +command = make wsrelay +directory = /awx_devel +{% else %} +command = awx-manage run_wsrelay +directory = /var/lib/awx +{% endif %} +autorestart = true +startsecs = 30 +stopasgroup=true +killasgroup=true +stdout_logfile=/dev/stdout +stdout_logfile_maxbytes=0 +stderr_logfile=/dev/stderr +stderr_logfile_maxbytes=0 + [program:callback-receiver] {% if kube_dev | bool %} command = make receiver