Replace rabbitmq with redis

Makefile

@@ -265,28 +265,6 @@ migrate:
 dbchange:
 	$(MANAGEMENT_COMMAND) makemigrations
 
-server_noattach:
-	tmux new-session -d -s awx 'exec make uwsgi'
-	tmux rename-window 'AWX'
-	tmux select-window -t awx:0
-	tmux split-window -v 'exec make dispatcher'
-	tmux new-window 'exec make daphne'
-	tmux select-window -t awx:1
-	tmux rename-window 'WebSockets'
-	tmux split-window -h 'exec make runworker'
-	tmux split-window -v 'exec make nginx'
-	tmux new-window 'exec make receiver'
-	tmux select-window -t awx:2
-	tmux rename-window 'Extra Services'
-	tmux select-window -t awx:0
-
-server: server_noattach
-	tmux -2 attach-session -t awx
-
-# Use with iterm2's native tmux protocol support
-servercc: server_noattach
-	tmux -2 -CC attach-session -t awx
-
 supervisor:
 	@if [ "$(VENV_BASE)" ]; then \
 		. $(VENV_BASE)/awx/bin/activate; \
@@ -311,18 +289,11 @@ daphne:
 	fi; \
 	daphne -b 127.0.0.1 -p 8051 awx.asgi:channel_layer
 
-runworker:
-	@if [ "$(VENV_BASE)" ]; then \
-		. $(VENV_BASE)/awx/bin/activate; \
-	fi; \
-	$(PYTHON) manage.py runworker --only-channels websocket.*
-
-# Run the built-in development webserver (by default on http://localhost:8013).
-runserver:
+wsbroadcast:
 	@if [ "$(VENV_BASE)" ]; then \
 		. $(VENV_BASE)/awx/bin/activate; \
 	fi; \
-	$(PYTHON) manage.py runserver
+	$(PYTHON) manage.py run_wsbroadcast
 
 # Run to start the background task dispatcher for development.
 dispatcher:

awx/api/generics.py

@@ -5,10 +5,12 @@
 import inspect
 import logging
 import time
+import uuid
 import urllib.parse
 
 # Django
 from django.conf import settings
+from django.core.cache import cache
 from django.db import connection
 from django.db.models.fields import FieldDoesNotExist
 from django.db.models.fields.related import OneToOneRel
@@ -973,14 +975,19 @@ def post(self, request, *args, **kwargs):
         if hasattr(new_obj, 'admin_role') and request.user not in new_obj.admin_role.members.all():
             new_obj.admin_role.members.add(request.user)
         if sub_objs:
+            # store the copied object dict into memcached, because it's
+            # often too large for postgres' notification bus
+            # (which has a default maximum message size of 8k)
+            key = 'deep-copy-{}'.format(str(uuid.uuid4()))
+            cache.set(key, sub_objs, timeout=3600)
             permission_check_func = None
             if hasattr(type(self), 'deep_copy_permission_check_func'):
                 permission_check_func = (
                     type(self).__module__, type(self).__name__, 'deep_copy_permission_check_func'
                 )
             trigger_delayed_deep_copy(
                 self.model.__module__, self.model.__name__,
-                obj.pk, new_obj.pk, request.user.pk, sub_objs,
+                obj.pk, new_obj.pk, request.user.pk, key,
                 permission_check_func=permission_check_func
             )
         serializer = self._get_copy_return_serializer(new_obj)

awx/api/urls/urls.py

@@ -34,7 +34,9 @@
     OAuth2ApplicationDetail,
 )
 
-from awx.api.views.metrics import MetricsView
+from awx.api.views.metrics import (
+    MetricsView,
+)
 
 from .organization import urls as organization_urls
 from .user import urls as user_urls

awx/api/views/metrics.py

@@ -39,3 +39,4 @@ def get(self, request):
         if (request.user.is_superuser or request.user.is_system_auditor):
             return Response(metrics().decode('UTF-8'))
         raise PermissionDenied()
+

awx/asgi.py

@@ -2,14 +2,15 @@
 # All Rights Reserved.
 import os
 import logging
+import django
 from awx import __version__ as tower_version
 
 # Prepare the AWX environment.
 from awx import prepare_env, MODE
 prepare_env() # NOQA
 
-from django.core.wsgi import get_wsgi_application  # NOQA
-from channels.asgi import get_channel_layer
+from channels.routing import get_default_application
+
 
 """
 ASGI config for AWX project.
@@ -32,6 +33,5 @@
 
 
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "awx.settings")
-
-
-channel_layer = get_channel_layer()
+django.setup()
+channel_layer = get_default_application()

awx/main/analytics/broadcast_websocket.py

@@ -0,0 +1,166 @@
+import datetime
+import asyncio
+import logging
+import aioredis
+import redis
+import re
+
+from prometheus_client import (
+    generate_latest,
+    Gauge,
+    Counter,
+    Enum,
+    CollectorRegistry,
+)
+
+from django.conf import settings
+
+
+BROADCAST_WEBSOCKET_REDIS_KEY_NAME = 'broadcast_websocket_stats'
+
+
+logger = logging.getLogger('awx.main.analytics.broadcast_websocket')
+
+
+def dt_to_seconds(dt):
+    return int((dt - datetime.datetime(1970,1,1)).total_seconds())
+
+
+def now_seconds():
+    return dt_to_seconds(datetime.datetime.now())
+
+
+# Second granularity; Per-minute
+class FixedSlidingWindow():
+    def __init__(self, start_time=None):
+        self.buckets = dict()
+        self.start_time = start_time or now_seconds()
+
+    def cleanup(self, now_bucket=None):
+        now_bucket = now_bucket or now_seconds()
+        if self.start_time + 60 <= now_bucket:
+            self.start_time = now_bucket + 60 + 1
+
+            # Delete old entries
+            for k in list(self.buckets.keys()):
+                if k < self.start_time:
+                    del self.buckets[k]
+
+    def record(self, ts=None):
+        ts = ts or datetime.datetime.now()
+        now_bucket = int((ts - datetime.datetime(1970,1,1)).total_seconds())
+
+        val = self.buckets.get(now_bucket, 0)
+        self.buckets[now_bucket] = val + 1
+
+        self.cleanup(now_bucket)
+
+    def render(self):
+        self.cleanup()
+        return sum(self.buckets.values()) or 0
+
+
+class BroadcastWebsocketStatsManager():
+    def __init__(self, event_loop, local_hostname):
+        self._local_hostname = local_hostname
+
+        self._event_loop = event_loop
+        self._stats = dict()
+        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)
+        return self._stats[remote_hostname]
+
+    def delete_remote_host_stats(self, remote_hostname):
+        del self._stats[remote_hostname]
+
+    async def run_loop(self):
+        try:
+            redis_conn = await aioredis.create_redis_pool(settings.BROKER_URL)
+            while True:
+                stats_data_str = ''.join(stat.serialize() for stat in self._stats.values())
+                await redis_conn.set(self._redis_key, stats_data_str)
+
+                await asyncio.sleep(settings.BROADCAST_WEBSOCKET_STATS_POLL_RATE_SECONDS)
+        except Exception as e:
+            logger.warn(e)
+            await asyncio.sleep(settings.BROADCAST_WEBSOCKET_STATS_POLL_RATE_SECONDS)
+            self.start()
+
+    def start(self):
+        self.async_task = self._event_loop.create_task(self.run_loop())
+        return self.async_task
+
+    @classmethod
+    def get_stats_sync(cls):
+        '''
+        Stringified verion of all the stats
+        '''
+        redis_conn = redis.Redis.from_url(settings.BROKER_URL)
+        return redis_conn.get(BROADCAST_WEBSOCKET_REDIS_KEY_NAME)
+
+
+class BroadcastWebsocketStats():
+    def __init__(self, local_hostname, remote_hostname):
+        self._local_hostname = local_hostname
+        self._remote_hostname = remote_hostname
+        self._registry = CollectorRegistry()
+
+        # TODO: More robust replacement
+        self.name = self.safe_name(self._local_hostname)
+        self.remote_name = self.safe_name(self._remote_hostname)
+
+        self._messages_received_total = Counter(f'awx_{self.remote_name}_messages_received_total',
+                                                'Number of messages received, to be forwarded, by the broadcast websocket system',
+                                                registry=self._registry)
+        self._messages_received = Gauge(f'awx_{self.remote_name}_messages_received',
+                                        'Number forwarded messages received by the broadcast websocket system, for the duration of the current connection',
+                                        registry=self._registry)
+        self._connection = Enum(f'awx_{self.remote_name}_connection',
+                                'Websocket broadcast connection',
+                                states=['disconnected', 'connected'],
+                                registry=self._registry)
+        self._connection_start = Gauge(f'awx_{self.remote_name}_connection_start',
+                                       'Time the connection was established',
+                                       registry=self._registry)
+
+        self._messages_received_per_minute = Gauge(f'awx_{self.remote_name}_messages_received_per_minute',
+                                                   'Messages received per minute',
+                                                   registry=self._registry)
+        self._internal_messages_received_per_minute = FixedSlidingWindow()
+
+    def safe_name(self, s):
+        # Replace all non alpha-numeric characters with _
+        return re.sub('[^0-9a-zA-Z]+', '_', s)
+
+    def unregister(self):
+        self._registry.unregister(f'awx_{self.remote_name}_messages_received')
+        self._registry.unregister(f'awx_{self.remote_name}_connection')
+
+    def record_message_received(self):
+        self._internal_messages_received_per_minute.record()
+        self._messages_received.inc()
+        self._messages_received_total.inc()
+
+    def record_connection_established(self):
+        self._connection.state('connected')
+        self._connection_start.set_to_current_time()
+        self._messages_received.set(0)
+
+    def record_connection_lost(self):
+        self._connection.state('disconnected')
+
+    def get_connection_duration(self):
+        return (datetime.datetime.now() - self._connection_established_ts).total_seconds()
+
+    def render(self):
+        msgs_per_min = self._internal_messages_received_per_minute.render()
+        self._messages_received_per_minute.set(msgs_per_min)
+
+    def serialize(self):
+        self.render()
+
+        registry_data = generate_latest(self._registry).decode('UTF-8')
+        return registry_data