Fix Redis broken pipe error in queue.py with reconnection logic and enhanced diagnostics for long-running jobs

Update code to pull subscriptions from
console.redhat.com instead of
subscription.rhsm.redhat.com

Uses service account client ID and client secret
instead of username/password, which is being
deprecated in July 2025.

Additional changes:

- In awx.awx.subscriptions module, use new service
account params rather than old basic auth params

- Update awx.awx.license module to use subscription_id
instead of pool_id. This is due to using a different API,
which identifies unique subscriptions by subscriptionID
instead of pool ID.

Signed-off-by: Seth Foster <fosterbseth@gmail.com>
Co-authored-by: Chris Meyers <chris.meyers.fsu@gmail.com>
Co-authored-by: Peter Braun <pbraun@redhat.com>

.gitignore

@@ -150,6 +150,8 @@ use_dev_supervisor.txt
 
 awx/ui/src
 awx/ui/build
+awx/ui/.ui-built
+awx/ui_next
 
 # Docs build stuff
 docs/docsite/build/

awx/api/views/analytics.py

@@ -10,7 +10,7 @@ from awx.api.generics import APIView, Response
 from awx.api.permissions import AnalyticsPermission
 from awx.api.versioning import reverse
 from awx.main.utils import get_awx_version
-from awx.main.utils.analytics_proxy import OIDCClient, DEFAULT_OIDC_TOKEN_ENDPOINT
+from awx.main.utils.analytics_proxy import OIDCClient
 from rest_framework import status
 
 from collections import OrderedDict
@@ -205,7 +205,7 @@ class AnalyticsGenericView(APIView):
             try:
                 rh_user = self._get_setting('REDHAT_USERNAME', None, ERROR_MISSING_USER)
                 rh_password = self._get_setting('REDHAT_PASSWORD', None, ERROR_MISSING_PASSWORD)
-                client = OIDCClient(rh_user, rh_password, DEFAULT_OIDC_TOKEN_ENDPOINT, ['api.console'])
+                client = OIDCClient(rh_user, rh_password)
                 response = client.make_request(
                     method,
                     url,
@@ -219,8 +219,8 @@ class AnalyticsGenericView(APIView):
                 logger.error("Automation Analytics API request failed, trying base auth method")
                 response = self._base_auth_request(request, method, url, rh_user, rh_password, headers)
             except MissingSettings:
-                rh_user = self._get_setting('SUBSCRIPTIONS_USERNAME', None, ERROR_MISSING_USER)
-                rh_password = self._get_setting('SUBSCRIPTIONS_PASSWORD', None, ERROR_MISSING_PASSWORD)
+                rh_user = self._get_setting('SUBSCRIPTIONS_CLIENT_ID', None, ERROR_MISSING_USER)
+                rh_password = self._get_setting('SUBSCRIPTIONS_CLIENT_SECRET', None, ERROR_MISSING_PASSWORD)
                 response = self._base_auth_request(request, method, url, rh_user, rh_password, headers)
             #
             # Missing or wrong user/pass

awx/api/views/root.py

@@ -32,6 +32,7 @@ from awx.api.versioning import URLPathVersioning, reverse, drf_reverse
 from awx.main.constants import PRIVILEGE_ESCALATION_METHODS
 from awx.main.models import Project, Organization, Instance, InstanceGroup, JobTemplate
 from awx.main.utils import set_environ
+from awx.main.utils.analytics_proxy import TokenError
 from awx.main.utils.licensing import get_licenser
 
 logger = logging.getLogger('awx.api.views.root')
@@ -176,19 +177,21 @@ class ApiV2SubscriptionView(APIView):
 
     def post(self, request):
         data = request.data.copy()
-        if data.get('subscriptions_password') == '$encrypted$':
-            data['subscriptions_password'] = settings.SUBSCRIPTIONS_PASSWORD
+        if data.get('subscriptions_client_secret') == '$encrypted$':
+            data['subscriptions_client_secret'] = settings.SUBSCRIPTIONS_CLIENT_SECRET
         try:
-            user, pw = data.get('subscriptions_username'), data.get('subscriptions_password')
+            user, pw = data.get('subscriptions_client_id'), data.get('subscriptions_client_secret')
             with set_environ(**settings.AWX_TASK_ENV):
                 validated = get_licenser().validate_rh(user, pw)
             if user:
-                settings.SUBSCRIPTIONS_USERNAME = data['subscriptions_username']
+                settings.SUBSCRIPTIONS_CLIENT_ID = data['subscriptions_client_id']
             if pw:
-                settings.SUBSCRIPTIONS_PASSWORD = data['subscriptions_password']
+                settings.SUBSCRIPTIONS_CLIENT_SECRET = data['subscriptions_client_secret']
         except Exception as exc:
             msg = _("Invalid Subscription")
-            if isinstance(exc, requests.exceptions.HTTPError) and getattr(getattr(exc, 'response', None), 'status_code', None) == 401:
+            if isinstance(exc, TokenError) or (
+                isinstance(exc, requests.exceptions.HTTPError) and getattr(getattr(exc, 'response', None), 'status_code', None) == 401
+            ):
                 msg = _("The provided credentials are invalid (HTTP 401).")
             elif isinstance(exc, requests.exceptions.ProxyError):
                 msg = _("Unable to connect to proxy server.")
@@ -215,12 +218,12 @@ class ApiV2AttachView(APIView):
 
     def post(self, request):
         data = request.data.copy()
-        pool_id = data.get('pool_id', None)
-        if not pool_id:
-            return Response({"error": _("No subscription pool ID provided.")}, status=status.HTTP_400_BAD_REQUEST)
-        user = getattr(settings, 'SUBSCRIPTIONS_USERNAME', None)
-        pw = getattr(settings, 'SUBSCRIPTIONS_PASSWORD', None)
-        if pool_id and user and pw:
+        subscription_id = data.get('subscription_id', None)
+        if not subscription_id:
+            return Response({"error": _("No subscription ID provided.")}, status=status.HTTP_400_BAD_REQUEST)
+        user = getattr(settings, 'SUBSCRIPTIONS_CLIENT_ID', None)
+        pw = getattr(settings, 'SUBSCRIPTIONS_CLIENT_SECRET', None)
+        if subscription_id and user and pw:
             data = request.data.copy()
             try:
                 with set_environ(**settings.AWX_TASK_ENV):
@@ -239,7 +242,7 @@ class ApiV2AttachView(APIView):
                     logger.exception(smart_str(u"Invalid subscription submitted."), extra=dict(actor=request.user.username))
                 return Response({"error": msg}, status=status.HTTP_400_BAD_REQUEST)
         for sub in validated:
-            if sub['pool_id'] == pool_id:
+            if sub['subscription_id'] == subscription_id:
                 sub['valid_key'] = True
                 settings.LICENSE = sub
                 return Response(sub)

awx/conf/fields.py

@@ -207,7 +207,8 @@ class URLField(CharField):
         if self.allow_plain_hostname:
             try:
                 url_parts = urlparse.urlsplit(value)
-                if url_parts.hostname and '.' not in url_parts.hostname:
+                looks_like_ipv6 = bool(url_parts.netloc and url_parts.netloc.startswith('[') and url_parts.netloc.endswith(']'))
+                if not looks_like_ipv6 and url_parts.hostname and '.' not in url_parts.hostname:
                     netloc = '{}.local'.format(url_parts.hostname)
                     if url_parts.port:
                         netloc = '{}:{}'.format(netloc, url_parts.port)

awx/conf/migrations/_subscriptions.py

@@ -27,5 +27,5 @@ def _migrate_setting(apps, old_key, new_key, encrypted=False):
 
 
 def prefill_rh_credentials(apps, schema_editor):
-    _migrate_setting(apps, 'REDHAT_USERNAME', 'SUBSCRIPTIONS_USERNAME', encrypted=False)
-    _migrate_setting(apps, 'REDHAT_PASSWORD', 'SUBSCRIPTIONS_PASSWORD', encrypted=True)
+    _migrate_setting(apps, 'REDHAT_USERNAME', 'SUBSCRIPTIONS_CLIENT_ID', encrypted=False)
+    _migrate_setting(apps, 'REDHAT_PASSWORD', 'SUBSCRIPTIONS_CLIENT_SECRET', encrypted=True)

awx/conf/tests/unit/test_fields.py

@@ -128,3 +128,41 @@ class TestURLField:
         else:
             with pytest.raises(ValidationError):
                 field.run_validators(url)
+
+    @pytest.mark.parametrize(
+        "url, expect_error",
+        [
+            ("https://[1:2:3]", True),
+            ("http://[1:2:3]", True),
+            ("https://[2001:db8:3333:4444:5555:6666:7777:8888", True),
+            ("https://2001:db8:3333:4444:5555:6666:7777:8888", True),
+            ("https://[2001:db8:3333:4444:5555:6666:7777:8888]", False),
+            ("https://[::1]", False),
+            ("https://[::]", False),
+            ("https://[2001:db8::1]", False),
+            ("https://[2001:db8:0:0:0:0:1:1]", False),
+            ("https://[fe80::2%eth0]", True),  # ipv6 scope identifier
+            ("https://[fe80:0:0:0:200:f8ff:fe21:67cf]", False),
+            ("https://[::ffff:192.168.1.10]", False),
+            ("https://[0:0:0:0:0:ffff:c000:0201]", False),
+            ("https://[2001:0db8:000a:0001:0000:0000:0000:0000]", False),
+            ("https://[2001:db8:a:1::]", False),
+            ("https://[ff02::1]", False),
+            ("https://[ff02:0:0:0:0:0:0:1]", False),
+            ("https://[fc00::1]", False),
+            ("https://[fd12:3456:789a:1::1]", False),
+            ("https://[2001:db8::abcd:ef12:3456:7890]", False),
+            ("https://[2001:db8:0000:abcd:0000:ef12:0000:3456]", False),
+            ("https://[::ffff:10.0.0.1]", False),
+            ("https://[2001:db8:cafe::]", False),
+            ("https://[2001:db8:cafe:0:0:0:0:0]", False),
+            ("https://[fe80::210:f3ff:fedf:4567%3]", True),  # ipv6 scope identifier, numerical interface
+        ],
+    )
+    def test_ipv6_urls(self, url, expect_error):
+        field = URLField()
+        if expect_error:
+            with pytest.raises(ValidationError, match="Enter a valid URL"):
+                field.run_validators(url)
+        else:
+            field.run_validators(url)

awx/main/analytics/core.py

@@ -22,7 +22,7 @@ from ansible_base.lib.utils.db import advisory_lock
 from awx.main.models import Job
 from awx.main.access import access_registry
 from awx.main.utils import get_awx_http_client_headers, set_environ, datetime_hook
-from awx.main.utils.analytics_proxy import OIDCClient, DEFAULT_OIDC_TOKEN_ENDPOINT
+from awx.main.utils.analytics_proxy import OIDCClient
 
 __all__ = ['register', 'gather', 'ship']
 
@@ -186,7 +186,7 @@ def gather(dest=None, module=None, subset=None, since=None, until=None, collecti
 
         if not (
             settings.AUTOMATION_ANALYTICS_URL
-            and ((settings.REDHAT_USERNAME and settings.REDHAT_PASSWORD) or (settings.SUBSCRIPTIONS_USERNAME and settings.SUBSCRIPTIONS_PASSWORD))
+            and ((settings.REDHAT_USERNAME and settings.REDHAT_PASSWORD) or (settings.SUBSCRIPTIONS_CLIENT_ID and settings.SUBSCRIPTIONS_CLIENT_SECRET))
         ):
             logger.log(log_level, "Not gathering analytics, configuration is invalid. Use --dry-run to gather locally without sending.")
             return None
@@ -368,8 +368,20 @@ def ship(path):
         logger.error('AUTOMATION_ANALYTICS_URL is not set')
         return False
 
-    rh_user = getattr(settings, 'REDHAT_USERNAME', None)
-    rh_password = getattr(settings, 'REDHAT_PASSWORD', None)
+    rh_id = getattr(settings, 'REDHAT_USERNAME', None)
+    rh_secret = getattr(settings, 'REDHAT_PASSWORD', None)
+
+    if not (rh_id and rh_secret):
+        rh_id = getattr(settings, 'SUBSCRIPTIONS_CLIENT_ID', None)
+        rh_secret = getattr(settings, 'SUBSCRIPTIONS_CLIENT_SECRET', None)
+
+    if not rh_id:
+        logger.error('Neither REDHAT_USERNAME nor SUBSCRIPTIONS_CLIENT_ID are set')
+        return False
+
+    if not rh_secret:
+        logger.error('Neither REDHAT_PASSWORD nor SUBSCRIPTIONS_CLIENT_SECRET are set')
+        return False
 
     with open(path, 'rb') as f:
         files = {'file': (os.path.basename(path), f, settings.INSIGHTS_AGENT_MIME)}
@@ -377,25 +389,13 @@ def ship(path):
         s.headers = get_awx_http_client_headers()
         s.headers.pop('Content-Type')
         with set_environ(**settings.AWX_TASK_ENV):
-            if rh_user and rh_password:
-                try:
-                    client = OIDCClient(rh_user, rh_password, DEFAULT_OIDC_TOKEN_ENDPOINT, ['api.console'])
-                    response = client.make_request("POST", url, headers=s.headers, files=files, verify=settings.INSIGHTS_CERT_PATH, timeout=(31, 31))
-                except requests.RequestException:
-                    logger.error("Automation Analytics API request failed, trying base auth method")
-                    response = s.post(url, files=files, verify=settings.INSIGHTS_CERT_PATH, auth=(rh_user, rh_password), headers=s.headers, timeout=(31, 31))
-            elif not rh_user or not rh_password:
-                logger.info('REDHAT_USERNAME and REDHAT_PASSWORD are not set, using SUBSCRIPTIONS_USERNAME and SUBSCRIPTIONS_PASSWORD')
-                rh_user = getattr(settings, 'SUBSCRIPTIONS_USERNAME', None)
-                rh_password = getattr(settings, 'SUBSCRIPTIONS_PASSWORD', None)
-                if rh_user and rh_password:
-                    response = s.post(url, files=files, verify=settings.INSIGHTS_CERT_PATH, auth=(rh_user, rh_password), headers=s.headers, timeout=(31, 31))
-                elif not rh_user:
-                    logger.error('REDHAT_USERNAME and SUBSCRIPTIONS_USERNAME are not set')
-                    return False
-                elif not rh_password:
-                    logger.error('REDHAT_PASSWORD and SUBSCRIPTIONS_USERNAME are not set')
-                    return False
+            try:
+                client = OIDCClient(rh_id, rh_secret)
+                response = client.make_request("POST", url, headers=s.headers, files=files, verify=settings.INSIGHTS_CERT_PATH, timeout=(31, 31))
+            except requests.RequestException:
+                logger.error("Automation Analytics API request failed, trying base auth method")
+                response = s.post(url, files=files, verify=settings.INSIGHTS_CERT_PATH, auth=(rh_id, rh_secret), headers=s.headers, timeout=(31, 31))
+
         # Accept 2XX status_codes
         if response.status_code >= 300:
             logger.error('Upload failed with status {}, {}'.format(response.status_code, response.text))

awx/main/conf.py

@@ -124,8 +124,8 @@ register(
     allow_blank=True,
     encrypted=False,
     read_only=False,
-    label=_('Red Hat customer username'),
-    help_text=_('This username is used to send data to Automation Analytics'),
+    label=_('Red Hat Client ID for Analytics'),
+    help_text=_('Client ID used to send data to Automation Analytics'),
     category=_('System'),
     category_slug='system',
 )
@@ -137,34 +137,34 @@ register(
     allow_blank=True,
     encrypted=True,
     read_only=False,
-    label=_('Red Hat customer password'),
-    help_text=_('This password is used to send data to Automation Analytics'),
+    label=_('Red Hat Client Secret for Analytics'),
+    help_text=_('Client secret used to send data to Automation Analytics'),
     category=_('System'),
     category_slug='system',
 )
 
 register(
-    'SUBSCRIPTIONS_USERNAME',
+    'SUBSCRIPTIONS_CLIENT_ID',
     field_class=fields.CharField,
     default='',
     allow_blank=True,
     encrypted=False,
     read_only=False,
-    label=_('Red Hat or Satellite username'),
-    help_text=_('This username is used to retrieve subscription and content information'),  # noqa
+    label=_('Red Hat Client ID for Subscriptions'),
+    help_text=_('Client ID used to retrieve subscription and content information'),  # noqa
     category=_('System'),
     category_slug='system',
 )
 
 register(
-    'SUBSCRIPTIONS_PASSWORD',
+    'SUBSCRIPTIONS_CLIENT_SECRET',
     field_class=fields.CharField,
     default='',
     allow_blank=True,
     encrypted=True,
     read_only=False,
-    label=_('Red Hat or Satellite password'),
-    help_text=_('This password is used to retrieve subscription and content information'),  # noqa
+    label=_('Red Hat Client Secret for Subscriptions'),
+    help_text=_('Client secret used to retrieve subscription and content information'),  # noqa
     category=_('System'),
     category_slug='system',
 )

awx/main/dispatch/pool.py

@@ -7,6 +7,7 @@ import time
 import traceback
 from datetime import datetime
 from uuid import uuid4
+import json
 
 import collections
 from multiprocessing import Process
@@ -25,7 +26,10 @@ from ansible_base.lib.logging.runtime import log_excess_runtime
 
 from awx.main.models import UnifiedJob
 from awx.main.dispatch import reaper
-from awx.main.utils.common import convert_mem_str_to_bytes, get_mem_effective_capacity
+from awx.main.utils.common import get_mem_effective_capacity, get_corrected_memory, get_corrected_cpu, get_cpu_effective_capacity
+
+# ansible-runner
+from ansible_runner.utils.capacity import get_mem_in_bytes, get_cpu_count
 
 if 'run_callback_receiver' in sys.argv:
     logger = logging.getLogger('awx.main.commands.run_callback_receiver')
@@ -307,6 +311,41 @@ class WorkerPool(object):
             logger.exception('could not kill {}'.format(worker.pid))
 
 
+def get_auto_max_workers():
+    """Method we normally rely on to get max_workers
+
+    Uses almost same logic as Instance.local_health_check
+    The important thing is to be MORE than Instance.capacity
+    so that the task-manager does not over-schedule this node
+
+    Ideally we would just use the capacity from the database plus reserve workers,
+    but this poses some bootstrap problems where OCP task containers
+    register themselves after startup
+    """
+    # Get memory from ansible-runner
+    total_memory_gb = get_mem_in_bytes()
+
+    # This may replace memory calculation with a user override
+    corrected_memory = get_corrected_memory(total_memory_gb)
+
+    # Get same number as max forks based on memory, this function takes memory as bytes
+    mem_capacity = get_mem_effective_capacity(corrected_memory, is_control_node=True)
+
+    # Follow same process for CPU capacity constraint
+    cpu_count = get_cpu_count()
+    corrected_cpu = get_corrected_cpu(cpu_count)
+    cpu_capacity = get_cpu_effective_capacity(corrected_cpu, is_control_node=True)
+
+    # Here is what is different from health checks,
+    auto_max = max(mem_capacity, cpu_capacity)
+
+    # add magic number of extra workers to ensure
+    # we have a few extra workers to run the heartbeat
+    auto_max += 7
+
+    return auto_max
+
+
 class AutoscalePool(WorkerPool):
     """
     An extended pool implementation that automatically scales workers up and
@@ -320,19 +359,7 @@ class AutoscalePool(WorkerPool):
         super(AutoscalePool, self).__init__(*args, **kwargs)
 
         if self.max_workers is None:
-            settings_absmem = getattr(settings, 'SYSTEM_TASK_ABS_MEM', None)
-            if settings_absmem is not None:
-                # There are 1073741824 bytes in a gigabyte. Convert bytes to gigabytes by dividing by 2**30
-                total_memory_gb = convert_mem_str_to_bytes(settings_absmem) // 2**30
-            else:
-                total_memory_gb = (psutil.virtual_memory().total >> 30) + 1  # noqa: round up
-
-            # Get same number as max forks based on memory, this function takes memory as bytes
-            self.max_workers = get_mem_effective_capacity(total_memory_gb * 2**30)
-
-            # add magic prime number of extra workers to ensure
-            # we have a few extra workers to run the heartbeat
-            self.max_workers += 7
+            self.max_workers = get_auto_max_workers()
 
         # max workers can't be less than min_workers
         self.max_workers = max(self.min_workers, self.max_workers)
@@ -346,6 +373,9 @@ class AutoscalePool(WorkerPool):
         self.scale_up_ct = 0
         self.worker_count_max = 0
 
+        # last time we wrote current tasks, to avoid too much log spam
+        self.last_task_list_log = time.monotonic()
+
     def produce_subsystem_metrics(self, metrics_object):
         metrics_object.set('dispatcher_pool_scale_up_events', self.scale_up_ct)
         metrics_object.set('dispatcher_pool_active_task_count', sum(len(w.managed_tasks) for w in self.workers))
@@ -463,6 +493,14 @@ class AutoscalePool(WorkerPool):
                 self.worker_count_max = new_worker_ct
             return ret
 
+    @staticmethod
+    def fast_task_serialization(current_task):
+        try:
+            return str(current_task.get('task')) + ' - ' + str(sorted(current_task.get('args', []))) + ' - ' + str(sorted(current_task.get('kwargs', {})))
+        except Exception:
+            # just make sure this does not make things worse
+            return str(current_task)
+
     def write(self, preferred_queue, body):
         if 'guid' in body:
             set_guid(body['guid'])
@@ -484,6 +522,15 @@ class AutoscalePool(WorkerPool):
                 if isinstance(body, dict):
                     task_name = body.get('task')
                 logger.warning(f'Workers maxed, queuing {task_name}, load: {sum(len(w.managed_tasks) for w in self.workers)} / {len(self.workers)}')
+                # Once every 10 seconds write out task list for debugging
+                if time.monotonic() - self.last_task_list_log >= 10.0:
+                    task_counts = {}
+                    for worker in self.workers:
+                        task_slug = self.fast_task_serialization(worker.current_task)
+                        task_counts.setdefault(task_slug, 0)
+                        task_counts[task_slug] += 1
+                    logger.info(f'Running tasks by count:\n{json.dumps(task_counts, indent=2)}')
+                    self.last_task_list_log = time.monotonic()
                 return super(AutoscalePool, self).write(preferred_queue, body)
         except Exception:
             for conn in connections.all():

awx/main/dispatch/worker/base.py

@@ -238,7 +238,7 @@ class AWXConsumerPG(AWXConsumerBase):
     def run(self, *args, **kwargs):
         super(AWXConsumerPG, self).run(*args, **kwargs)
 
-        logger.info(f"Running worker {self.name} listening to queues {self.queues}")
+        logger.info(f"Running {self.name}, workers min={self.pool.min_workers} max={self.pool.max_workers}, listening to queues {self.queues}")
         init = False
 
         while True:

awx/main/queue.py

@@ -1,6 +1,9 @@
 # Copyright (c) 2015 Ansible, Inc.
 # All Rights Reserved.
 
+# AIA: Primarily AI, Modified content, Human-initiated, Reviewed, Claude (Anthropic AI) via Cursor
+# AIA PAI Mc Hin R Claude Cursor - https://aiattribution.github.io/interpret-attribution
+
 # Python
 import json
 import logging
@@ -26,7 +29,151 @@ class CallbackQueueDispatcher(object):
     def __init__(self):
         self.queue = getattr(settings, 'CALLBACK_QUEUE', '')
         self.logger = logging.getLogger('awx.main.queue.CallbackQueueDispatcher')
-        self.connection = redis.Redis.from_url(settings.BROKER_URL)
+        self._broker_url = settings.BROKER_URL
+        self.connection = redis.Redis.from_url(self._broker_url)
+        self._connection_failures = 0
+        self._max_reconnect_attempts = 3
+        self._total_reconnections = 0
+        self._events_lost = 0
+
+    def _reconnect(self):
+        """
+        Attempt to reconnect to Redis after connection failure.
+
+        Returns:
+            bool: True if reconnection successful, False otherwise
+        """
+        try:
+            attempt = self._connection_failures + 1
+            self.logger.warning(
+                f"Redis reconnection attempt {attempt}/{self._max_reconnect_attempts} " f"(total reconnections this session: {self._total_reconnections})"
+            )
+
+            # Create new connection
+            self.connection = redis.Redis.from_url(self._broker_url)
+
+            # Verify connection works
+            self.connection.ping()
+
+            # Success
+            self._connection_failures = 0
+            self._total_reconnections += 1
+            self.logger.info(f"Successfully reconnected to Redis (session reconnections: {self._total_reconnections})")
+            return True
+
+        except Exception as e:
+            self._connection_failures += 1
+            self.logger.error(f"Redis reconnection failed (attempt {self._connection_failures}): {type(e).__name__}: {e}")
+            return False
 
     def dispatch(self, obj):
-        self.connection.rpush(self.queue, json.dumps(obj, cls=AnsibleJSONEncoder))
+        """
+        Dispatch event to Redis queue with automatic reconnection on failure.
+
+        Handles BrokenPipeError and ConnectionError by attempting reconnection.
+        If all reconnection attempts fail, logs the event loss but allows job to continue.
+
+        Args:
+            obj: Event data to dispatch (dict or serializable object)
+        """
+        max_attempts = self._max_reconnect_attempts + 1
+        last_error = None
+
+        # Extract diagnostic info from event
+        event_type = 'unknown'
+        job_id = 'unknown'
+        if isinstance(obj, dict):
+            event_type = obj.get('event', obj.get('event_name', 'unknown'))
+            job_id = obj.get('job_id', obj.get('unified_job_id', 'unknown'))
+
+        for attempt in range(max_attempts):
+            try:
+                # Attempt to push event to Redis
+                self.connection.rpush(self.queue, json.dumps(obj, cls=AnsibleJSONEncoder))
+
+                # Success - reset failure counter if this was a recovery
+                if self._connection_failures > 0:
+                    self.logger.info(f"Redis connection recovered after reconnection. " f"job_id={job_id}, event_type={event_type}")
+                    self._connection_failures = 0
+
+                return  # Successfully dispatched
+
+            except (BrokenPipeError, redis.exceptions.ConnectionError) as e:
+                last_error = e
+                error_type = type(e).__name__
+
+                self.logger.warning(f"Redis connection error during event dispatch " f"(attempt {attempt + 1}/{max_attempts}): {error_type}: {e}")
+
+                # Enhanced diagnostics
+                self.logger.warning(
+                    f"Failed event details: job_id={job_id}, event_type={event_type}, " f"queue={self.queue}, attempt={attempt + 1}/{max_attempts}"
+                )
+
+                if attempt < max_attempts - 1:
+                    # Try to reconnect before next attempt
+                    reconnected = self._reconnect()
+                    if reconnected:
+                        self.logger.info("Retrying event dispatch after successful reconnection")
+                    else:
+                        self.logger.warning(f"Reconnection failed, will retry dispatch anyway " f"(attempt {attempt + 2} coming)")
+                    # Continue to next attempt
+                    continue
+                else:
+                    # All attempts exhausted
+                    self._events_lost += 1
+                    self.logger.error(
+                        f"CRITICAL: Failed to dispatch event after {max_attempts} attempts. "
+                        f"Event will be lost. Total events lost this session: {self._events_lost}"
+                    )
+                    self.logger.error(
+                        f"DIAGNOSTIC INFO: "
+                        f"job_id={job_id}, "
+                        f"event_type={event_type}, "
+                        f"queue={self.queue}, "
+                        f"broker_url={self._broker_url}, "
+                        f"last_error={error_type}: {last_error}, "
+                        f"session_reconnections={self._total_reconnections}, "
+                        f"session_events_lost={self._events_lost}"
+                    )
+
+                    # IMPORTANT: Don't raise exception
+                    # Allow job to continue even though this event was lost
+                    # This prevents losing 17+ minutes of work due to event logging failure
+                    break
+
+            except Exception as e:
+                # Catch any other unexpected Redis errors
+                self.logger.error(f"Unexpected error dispatching event to Redis: {type(e).__name__}: {e}")
+                self.logger.error(f"Event context: job_id={job_id}, event_type={event_type}")
+                # Don't raise - allow job to continue
+                break
+
+    def health_check(self):
+        """
+        Check Redis connection health.
+
+        Returns:
+            bool: True if connection is healthy, False otherwise
+        """
+        try:
+            self.connection.ping()
+            return True
+        except Exception as e:
+            self.logger.warning(f"Redis health check failed: {type(e).__name__}: {e}")
+            return False
+
+    def get_connection_stats(self):
+        """
+        Get Redis connection statistics for monitoring.
+
+        Returns:
+            dict: Connection statistics
+        """
+        return {
+            'broker_url': self._broker_url,
+            'queue': self.queue,
+            'connected': self.health_check(),
+            'connection_failures': self._connection_failures,
+            'total_reconnections': self._total_reconnections,
+            'events_lost': self._events_lost,
+        }

awx/main/tasks/host_indirect.py

@@ -77,7 +77,14 @@ def build_indirect_host_data(job: Job, job_event_queries: dict[str, dict[str, st
         if jq_str_for_event not in compiled_jq_expressions:
             compiled_jq_expressions[resolved_action] = jq.compile(jq_str_for_event)
         compiled_jq = compiled_jq_expressions[resolved_action]
-        for data in compiled_jq.input(event.event_data['res']).all():
+
+        try:
+            data_source = compiled_jq.input(event.event_data['res']).all()
+        except Exception as e:
+            logger.warning(f'error for module {resolved_action} and data {event.event_data["res"]}: {e}')
+            continue
+
+        for data in data_source:
             # From this jq result (specific to a single Ansible module), get index information about this host record
             if not data.get('canonical_facts'):
                 if not facts_missing_logged:

awx/main/tests/data/sleep_task.py

@@ -0,0 +1,17 @@
+import time
+import logging
+
+from awx.main.dispatch import get_task_queuename
+from awx.main.dispatch.publish import task
+
+
+logger = logging.getLogger(__name__)
+
+
+@task(queue=get_task_queuename)
+def sleep_task(seconds=10, log=False):
+    if log:
+        logger.info('starting sleep_task')
+    time.sleep(seconds)
+    if log:
+        logger.info('finished sleep_task')

awx/main/tests/functional/analytics/test_core.py

@@ -87,8 +87,8 @@ def mock_analytic_post():
             {
                 'REDHAT_USERNAME': 'redhat_user',
                 'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
-                'SUBSCRIPTIONS_USERNAME': '',
-                'SUBSCRIPTIONS_PASSWORD': '',
+                'SUBSCRIPTIONS_CLIENT_ID': '',
+                'SUBSCRIPTIONS_CLIENT_SECRET': '',
             },
             True,
             ('redhat_user', 'redhat_pass'),
@@ -98,8 +98,8 @@ def mock_analytic_post():
             {
                 'REDHAT_USERNAME': None,
                 'REDHAT_PASSWORD': None,
-                'SUBSCRIPTIONS_USERNAME': 'subs_user',
-                'SUBSCRIPTIONS_PASSWORD': 'subs_pass',  # NOSONAR
+                'SUBSCRIPTIONS_CLIENT_ID': 'subs_user',
+                'SUBSCRIPTIONS_CLIENT_SECRET': 'subs_pass',  # NOSONAR
             },
             True,
             ('subs_user', 'subs_pass'),
@@ -109,8 +109,8 @@ def mock_analytic_post():
             {
                 'REDHAT_USERNAME': '',
                 'REDHAT_PASSWORD': '',
-                'SUBSCRIPTIONS_USERNAME': 'subs_user',
-                'SUBSCRIPTIONS_PASSWORD': 'subs_pass',  # NOSONAR
+                'SUBSCRIPTIONS_CLIENT_ID': 'subs_user',
+                'SUBSCRIPTIONS_CLIENT_SECRET': 'subs_pass',  # NOSONAR
             },
             True,
             ('subs_user', 'subs_pass'),
@@ -120,8 +120,8 @@ def mock_analytic_post():
             {
                 'REDHAT_USERNAME': '',
                 'REDHAT_PASSWORD': '',
-                'SUBSCRIPTIONS_USERNAME': '',
-                'SUBSCRIPTIONS_PASSWORD': '',
+                'SUBSCRIPTIONS_CLIENT_ID': '',
+                'SUBSCRIPTIONS_CLIENT_SECRET': '',
             },
             False,
             None,  # No request should be made
@@ -131,8 +131,8 @@ def mock_analytic_post():
             {
                 'REDHAT_USERNAME': '',
                 'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
-                'SUBSCRIPTIONS_USERNAME': 'subs_user',
-                'SUBSCRIPTIONS_PASSWORD': '',
+                'SUBSCRIPTIONS_CLIENT_ID': 'subs_user',
+                'SUBSCRIPTIONS_CLIENT_SECRET': '',
             },
             False,
             None,  # Invalid, no request should be made

awx/main/tests/functional/api/test_analytics.py

@@ -97,8 +97,8 @@ class TestAnalyticsGenericView:
                     'INSIGHTS_TRACKING_STATE': True,
                     'REDHAT_USERNAME': 'redhat_user',
                     'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
-                    'SUBSCRIPTIONS_USERNAME': '',
-                    'SUBSCRIPTIONS_PASSWORD': '',
+                    'SUBSCRIPTIONS_CLIENT_ID': '',
+                    'SUBSCRIPTIONS_CLIENT_SECRET': '',
                 },
                 ('redhat_user', 'redhat_pass'),
                 None,
@@ -109,8 +109,8 @@ class TestAnalyticsGenericView:
                     'INSIGHTS_TRACKING_STATE': True,
                     'REDHAT_USERNAME': '',
                     'REDHAT_PASSWORD': '',
-                    'SUBSCRIPTIONS_USERNAME': 'subs_user',
-                    'SUBSCRIPTIONS_PASSWORD': 'subs_pass',  # NOSONAR
+                    'SUBSCRIPTIONS_CLIENT_ID': 'subs_user',
+                    'SUBSCRIPTIONS_CLIENT_SECRET': 'subs_pass',  # NOSONAR
                 },
                 ('subs_user', 'subs_pass'),
                 None,
@@ -121,8 +121,8 @@ class TestAnalyticsGenericView:
                     'INSIGHTS_TRACKING_STATE': True,
                     'REDHAT_USERNAME': '',
                     'REDHAT_PASSWORD': '',
-                    'SUBSCRIPTIONS_USERNAME': '',
-                    'SUBSCRIPTIONS_PASSWORD': '',
+                    'SUBSCRIPTIONS_CLIENT_ID': '',
+                    'SUBSCRIPTIONS_CLIENT_SECRET': '',
                 },
                 None,
                 ERROR_MISSING_USER,
@@ -133,8 +133,8 @@ class TestAnalyticsGenericView:
                     'INSIGHTS_TRACKING_STATE': True,
                     'REDHAT_USERNAME': 'redhat_user',
                     'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
-                    'SUBSCRIPTIONS_USERNAME': 'subs_user',
-                    'SUBSCRIPTIONS_PASSWORD': 'subs_pass',  # NOSONAR
+                    'SUBSCRIPTIONS_CLIENT_ID': 'subs_user',
+                    'SUBSCRIPTIONS_CLIENT_SECRET': 'subs_pass',  # NOSONAR
                 },
                 ('redhat_user', 'redhat_pass'),
                 None,
@@ -145,8 +145,8 @@ class TestAnalyticsGenericView:
                     'INSIGHTS_TRACKING_STATE': True,
                     'REDHAT_USERNAME': '',
                     'REDHAT_PASSWORD': '',
-                    'SUBSCRIPTIONS_USERNAME': 'subs_user',  # NOSONAR
-                    'SUBSCRIPTIONS_PASSWORD': '',
+                    'SUBSCRIPTIONS_CLIENT_ID': 'subs_user',  # NOSONAR
+                    'SUBSCRIPTIONS_CLIENT_SECRET': '',
                 },
                 None,
                 ERROR_MISSING_PASSWORD,

awx/main/tests/functional/commands/test_callback_receiver.py

@@ -34,40 +34,18 @@ def test_wrapup_does_send_notifications(mocker):
     mock.assert_called_once_with('succeeded')
 
 
-class FakeRedis:
-    def keys(self, *args, **kwargs):
-        return []
-
-    def set(self):
-        pass
-
-    def get(self):
-        return None
-
-    @classmethod
-    def from_url(cls, *args, **kwargs):
-        return cls()
-
-    def pipeline(self):
-        return self
-
-
 class TestCallbackBrokerWorker(TransactionTestCase):
     @pytest.fixture(autouse=True)
-    def turn_off_websockets(self):
+    def turn_off_websockets_and_redis(self, fake_redis):
         with mock.patch('awx.main.dispatch.worker.callback.emit_event_detail', lambda *a, **kw: None):
             yield
 
-    def get_worker(self):
-        with mock.patch('redis.Redis', new=FakeRedis):  # turn off redis stuff
-            return CallbackBrokerWorker()
-
     def event_create_kwargs(self):
         inventory_update = InventoryUpdate.objects.create(source='file', inventory_source=InventorySource.objects.create(source='file'))
         return dict(inventory_update=inventory_update, created=inventory_update.created)
 
     def test_flush_with_valid_event(self):
-        worker = self.get_worker()
+        worker = CallbackBrokerWorker()
         events = [InventoryUpdateEvent(uuid=str(uuid4()), **self.event_create_kwargs())]
         worker.buff = {InventoryUpdateEvent: events}
         worker.flush()
@@ -75,7 +53,7 @@ class TestCallbackBrokerWorker(TransactionTestCase):
         assert InventoryUpdateEvent.objects.filter(uuid=events[0].uuid).count() == 1
 
     def test_flush_with_invalid_event(self):
-        worker = self.get_worker()
+        worker = CallbackBrokerWorker()
         kwargs = self.event_create_kwargs()
         events = [
             InventoryUpdateEvent(uuid=str(uuid4()), stdout='good1', **kwargs),
@@ -90,7 +68,7 @@ class TestCallbackBrokerWorker(TransactionTestCase):
         assert worker.buff == {InventoryUpdateEvent: [events[1]]}
 
     def test_duplicate_key_not_saved_twice(self):
-        worker = self.get_worker()
+        worker = CallbackBrokerWorker()
         events = [InventoryUpdateEvent(uuid=str(uuid4()), **self.event_create_kwargs())]
         worker.buff = {InventoryUpdateEvent: events.copy()}
         worker.flush()
@@ -104,7 +82,7 @@ class TestCallbackBrokerWorker(TransactionTestCase):
         assert worker.buff.get(InventoryUpdateEvent, []) == []
 
     def test_give_up_on_bad_event(self):
-        worker = self.get_worker()
+        worker = CallbackBrokerWorker()
         events = [InventoryUpdateEvent(uuid=str(uuid4()), counter=-2, **self.event_create_kwargs())]
         worker.buff = {InventoryUpdateEvent: events.copy()}
 
@@ -117,7 +95,7 @@ class TestCallbackBrokerWorker(TransactionTestCase):
         assert InventoryUpdateEvent.objects.filter(uuid=events[0].uuid).count() == 0  # sanity
 
     def test_flush_with_empty_buffer(self):
-        worker = self.get_worker()
+        worker = CallbackBrokerWorker()
         worker.buff = {InventoryUpdateEvent: []}
         with mock.patch.object(InventoryUpdateEvent.objects, 'bulk_create') as flush_mock:
             worker.flush()
@@ -127,7 +105,7 @@ class TestCallbackBrokerWorker(TransactionTestCase):
         # In postgres, text fields reject NUL character, 0x00
         # tests use sqlite3 which will not raise an error
         # but we can still test that it is sanitized before saving
-        worker = self.get_worker()
+        worker = CallbackBrokerWorker()
         kwargs = self.event_create_kwargs()
         events = [InventoryUpdateEvent(uuid=str(uuid4()), stdout="\x00", **kwargs)]
         assert "\x00" in events[0].stdout  # sanity

awx/main/tests/functional/conftest.py

@@ -63,6 +63,33 @@ def swagger_autogen(requests=__SWAGGER_REQUESTS__):
     return requests
 
 
+class FakeRedis:
+    def keys(self, *args, **kwargs):
+        return []
+
+    def set(self):
+        pass
+
+    def get(self):
+        return None
+
+    @classmethod
+    def from_url(cls, *args, **kwargs):
+        return cls()
+
+    def pipeline(self):
+        return self
+
+    def ping(self):
+        return
+
+
+@pytest.fixture
+def fake_redis():
+    with mock.patch('redis.Redis', new=FakeRedis):  # turn off redis stuff
+        yield
+
+
 @pytest.fixture
 def user():
     def u(name, is_superuser=False):

awx/main/tests/functional/models/test_ha.py

@@ -3,6 +3,10 @@ import pytest
 # AWX
 from awx.main.ha import is_ha_environment
 from awx.main.models.ha import Instance
+from awx.main.dispatch.pool import get_auto_max_workers
+
+# Django
+from django.test.utils import override_settings
 
 
 @pytest.mark.django_db
@@ -17,3 +21,25 @@ def test_db_localhost():
     Instance.objects.create(hostname='foo', node_type='hybrid')
     Instance.objects.create(hostname='bar', node_type='execution')
     assert is_ha_environment() is False
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    'settings',
+    [
+        dict(SYSTEM_TASK_ABS_MEM='16Gi', SYSTEM_TASK_ABS_CPU='24', SYSTEM_TASK_FORKS_MEM=400, SYSTEM_TASK_FORKS_CPU=4),
+        dict(SYSTEM_TASK_ABS_MEM='124Gi', SYSTEM_TASK_ABS_CPU='2', SYSTEM_TASK_FORKS_MEM=None, SYSTEM_TASK_FORKS_CPU=None),
+    ],
+    ids=['cpu_dominated', 'memory_dominated'],
+)
+def test_dispatcher_max_workers_reserve(settings, fake_redis):
+    """This tests that the dispatcher max_workers matches instance capacity
+
+    Assumes capacity_adjustment is 1,
+    plus reserve worker count
+    """
+    with override_settings(**settings):
+        i = Instance.objects.create(hostname='test-1', node_type='hybrid')
+        i.local_health_check()
+
+        assert get_auto_max_workers() == i.capacity + 7, (i.cpu, i.memory, i.cpu_capacity, i.mem_capacity)

awx/main/tests/unit/utils/test_licensing.py

@@ -0,0 +1,37 @@
+import json
+from http import HTTPStatus
+from unittest.mock import patch
+
+from requests import Response
+
+from awx.main.utils.licensing import Licenser
+
+
+def test_rhsm_licensing():
+    def mocked_requests_get(*args, **kwargs):
+        assert kwargs['verify'] == True
+        response = Response()
+        subs = json.dumps({'body': []})
+        response.status_code = HTTPStatus.OK
+        response._content = bytes(subs, 'utf-8')
+        return response
+
+    licenser = Licenser()
+    with patch('awx.main.utils.analytics_proxy.OIDCClient.make_request', new=mocked_requests_get):
+        subs = licenser.get_rhsm_subs('localhost', 'admin', 'admin')
+        assert subs == []
+
+
+def test_satellite_licensing():
+    def mocked_requests_get(*args, **kwargs):
+        assert kwargs['verify'] == True
+        response = Response()
+        subs = json.dumps({'results': []})
+        response.status_code = HTTPStatus.OK
+        response._content = bytes(subs, 'utf-8')
+        return response
+
+    licenser = Licenser()
+    with patch('requests.get', new=mocked_requests_get):
+        subs = licenser.get_satellite_subs('localhost', 'admin', 'admin')
+        assert subs == []

awx/main/utils/analytics_proxy.py

@@ -23,7 +23,7 @@ class TokenError(requests.RequestException):
     try:
       client = OIDCClient(...)
       client.make_request(...)
-    except TokenGenerationError as e:
+    except TokenError as e:
         print(f"Token generation failed due to {e.__cause__}")
     except requests.RequestException:
         print("API request failed)
@@ -102,13 +102,15 @@ class OIDCClient:
         self,
         client_id: str,
         client_secret: str,
-        token_url: str,
-        scopes: list[str],
+        token_url: str = DEFAULT_OIDC_TOKEN_ENDPOINT,
+        scopes: list[str] = None,
         base_url: str = '',
     ) -> None:
         self.client_id: str = client_id
         self.client_secret: str = client_secret
         self.token_url: str = token_url
+        if scopes is None:
+            scopes = ['api.console']
         self.scopes = scopes
         self.base_url: str = base_url
         self.token: Optional[Token] = None

awx/main/utils/licensing.py

@@ -38,6 +38,7 @@ from django.utils.translation import gettext_lazy as _
 from awx_plugins.interfaces._temporary_private_licensing_api import detect_server_product_name
 
 from awx.main.constants import SUBSCRIPTION_USAGE_MODEL_UNIQUE_HOSTS
+from awx.main.utils.analytics_proxy import OIDCClient
 
 MAX_INSTANCES = 9999999
 
@@ -228,37 +229,38 @@ class Licenser(object):
             host = getattr(settings, 'REDHAT_CANDLEPIN_HOST', None)
 
         if not user:
-            raise ValueError('subscriptions_username is required')
+            raise ValueError('subscriptions_client_id is required')
 
         if not pw:
-            raise ValueError('subscriptions_password is required')
+            raise ValueError('subscriptions_client_secret is required')
 
         if host and user and pw:
             if 'subscription.rhsm.redhat.com' in host:
-                json = self.get_rhsm_subs(host, user, pw)
+                json = self.get_rhsm_subs(settings.SUBSCRIPTIONS_RHSM_URL, user, pw)
             else:
                 json = self.get_satellite_subs(host, user, pw)
             return self.generate_license_options_from_entitlements(json)
         return []
 
-    def get_rhsm_subs(self, host, user, pw):
-        verify = getattr(settings, 'REDHAT_CANDLEPIN_VERIFY', True)
-        json = []
-        try:
-            subs = requests.get('/'.join([host, 'subscription/users/{}/owners'.format(user)]), verify=verify, auth=(user, pw))
-        except requests.exceptions.ConnectionError as error:
-            raise error
-        except OSError as error:
-            raise OSError(
-                'Unable to open certificate bundle {}. Check that the service is running on Red Hat Enterprise Linux.'.format(verify)
-            ) from error  # noqa
-        subs.raise_for_status()
+    def get_rhsm_subs(self, host, client_id, client_secret):
+        client = OIDCClient(client_id, client_secret)
+        subs = client.make_request(
+            'GET',
+            host,
+            verify=True,
+            timeout=(31, 31),
+        )
 
-        for sub in subs.json():
-            resp = requests.get('/'.join([host, 'subscription/owners/{}/pools/?match=*tower*'.format(sub['key'])]), verify=verify, auth=(user, pw))
-            resp.raise_for_status()
-            json.extend(resp.json())
-        return json
+        subs.raise_for_status()
+        subs_formatted = []
+        for sku in subs.json()['body']:
+            sku_data = {k: v for k, v in sku.items() if k != 'subscriptions'}
+            for sub in sku['subscriptions']:
+                sub_data = sku_data.copy()
+                sub_data['subscriptions'] = sub
+                subs_formatted.append(sub_data)
+
+        return subs_formatted
 
     def get_satellite_subs(self, host, user, pw):
         port = None
@@ -267,7 +269,7 @@ class Licenser(object):
             port = str(self.config.get("server", "port"))
         except Exception as e:
             logger.exception('Unable to read rhsm config to get ca_cert location. {}'.format(str(e)))
-            verify = getattr(settings, 'REDHAT_CANDLEPIN_VERIFY', True)
+            verify = True
         if port:
             host = ':'.join([host, port])
         json = []
@@ -314,20 +316,11 @@ class Licenser(object):
             return False
         return True
 
-    def is_appropriate_sub(self, sub):
-        if sub['activeSubscription'] is False:
-            return False
-        # Products that contain Ansible Tower
-        products = sub.get('providedProducts', [])
-        if any(product.get('productId') == '480' for product in products):
-            return True
-        return False
-
     def generate_license_options_from_entitlements(self, json):
         from dateutil.parser import parse
 
         ValidSub = collections.namedtuple(
-            'ValidSub', 'sku name support_level end_date trial developer_license quantity pool_id satellite subscription_id account_number usage'
+            'ValidSub', 'sku name support_level end_date trial developer_license quantity satellite subscription_id account_number usage'
         )
         valid_subs = []
         for sub in json:
@@ -335,10 +328,14 @@ class Licenser(object):
             if satellite:
                 is_valid = self.is_appropriate_sat_sub(sub)
             else:
-                is_valid = self.is_appropriate_sub(sub)
+                # the list of subs from console.redhat.com are already valid based on the query params we provided
+                is_valid = True
             if is_valid:
                 try:
-                    end_date = parse(sub.get('endDate'))
+                    if satellite:
+                        end_date = parse(sub.get('endDate'))
+                    else:
+                        end_date = parse(sub['subscriptions']['endDate'])
                 except Exception:
                     continue
                 now = datetime.utcnow()
@@ -346,44 +343,50 @@ class Licenser(object):
                 if end_date < now:
                     # If the sub has a past end date, skip it
                     continue
-                try:
-                    quantity = int(sub['quantity'])
-                    if quantity == -1:
-                        # effectively, unlimited
-                        quantity = MAX_INSTANCES
-                except Exception:
-                    continue
 
-                sku = sub['productId']
-                trial = sku.startswith('S')  # i.e.,, SER/SVC
                 developer_license = False
                 support_level = ''
-                usage = ''
-                pool_id = sub['id']
-                subscription_id = sub['subscriptionId']
-                account_number = sub['accountNumber']
+                account_number = ''
+                usage = sub.get('usage', '')
                 if satellite:
+                    try:
+                        quantity = int(sub['quantity'])
+                    except Exception:
+                        continue
+                    sku = sub['productId']
+                    subscription_id = sub['subscriptionId']
+                    sub_name = sub['productName']
                     support_level = sub['support_level']
-                    usage = sub['usage']
+                    account_number = sub['accountNumber']
                 else:
-                    for attr in sub.get('productAttributes', []):
-                        if attr.get('name') == 'support_level':
-                            support_level = attr.get('value')
-                        elif attr.get('name') == 'usage':
-                            usage = attr.get('value')
-                        elif attr.get('name') == 'ph_product_name' and attr.get('value') == 'RHEL Developer':
-                            developer_license = True
+                    try:
+                        if sub['capacity']['name'] == "Nodes":
+                            quantity = int(sub['capacity']['quantity']) * int(sub['subscriptions']['quantity'])
+                        else:
+                            continue
+                    except Exception:
+                        continue
+                    sku = sub['sku']
+                    sub_name = sub['name']
+                    support_level = sub['serviceLevel']
+                    subscription_id = sub['subscriptions']['number']
+                    if sub.get('name') == 'RHEL Developer':
+                        developer_license = True
+
+                if quantity == -1:
+                    # effectively, unlimited
+                    quantity = MAX_INSTANCES
+                trial = sku.startswith('S')  # i.e.,, SER/SVC
 
                 valid_subs.append(
                     ValidSub(
                         sku,
-                        sub['productName'],
+                        sub_name,
                         support_level,
                         end_date,
                         trial,
                         developer_license,
                         quantity,
-                        pool_id,
                         satellite,
                         subscription_id,
                         account_number,
@@ -414,7 +417,6 @@ class Licenser(object):
                 license._attrs['satellite'] = satellite
                 license._attrs['valid_key'] = True
                 license.update(license_date=int(sub.end_date.strftime('%s')))
-                license.update(pool_id=sub.pool_id)
                 license.update(subscription_id=sub.subscription_id)
                 license.update(account_number=sub.account_number)
                 licenses.append(license._attrs.copy())

awx/settings/defaults.py

@@ -964,6 +964,9 @@ CLUSTER_HOST_ID = socket.gethostname()
 # - 'unique_managed_hosts': Compliant = automated - deleted hosts (using /api/v2/host_metrics/)
 SUBSCRIPTION_USAGE_MODEL = ''
 
+# Default URL and query params for obtaining valid AAP subscriptions
+SUBSCRIPTIONS_RHSM_URL = 'https://console.redhat.com/api/rhsm/v2/products?include=providedProducts&oids=480&status=Active'
+
 # Host metrics cleanup - last time of the task/command run
 CLEANUP_HOST_METRICS_LAST_TS = None
 # Host metrics cleanup - minimal interval between two cleanups in days

awx_collection/plugins/modules/license.py

@@ -31,9 +31,9 @@ options:
           unlicensed or trial licensed.  When force=true, the license is always applied.
       type: bool
       default: 'False'
-    pool_id:
+    subscription_id:
       description:
-        - Red Hat or Red Hat Satellite pool_id to attach to
+        - Red Hat or Red Hat Satellite subscription_id to attach to
       required: False
       type: str
     state:
@@ -57,9 +57,9 @@ EXAMPLES = '''
     username: "my_satellite_username"
     password: "my_satellite_password"
 
-- name: Attach to a pool (requires fetching subscriptions at least once before)
+- name: Attach to a subscription (requires fetching subscriptions at least once before)
   license:
-    pool_id: 123456
+    subscription_id: 123456
 
 - name: Remove license
   license:
@@ -75,14 +75,14 @@ def main():
     module = ControllerAPIModule(
         argument_spec=dict(
             manifest=dict(type='str', required=False),
-            pool_id=dict(type='str', required=False),
+            subscription_id=dict(type='str', required=False),
             force=dict(type='bool', default=False),
             state=dict(choices=['present', 'absent'], default='present'),
         ),
         required_if=[
-            ['state', 'present', ['manifest', 'pool_id'], True],
+            ['state', 'present', ['manifest', 'subscription_id'], True],
         ],
-        mutually_exclusive=[("manifest", "pool_id")],
+        mutually_exclusive=[("manifest", "subscription_id")],
     )
 
     json_output = {'changed': False}
@@ -124,7 +124,7 @@ def main():
         if module.params.get('manifest', None):
             module.post_endpoint('config', data={'manifest': manifest.decode()})
         else:
-            module.post_endpoint('config/attach', data={'pool_id': module.params.get('pool_id')})
+            module.post_endpoint('config/attach', data={'subscription_id': module.params.get('subscription_id')})
 
     module.exit_json(**json_output)
 

awx_collection/plugins/modules/subscriptions.py

@@ -20,15 +20,15 @@ description:
     - Get subscriptions available to Automation Platform Controller. See
       U(https://www.ansible.com/tower) for an overview.
 options:
-    username:
+    client_id:
       description:
-        - Red Hat or Red Hat Satellite username to get available subscriptions.
+        - Red Hat service account client ID or Red Hat Satellite username to get available subscriptions.
         - The credentials you use will be stored for future use in retrieving renewal or expanded subscriptions
       required: True
       type: str
-    password:
+    client_secret:
       description:
-        - Red Hat or Red Hat Satellite password to get available subscriptions.
+        - Red Hat service account client secret or Red Hat Satellite password to get available subscriptions.
         - The credentials you use will be stored for future use in retrieving renewal or expanded subscriptions
       required: True
       type: str
@@ -53,13 +53,13 @@ subscriptions:
 EXAMPLES = '''
 - name: Get subscriptions
   subscriptions:
-    username: "my_username"
-    password: "My Password"
+    client_id: "c6bd7594-d776-46e5-8156-6d17af147479"
+    client_secret: "MO9QUvoOZ5fc5JQKXoTch1AsTLI7nFsZ"
 
 - name: Get subscriptions with a filter
   subscriptions:
-    username: "my_username"
-    password: "My Password"
+    client_id: "c6bd7594-d776-46e5-8156-6d17af147479"
+    client_secret: "MO9QUvoOZ5fc5JQKXoTch1AsTLI7nFsZ"
     filters:
       product_name: "Red Hat Ansible Automation Platform"
       support_level: "Self-Support"
@@ -72,8 +72,8 @@ def main():
 
     module = ControllerAPIModule(
         argument_spec=dict(
-            username=dict(type='str', required=True),
-            password=dict(type='str', no_log=True, required=True),
+            client_id=dict(type='str', required=True),
+            client_secret=dict(type='str', no_log=True, required=True),
             filters=dict(type='dict', required=False, default={}),
         ),
     )
@@ -82,8 +82,8 @@ def main():
 
     # Check if Tower is already licensed
     post_data = {
-        'subscriptions_password': module.params.get('password'),
-        'subscriptions_username': module.params.get('username'),
+        'subscriptions_client_secret': module.params.get('client_secret'),
+        'subscriptions_client_id': module.params.get('client_id'),
     }
     all_subscriptions = module.post_endpoint('config/subscriptions', data=post_data)['json']
     json_output['subscriptions'] = []

tools/scripts/firehose_tasks.py

@@ -0,0 +1,16 @@
+#!/usr/bin/env python
+
+from django import setup
+
+from awx import prepare_env
+
+prepare_env()
+
+setup()
+
+# Keeping this in test folder allows it to be importable
+from awx.main.tests.data.sleep_task import sleep_task
+
+
+for i in range(634):
+    sleep_task.delay()