Optimize HostManager.active_count() with cache (#16505)

* [AAP-78392] Optimize HostManager.active_count() with cache and functional index

active_count() runs a full sequential scan with LOWER()+DISTINCT on main_host for every license check. At customer scale this consumed 74.5 minutes of DB time over 4 hours (47K calls at 93ms avg).

Add a 60-second Redis-backed cache via the existing memoize decorator to reduce call volume by ~99.5%. Add a functional btree index on LOWER(name) to eliminate the sequential scan for the remaining calls.

* Use AddIndexConcurrently instead of AddIndex in the migration for host name lower index

* Revert AddIndexConcurrently to AddIndex for CI compatibility

The api-migrations CI job runs against SQLite which does not support PostgreSQL-specific AddIndexConcurrently. Standard AddIndex works across all backends and the brief write lock during production upgrades is acceptable for this table size.

* Remove functional index, keep cache-only fix per reviewer feedback

Drop the LOWER(name) functional index and migration to minimize
the change footprint.

----
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

awx/main/managers.py

@@ -10,6 +10,7 @@ from django.db.models.functions import Lower
 
 from ansible_base.lib.utils.db import advisory_lock
 
+from awx.main.utils.common import memoize
 from awx.main.utils.filters import SmartFilter
 from awx.main.constants import RECEPTOR_PENDING
 
@@ -85,6 +86,7 @@ class HostLatestSummaryQuerySet(models.QuerySet):
 class HostManager(models.Manager.from_queryset(HostLatestSummaryQuerySet)):
     """Custom manager class for Hosts model."""
 
+    @memoize(ttl=60, cache_key='host_active_count')
     def active_count(self):
         """Return count of active, unique hosts for licensing.
         Construction of query involves:

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

@@ -92,6 +92,22 @@ class TestInventoryScript:
 
 @pytest.mark.django_db
 class TestActiveCount:
+    @pytest.fixture(autouse=True)
+    def _bypass_active_count_cache(self):
+        from django.core.cache import cache
+
+        cache.delete('host_active_count')
+        original_set = cache.set
+
+        def skip_host_cache(key, *args, **kwargs):
+            if key == 'host_active_count':
+                return
+            return original_set(key, *args, **kwargs)
+
+        with mock.patch.object(cache, 'set', side_effect=skip_host_cache):
+            yield
+        cache.delete('host_active_count')
+
     def test_host_active_count(self, organization):
         inv1 = Inventory.objects.create(name='inv1', organization=organization)
         inv2 = Inventory.objects.create(name='inv2', organization=organization)
@@ -141,6 +157,41 @@ class TestActiveCount:
         assert Host.objects.active_count() == 2
 
 
+@pytest.mark.django_db
+class TestActiveCountCache:
+    @pytest.fixture(autouse=True)
+    def _clear_active_count_cache(self):
+        from awx.main.utils.common import memoize_delete
+
+        memoize_delete('host_active_count')
+        yield
+        memoize_delete('host_active_count')
+
+    def test_active_count_cache(self, organization):
+        inv = Inventory.objects.create(name='inv1', organization=organization)
+        inv.hosts.create(name='host1')
+        assert Host.objects.active_count() == 1
+
+        inv.hosts.create(name='host2')
+        assert Host.objects.active_count() == 1  # still cached
+
+        from awx.main.utils.common import memoize_delete
+
+        memoize_delete('host_active_count')
+        assert Host.objects.active_count() == 2
+
+    def test_active_count_cache_after_delete(self, organization):
+        inv = Inventory.objects.create(name='inv1', organization=organization)
+        h = inv.hosts.create(name='host1')
+        assert Host.objects.active_count() == 1
+
+        h.delete()
+        from awx.main.utils.common import memoize_delete
+
+        memoize_delete('host_active_count')
+        assert Host.objects.active_count() == 0
+
+
 @pytest.mark.django_db
 class TestSCMUpdateFeatures:
     def test_source_location(self, scm_inventory_source):