Swap in-memory logger to write stdout in post_save_hook

This commit makes the needed changes to inventory update
post_save_hook logic so that the historic log lines that
inventory updates write will be written to stdout,
but this hack bypasses the ansible-runner verbose event
logic and dispatches verbose events directly.

Fix the venv application with the ansible-inventory system
(note: much of this is undone in a later commit)

Deal with some minor test updates for
the ansible-inventory interface changes

6 files changed, 106 insertions, 73 deletions

diff --git a/awx/main/management/commands/inventory_import.py b/awx/main/management/commands/inventory_import.py
index 3434869c42..d987d3739a 100644
--- a/awx/main/management/commands/inventory_import.py
+++ b/awx/main/management/commands/inventory_import.py
@@ -217,10 +217,10 @@ class Command(BaseCommand):
                             'specifies the unique, immutable instance ID, may be '
                             'specified as "foo.bar" to traverse nested dicts.')
 
-    def set_logging_level(self):
+    def set_logging_level(self, verbosity):
         log_levels = dict(enumerate([logging.WARNING, logging.INFO,
                                      logging.DEBUG, 0]))
-        logger.setLevel(log_levels.get(self.verbosity, 0))
+        logger.setLevel(log_levels.get(verbosity, 0))
 
     def _get_instance_id(self, variables, default=''):
         '''
@@ -826,7 +826,7 @@ class Command(BaseCommand):
         # special check for tower-type inventory sources
         # but only if running the plugin
         TOWER_SOURCE_FILES = ['tower.yml', 'tower.yaml']
-        if self.inventory_source.source == 'tower' and any(f in self.source for f in TOWER_SOURCE_FILES):
+        if self.inventory_source.source == 'tower' and any(f in self.inventory_source.source_path for f in TOWER_SOURCE_FILES):
             # only if this is the 2nd call to license check, we cannot compare before running plugin
             if hasattr(self, 'all_group'):
                 self.remote_tower_license_compare(local_license_type)
@@ -871,6 +871,15 @@ class Command(BaseCommand):
         elif not inventory_name and not inventory_id:
             raise CommandError('--inventory-name or --inventory-id is required')
 
+        # Obtain rest of the options needed to run update
+        raw_source = options.get('source', None)
+        if not raw_source:
+            raise CommandError('--source is required')
+        source = Command.get_source_absolute_path(raw_source)
+        verbosity = int(options.get('verbosity', 1))
+        self.set_logging_level(verbosity)
+        venv_path = options.get('venv', None)
+
         # Load inventory object based on name or ID.
         if inventory_id:
             q = dict(id=inventory_id)
@@ -884,14 +893,6 @@ class Command(BaseCommand):
             raise CommandError('Inventory with %s = %s returned multiple results' % list(q.items())[0])
         logger.info('Updating inventory %d: %s' % (inventory.pk, inventory.name))
 
-        # Obtain rest of the options needed to run update
-        raw_source = options.get('source', None)
-        if not raw_source:
-            raise CommandError('--source is required')
-        source = Command.get_source_absolute_path(raw_source)
-        verbosity = int(options.get('verbosity', 1))
-        venv_path = options.get('venv', None)
-
         # Create ad-hoc inventory source and inventory update objects
         with ignore_inventory_computed_fields():
             inventory_source, created = InventorySource.objects.get_or_create(
@@ -944,8 +945,6 @@ class Command(BaseCommand):
         self.inventory_update = inventory_update
 
         # the update options, could be parser object or dict
-        self.verbosity = int(options.get('verbosity', 1))
-        self.set_logging_level()
         self.overwrite = bool(options.get('overwrite', False))
         self.overwrite_vars = bool(options.get('overwrite_vars', False))
         self.enabled_var = options.get('enabled_var', None)
@@ -1039,19 +1038,9 @@ class Command(BaseCommand):
                                     queries_before2 = len(connection.queries)
                                 self.inventory.update_computed_fields()
                             if settings.SQL_DEBUG:
-                                logger.warning('loading into database...')
-                            with ignore_inventory_computed_fields():
-                                if getattr(settings, 'ACTIVITY_STREAM_ENABLED_FOR_INVENTORY_SYNC', True):
-                                    self.load_into_database()
-                                else:
-                                    with disable_activity_stream():
-                                        self.load_into_database()
-                                if settings.SQL_DEBUG:
-                                    queries_before2 = len(connection.queries)
-                                self.inventory.update_computed_fields()
-                                if settings.SQL_DEBUG:
-                                    logger.warning('update computed fields took %d queries',
-                                                   len(connection.queries) - queries_before2)
+                                logger.warning('update computed fields took %d queries',
+                                               len(connection.queries) - queries_before2)
+
                             # Check if the license is valid.
                             # If the license is not valid, a CommandError will be thrown,
                             # and inventory update will be marked as invalid.
@@ -1064,9 +1053,9 @@ class Command(BaseCommand):
                             self.check_org_host_limit()
                     except CommandError as e:
                         if license_fail:
-                            self.mark_license_failure()
+                            self.mark_license_failure(save=True)
                         else:
-                            self.mark_org_limits_failure()
+                            self.mark_org_limits_failure(save=True)
                         raise e
 
                     if settings.SQL_DEBUG:
diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index 5e51bac386..04ac11b219 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -2485,15 +2485,11 @@ class RunInventoryUpdate(BaseTask):
             return injector.build_private_data(inventory_update, private_data_dir)
 
     def build_env(self, inventory_update, private_data_dir, isolated, private_data_files=None):
-        """Build environment dictionary for inventory import.
+        """Build environment dictionary for ansible-inventory.
 
-        This used to be the mechanism by which any data that needs to be passed
-        to the inventory update script is set up. In particular, this is how
-        inventory update is aware of its proper credentials.
-
-        Most environment injection is now accomplished by the credential
-        injectors. The primary purpose this still serves is to
-        still point to the inventory update INI or config file.
+        Most environment variables related to credentials or configuration
+        are accomplished by the inventory source injectors (in this method)
+        or custom credential type injectors (in main run method).
         """
         env = super(RunInventoryUpdate, self).build_env(inventory_update,
                                                         private_data_dir,
@@ -2501,8 +2497,10 @@ class RunInventoryUpdate(BaseTask):
                                                         private_data_files=private_data_files)
         if private_data_files is None:
             private_data_files = {}
-        self.add_awx_venv(env)
-        # Pass inventory source ID to inventory script.
+        self.add_ansible_venv(settings.ANSIBLE_VENV_PATH, env)
+
+        # Legacy environment variables, were used as signal to awx-manage command
+        # now they are provided in case some scripts may be relying on them
         env['INVENTORY_SOURCE_ID'] = str(inventory_update.inventory_source_id)
         env['INVENTORY_UPDATE_ID'] = str(inventory_update.pk)
         env.update(STANDARD_INVENTORY_UPDATE_ENV)
@@ -2568,14 +2566,21 @@ class RunInventoryUpdate(BaseTask):
         args = ['ansible-inventory', '--list', '--export']
 
         # Add arguments for the source inventory file/script/thing
+        source_location = self.pseudo_build_inventory(inventory_update, private_data_dir)
         args.append('-i')
-        args.append(self.pseudo_build_inventory(inventory_update, private_data_dir))
+        args.append(source_location)
 
         args.append('--output')
         args.append(os.path.join(private_data_dir, 'artifacts', 'output.json'))
 
+        if os.path.isdir(source_location):
+            playbook_dir = source_location
+        else:
+            playbook_dir = os.path.dirname(source_location)
+        args.extend(['--playbook-dir', playbook_dir])
+
         if inventory_update.verbosity:
-            args.append('-' + 'v' * inventory_update.verbosity)
+            args.append('-' + 'v' * min(5, inventory_update.verbosity * 2 + 1))
 
         return args
 
@@ -2713,19 +2718,82 @@ class RunInventoryUpdate(BaseTask):
         if inventory_update.verbosity:
             options['verbosity'] = inventory_update.verbosity
 
+        # Mock ansible-runner events
+        class CallbackHandler(logging.Handler):
+            def __init__(self, event_handler, cancel_callback, job_timeout, verbosity,
+                         counter=0, **kwargs):
+                self.event_handler = event_handler
+                self.cancel_callback = cancel_callback
+                self.job_timeout = job_timeout
+                self.job_start = time.time()
+                self.last_check = self.job_start
+                # TODO: we do not have events from the ansible-inventory process
+                # so there is no way to know initial counter of start line
+                self.counter = counter
+                self.skip_level = [logging.WARNING, logging.INFO, logging.DEBUG, 0][verbosity]
+                self._start_line = 0
+                super(CallbackHandler, self).__init__(**kwargs)
+
+            def emit(self, record):
+                this_time = time.time()
+                if this_time - self.last_check > 0.5:
+                    self.last_check = this_time
+                    if self.cancel_callback():
+                        raise RuntimeError('Inventory update has been canceled')
+                if self.job_timeout and ((this_time - self.job_start) > self.job_timeout):
+                    raise RuntimeError('Inventory update has timed out')
+
+                # skip logging for low severity logs
+                if record.levelno < self.skip_level:
+                    return
+
+                self.counter += 1
+                msg = self.format(record)
+                n_lines = msg.count('\n')
+                dispatch_data = dict(
+                    created=now().isoformat(),
+                    event='verbose',
+                    counter=self.counter,
+                    stdout=msg + '\n',
+                    start_line=self._start_line,
+                    end_line=self._start_line + n_lines
+                )
+                self._start_line += n_lines + 1
+
+                self.event_handler(dispatch_data)
+
+        handler = CallbackHandler(
+            self.event_handler, self.cancel_callback,
+            verbosity=inventory_update.verbosity,
+            job_timeout=self.get_instance_timeout(self.instance),
+            counter=self.event_ct
+        )
+        inv_logger = logging.getLogger('awx.main.commands.inventory_import')
+        handler.formatter = inv_logger.handlers[0].formatter
+        inv_logger.handlers[0] = handler
+
         from awx.main.management.commands.inventory_import import Command as InventoryImportCommand
         cmd = InventoryImportCommand()
-        save_status, tb, exc = cmd.perform_update(options, data, inventory_update)
+        try:
+            save_status, tb, exc = cmd.perform_update(options, data, inventory_update)
+        except Exception as raw_exc:
+            # Ignore license errors specifically
+            if 'Host limit for organization' not in str(exc) and 'License' not in str(exc):
+                raise raw_exc
 
         model_updates = {}
-        if save_status != inventory_update.status:
+        if save_status != status:
             model_updates['status'] = save_status
         if tb:
             model_updates['result_traceback'] = tb
 
         if model_updates:
-            logger.debug('{} saw problems saving to database.'.format(inventory_update.log_format))
-            model_updates['job_explanation'] = 'Update failed to save all changes to database properly'
+            logger.info('{} had problems saving to database with {}'.format(
+                inventory_update.log_format, ', '.join(list(model_updates.keys()))
+            ))
+            model_updates['job_explanation'] = 'Update failed to save all changes to database properly.'
+            if exc:
+                model_updates['job_explanation'] += ' {}'.format(exc)
             self.update_model(inventory_update.pk, **model_updates)
 
 
diff --git a/awx/main/tests/functional/commands/test_inventory_import.py b/awx/main/tests/functional/commands/test_inventory_import.py
index a0b1095c98..3fe4b92b5a 100644
--- a/awx/main/tests/functional/commands/test_inventory_import.py
+++ b/awx/main/tests/functional/commands/test_inventory_import.py
@@ -83,7 +83,7 @@ class MockLoader:
         return self._data
 
 
-def mock_logging(self):
+def mock_logging(self, level):
     pass
 
 
diff --git a/awx/main/tests/functional/test_inventory_source_injectors.py b/awx/main/tests/functional/test_inventory_source_injectors.py
index 4601668d25..dc54301d01 100644
--- a/awx/main/tests/functional/test_inventory_source_injectors.py
+++ b/awx/main/tests/functional/test_inventory_source_injectors.py
@@ -214,6 +214,8 @@ def test_inventory_update_injected_content(this_kind, inventory, fake_credential
             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
+        env.pop('PYTHONPATH')
+        env.pop('VIRTUAL_ENV')
         base_dir = os.path.join(DATA, 'plugins')
         if not os.path.exists(base_dir):
             os.mkdir(base_dir)
diff --git a/awx/main/tests/unit/commands/test_inventory_import.py b/awx/main/tests/unit/commands/test_inventory_import.py
index 105086bcb8..db3e01408b 100644
--- a/awx/main/tests/unit/commands/test_inventory_import.py
+++ b/awx/main/tests/unit/commands/test_inventory_import.py
@@ -33,32 +33,6 @@ class TestInvalidOptions:
         assert 'inventory-id' in str(err.value)
         assert 'exclusive' in str(err.value)
 
-    def test_invalid_options_id_and_keep_vars(self):
-        # You can't overwrite and keep_vars at the same time, that wouldn't make sense
-        cmd = Command()
-        with pytest.raises(CommandError) as err:
-            cmd.handle(
-                inventory_id=42, overwrite=True, keep_vars=True
-            )
-        assert 'overwrite-vars' in str(err.value)
-        assert 'exclusive' in str(err.value)
-
-    def test_invalid_options_id_but_no_source(self):
-        # Need a source to import
-        cmd = Command()
-        with pytest.raises(CommandError) as err:
-            cmd.handle(
-                inventory_id=42, overwrite=True, keep_vars=True
-            )
-        assert 'overwrite-vars' in str(err.value)
-        assert 'exclusive' in str(err.value)
-        with pytest.raises(CommandError) as err:
-            cmd.handle(
-                inventory_id=42, overwrite_vars=True, keep_vars=True
-            )
-        assert 'overwrite-vars' in str(err.value)
-        assert 'exclusive' in str(err.value)
-
     def test_invalid_options_missing_source(self):
         cmd = Command()
         with pytest.raises(CommandError) as err:
diff --git a/awx/main/tests/unit/test_tasks.py b/awx/main/tests/unit/test_tasks.py
index 0de5dce996..8dd9225f00 100644
--- a/awx/main/tests/unit/test_tasks.py
+++ b/awx/main/tests/unit/test_tasks.py
@@ -2061,8 +2061,8 @@ class TestInventoryUpdateCredentials(TestJobExecution):
                     credential, env, {}, [], private_data_dir
                 )
 
-        assert '--custom' in ' '.join(args)
-        script = args[args.index('--source') + 1]
+        assert '-i' in ' '.join(args)
+        script = args[args.index('-i') + 1]
         with open(script, 'r') as f:
             assert f.read() == inventory_update.source_script.script
         assert env['FOO'] == 'BAR'