Move plugin loading to dispatcher startup

awx/api/serializers.py

@@ -117,7 +117,7 @@
 from awx.main.utils.named_url_graph import reset_counters
 from awx.main.scheduler.task_manager_models import TaskManagerModels
 from awx.main.redact import UriCleaner, REPLACE_STR
-from awx.main.signals import update_inventory_computed_fields
+from awx.main.tasks.system import update_inventory_computed_fields
 
 
 from awx.main.validators import vars_validate_or_raise

awx/main/apps.py

@@ -1,14 +1,8 @@
-import os
-
 from django.apps import AppConfig
 from django.utils.translation import gettext_lazy as _
-from awx.main.utils.common import bypass_in_test, load_all_entry_points_for
-from awx.main.utils.migration import is_database_synchronized
 from awx.main.utils.named_url_graph import _customize_graph, generate_graph
 from awx.conf import register, fields
 
-from awx_plugins.interfaces._temporary_private_licensing_api import detect_server_product_name
-
 
 class MainConfig(AppConfig):
     name = 'awx.main'
@@ -40,48 +34,6 @@ def load_named_url_feature(self):
             category_slug='named-url',
         )
 
-    def _load_credential_types_feature(self):
-        """
-        Create CredentialType records for any discovered credentials.
-
-        Note that Django docs advise _against_ interacting with the database using
-        the ORM models in the ready() path. Specifically, during testing.
-        However, we explicitly use the @bypass_in_test decorator to avoid calling this
-        method during testing.
-
-        Django also advises against running pattern because it runs everywhere i.e.
-        every management command. We use an advisory lock to ensure correctness and
-        we will deal performance if it becomes an issue.
-        """
-        from awx.main.models.credential import CredentialType
-
-        if is_database_synchronized():
-            CredentialType.setup_tower_managed_defaults(app_config=self)
-
-    @bypass_in_test
-    def load_credential_types_feature(self):
-        return self._load_credential_types_feature()
-
-    def load_inventory_plugins(self):
-        from awx.main.models.inventory import InventorySourceOptions
-
-        is_awx = detect_server_product_name() == 'AWX'
-        extra_entry_point_groups = () if is_awx else ('inventory.supported',)
-        entry_points = load_all_entry_points_for(['inventory', *extra_entry_point_groups])
-
-        for entry_point_name, entry_point in entry_points.items():
-            cls = entry_point.load()
-            InventorySourceOptions.injectors[entry_point_name] = cls
-
     def ready(self):
         super().ready()
-
-        """
-        Credential loading triggers database operations. There are cases we want to call
-        awx-manage collectstatic without a database. All management commands invoke the ready() code
-        path. Using settings.AWX_SKIP_CREDENTIAL_TYPES_DISCOVER _could_ invoke a database operation.
-        """
-        if not os.environ.get('AWX_SKIP_CREDENTIAL_TYPES_DISCOVER', None):
-            self.load_credential_types_feature()
         self.load_named_url_feature()
-        self.load_inventory_plugins()

awx/main/signals.py

@@ -55,7 +55,6 @@
 )
 from awx.main.utils import model_instance_diff, model_to_dict, camelcase_to_underscore, get_current_apps
 from awx.main.utils import ignore_inventory_computed_fields, ignore_inventory_group_removal, _inventory_updates
-from awx.main.tasks.system import update_inventory_computed_fields, handle_removed_image
 from awx.main.fields import (
     is_implicit_parent,
     update_role_parentage_for_instance,
@@ -104,6 +103,8 @@ def emit_update_inventory_on_created_or_deleted(sender, **kwargs):
         pass
     else:
         if inventory is not None:
+            from awx.main.tasks.system import update_inventory_computed_fields
+
             connection.on_commit(lambda: update_inventory_computed_fields.delay(inventory.id))
 
 
@@ -623,6 +624,8 @@ def deny_orphaned_approvals(sender, instance, **kwargs):
 def _handle_image_cleanup(removed_image, pk):
     if (not removed_image) or ExecutionEnvironment.objects.filter(image=removed_image).exclude(pk=pk).exists():
         return  # if other EE objects reference the tag, then do not purge it
+    from awx.main.tasks.system import handle_removed_image
+
     handle_removed_image.delay(remove_images=[removed_image])
 
 

awx/main/tasks/system.py

@@ -16,6 +16,7 @@
 
 # Django
 from django.conf import settings
+from django.apps import apps
 from django.db import connection, transaction, DatabaseError, IntegrityError
 from django.db.models.fields.related import ForeignKey
 from django.utils.timezone import now, timedelta
@@ -53,12 +54,14 @@
     SmartInventoryMembership,
     Job,
     convert_jsonfields,
+    CredentialType,
 )
+from awx.main.models.inventory import InventorySourceOptions
 from awx.main.constants import ACTIVE_STATES, ERROR_STATES
 from awx.main.dispatch.publish import task
 from awx.main.dispatch import get_task_queuename, reaper
-from awx.main.utils.common import ignore_inventory_computed_fields, ignore_inventory_group_removal
-
+from awx.main.utils.common import ignore_inventory_computed_fields, ignore_inventory_group_removal, load_all_entry_points_for
+from awx.main.utils.migration import is_database_synchronized
 from awx.main.utils.reload import stop_local_services
 from awx.main.utils.pglock import advisory_lock
 from awx.main.tasks.helpers import is_run_threshold_reached
@@ -70,6 +73,8 @@
 
 from rest_framework.exceptions import PermissionDenied
 
+from awx_plugins.interfaces._temporary_private_licensing_api import detect_server_product_name
+
 logger = logging.getLogger('awx.main.tasks.system')
 
 OPENSSH_KEY_ERROR = u'''\
@@ -87,6 +92,8 @@ def dispatch_startup():
         write_receptor_config()
 
     try:
+        load_inventory_plugins()
+        load_credential_types_feature()
         convert_jsonfields()
     except Exception:
         logger.exception("Failed json field conversion, skipping.")
@@ -134,6 +141,36 @@ def inform_cluster_of_shutdown():
         logger.exception('Encountered problem with normal shutdown signal.')
 
 
+def load_inventory_plugins():
+    with advisory_lock('load_inventory_plugins', wait=False) as acquired:
+        if not acquired:
+            return
+
+        is_awx = detect_server_product_name() == 'AWX'
+        extra_entry_point_groups = () if is_awx else ('inventory.supported',)
+        entry_points = load_all_entry_points_for(['inventory', *extra_entry_point_groups])
+
+        for entry_point_name, entry_point in entry_points.items():
+            cls = entry_point.load()
+            InventorySourceOptions.injectors[entry_point_name] = cls
+
+
+def load_credential_types_feature():
+    """
+    Create CredentialType records for any discovered credentials.
+    """
+    if os.environ.get('AWX_SKIP_CREDENTIAL_TYPES_DISCOVER', None):
+        logger.info('Credential type loading disabled, skipping')
+        return
+
+    with advisory_lock('load_inventory_plugins', wait=False) as acquired:
+        if not acquired:
+            return
+
+        if is_database_synchronized():
+            CredentialType.setup_tower_managed_defaults(app_config=apps.get_app_config('main'))
+
+
 @task(queue=get_task_queuename)
 def migrate_jsonfield(table, pkfield, columns):
     batchsize = 10000

awx/main/tests/functional/conftest.py

@@ -45,6 +45,7 @@
 from awx.main.models.workflow import WorkflowJobTemplate
 from awx.main.models.ad_hoc_commands import AdHocCommand
 from awx.main.models.execution_environments import ExecutionEnvironment
+from awx.main.tasks.system import load_credential_types_feature, load_inventory_plugins
 from awx.main.utils import is_testing
 
 __SWAGGER_REQUESTS__ = {}
@@ -895,3 +896,9 @@ def control_plane_execution_environment():
 @pytest.fixture
 def default_job_execution_environment():
     return ExecutionEnvironment.objects.create(name="Default Job EE", managed=False)
+
+
+@pytest.fixture
+def setup_awx_plugins():
+    load_credential_types_feature()
+    load_inventory_plugins()

awx/main/tests/functional/migrations/test_rbac_migration.py

@@ -5,7 +5,7 @@
 
 
 @pytest.mark.django_db
-def test_implied_organization_subquery_inventory():
+def test_implied_organization_subquery_inventory(setup_awx_plugins):
     orgs = []
     for i in range(3):
         orgs.append(Organization.objects.create(name='foo{}'.format(i)))

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

@@ -1,7 +1,8 @@
 import pytest
 
 # AWX context managers for testing
-from awx.main.signals import disable_activity_stream, disable_computed_fields, update_inventory_computed_fields
+from awx.main.signals import disable_activity_stream, disable_computed_fields
+from awx.main.tasks.system import update_inventory_computed_fields
 
 # AWX models
 from awx.main.models.organization import Organization

awx/main/tests/functional/test_inventory_source_injectors.py

@@ -57,7 +57,11 @@ def credential_kind(source):
 def fake_credential_factory():
     def wrap(source):
         ct = CredentialType.defaults[credential_kind(source)]()
-        ct.save()
+        existing_ct = CredentialType.objects.filter(name=ct.name).first()
+        if existing_ct:
+            ct = existing_ct
+        else:
+            ct.save()
 
         inputs = {}
         var_specs = {}  # pivoted version of inputs
@@ -69,7 +73,7 @@ def wrap(source):
         if source == 'controller':
             inputs.pop('oauth_token')  # mutually exclusive with user/pass
 
-        return Credential.objects.create(credential_type=ct, inputs=inputs)
+        ct = Credential.objects.create(credential_type=ct, inputs=inputs)
 
     return wrap
 
@@ -194,82 +198,82 @@ def create_reference_data(source_dir, env, content):
 
 @mock.patch('awx_plugins.interfaces._temporary_private_licensing_api.detect_server_product_name', return_value='NOT-AWX')
 @pytest.mark.django_db
-@pytest.mark.parametrize('this_kind', discover_available_cloud_provider_plugin_names())
-def test_inventory_update_injected_content(product_name, this_kind, inventory, fake_credential_factory, mock_me):
-    ExecutionEnvironment.objects.create(name='Control Plane EE', managed=True)
-    ExecutionEnvironment.objects.create(name='Default Job EE', managed=False)
-
-    injector = InventorySource.injectors[this_kind]
-
-    src_vars = dict(base_source_var='value_of_var')
-    src_vars['plugin'] = injector.get_proper_name()
-    inventory_source = InventorySource.objects.create(
-        inventory=inventory,
-        source=this_kind,
-        source_vars=src_vars,
-    )
-    inventory_source.credentials.add(fake_credential_factory(this_kind))
-    inventory_update = inventory_source.create_unified_job()
-    task = RunInventoryUpdate()
-
-    def substitute_run(awx_receptor_job):
-        """This method will replace run_pexpect
-        instead of running, it will read the private data directory contents
-        It will make assertions that the contents are correct
-        If MAKE_INVENTORY_REFERENCE_FILES is set, it will produce reference files
-        """
-        envvars = awx_receptor_job.runner_params['envvars']
-
-        private_data_dir = envvars.pop('AWX_PRIVATE_DATA_DIR')
-        assert envvars.pop('ANSIBLE_INVENTORY_ENABLED') == 'auto'
-        set_files = bool(os.getenv("MAKE_INVENTORY_REFERENCE_FILES", 'false').lower()[0] not in ['f', '0'])
-        env, content = read_content(private_data_dir, envvars, inventory_update)
-
-        # Assert inventory plugin inventory file is in private_data_dir
-        inventory_filename = InventorySource.injectors[inventory_update.source]().filename
-        assert (
-            len([True for k in content.keys() if k.endswith(inventory_filename)]) > 0
-        ), f"'{inventory_filename}' file not found in inventory update runtime files {content.keys()}"
-
-        env.pop('ANSIBLE_COLLECTIONS_PATHS', None)  # collection paths not relevant to this test
-        base_dir = os.path.join(DATA, 'plugins')
-        if not os.path.exists(base_dir):
-            os.mkdir(base_dir)
-        source_dir = os.path.join(base_dir, this_kind)  # this_kind is a global
-        if set_files:
-            create_reference_data(source_dir, env, content)
-            pytest.skip('You set MAKE_INVENTORY_REFERENCE_FILES, so this created files, unset to run actual test.')
-        else:
+def test_inventory_update_injected_content(product_name, inventory, fake_credential_factory, mock_me, setup_awx_plugins):
+    for this_kind in discover_available_cloud_provider_plugin_names():
+        ExecutionEnvironment.objects.create(name='Control Plane EE', managed=True)
+        ExecutionEnvironment.objects.create(name='Default Job EE', managed=False)
+
+        injector = InventorySource.injectors[this_kind]
+
+        src_vars = dict(base_source_var='value_of_var')
+        src_vars['plugin'] = injector.get_proper_name()
+        inventory_source = InventorySource.objects.create(
+            inventory=inventory,
+            source=this_kind,
+            source_vars=src_vars,
+        )
+        inventory_source.credentials.add(fake_credential_factory(this_kind.replace('-', '_')))
+        inventory_update = inventory_source.create_unified_job()
+        task = RunInventoryUpdate()
+
+        def substitute_run(awx_receptor_job):
+            """This method will replace run_pexpect
+            instead of running, it will read the private data directory contents
+            It will make assertions that the contents are correct
+            If MAKE_INVENTORY_REFERENCE_FILES is set, it will produce reference files
+            """
+            envvars = awx_receptor_job.runner_params['envvars']
+
+            private_data_dir = envvars.pop('AWX_PRIVATE_DATA_DIR')
+            assert envvars.pop('ANSIBLE_INVENTORY_ENABLED') == 'auto'
+            set_files = bool(os.getenv("MAKE_INVENTORY_REFERENCE_FILES", 'false').lower()[0] not in ['f', '0'])
+            env, content = read_content(private_data_dir, envvars, inventory_update)
+
+            # Assert inventory plugin inventory file is in private_data_dir
+            inventory_filename = InventorySource.injectors[inventory_update.source]().filename
+            assert (
+                len([True for k in content.keys() if k.endswith(inventory_filename)]) > 0
+            ), f"'{inventory_filename}' file not found in inventory update runtime files {content.keys()}"
+
+            env.pop('ANSIBLE_COLLECTIONS_PATHS', None)  # collection paths not relevant to this test
+            base_dir = os.path.join(DATA, 'plugins')
+            if not os.path.exists(base_dir):
+                os.mkdir(base_dir)
             source_dir = os.path.join(base_dir, this_kind)  # this_kind is a global
-
-            if not os.path.exists(source_dir):
-                raise FileNotFoundError('Maybe you never made reference files? MAKE_INVENTORY_REFERENCE_FILES=true py.test ...\noriginal: {}')
-            files_dir = os.path.join(source_dir, 'files')
-            try:
-                expected_file_list = os.listdir(files_dir)
-            except FileNotFoundError:
-                expected_file_list = []
-            for f_name in expected_file_list:
-                with open(os.path.join(files_dir, f_name), 'r') as f:
-                    ref_content = f.read()
-                    assert ref_content == content[f_name], f_name
-            try:
-                with open(os.path.join(source_dir, 'env.json'), 'r') as f:
-                    ref_env_text = f.read()
-                    ref_env = json.loads(ref_env_text)
-            except FileNotFoundError:
-                ref_env = {}
-            assert ref_env == env
-        Res = namedtuple('Result', ['status', 'rc'])
-        return Res('successful', 0)
-
-    # Mock this so that it will not send events to the callback receiver
-    # because doing so in pytest land creates large explosions
-    with mock.patch('awx.main.queue.CallbackQueueDispatcher.dispatch', lambda self, obj: None):
-        # Also do not send websocket status updates
-        with mock.patch.object(UnifiedJob, 'websocket_emit_status', mock.Mock()):
-            # The point of this test is that we replace run with assertions
-            with mock.patch('awx.main.tasks.receptor.AWXReceptorJob.run', substitute_run):
-                with mock.patch('awx.main.tasks.jobs.create_partition'):
-                    # so this sets up everything for a run and then yields control over to substitute_run
-                    task.run(inventory_update.pk)
+            if set_files:
+                create_reference_data(source_dir, env, content)
+                pytest.skip('You set MAKE_INVENTORY_REFERENCE_FILES, so this created files, unset to run actual test.')
+            else:
+                source_dir = os.path.join(base_dir, this_kind)  # this_kind is a global
+
+                if not os.path.exists(source_dir):
+                    raise FileNotFoundError('Maybe you never made reference files? MAKE_INVENTORY_REFERENCE_FILES=true py.test ...\noriginal: {}')
+                files_dir = os.path.join(source_dir, 'files')
+                try:
+                    expected_file_list = os.listdir(files_dir)
+                except FileNotFoundError:
+                    expected_file_list = []
+                for f_name in expected_file_list:
+                    with open(os.path.join(files_dir, f_name), 'r') as f:
+                        ref_content = f.read()
+                        assert ref_content == content[f_name], f_name
+                try:
+                    with open(os.path.join(source_dir, 'env.json'), 'r') as f:
+                        ref_env_text = f.read()
+                        ref_env = json.loads(ref_env_text)
+                except FileNotFoundError:
+                    ref_env = {}
+                assert ref_env == env
+            Res = namedtuple('Result', ['status', 'rc'])
+            return Res('successful', 0)
+
+        # Mock this so that it will not send events to the callback receiver
+        # because doing so in pytest land creates large explosions
+        with mock.patch('awx.main.queue.CallbackQueueDispatcher.dispatch', lambda self, obj: None):
+            # Also do not send websocket status updates
+            with mock.patch.object(UnifiedJob, 'websocket_emit_status', mock.Mock()):
+                # The point of this test is that we replace run with assertions
+                with mock.patch('awx.main.tasks.receptor.AWXReceptorJob.run', substitute_run):
+                    with mock.patch('awx.main.tasks.jobs.create_partition'):
+                        # so this sets up everything for a run and then yields control over to substitute_run
+                        task.run(inventory_update.pk)