Repository: ansible/ansible-navigator (language: python)

Issue:
ansible-navigator artifacts don't contain EE image, args or runtime information

For debugging and other purposes the log artifact created per run contains no information about the EE, name, tags, runtime, etc. So logs passed, for example, back to an SME, 3rd level support etc., lack encapsulated context. 

##### ISSUE TYPE

 - Feature Idea

##### SUMMARY

Provide an extensive `execution-environment:` dictionary within the log so the exact context is captured. Useful attributes include:

- host OS
- container runtime (including version)
- EE name and tag
- CLI arguments.
- volume mounts
- extra vars

Goal: Anyone with the log should be able to exactly recreate the environment and test without additional files or context being supplied. 



Discussion / hints:
Hello @tonykay 

This is a great idea.

We can add a `meta` section to the artifact file with some information about the current settings for the run, we have access to every setting the user has and dump it already in a debug log, but no reason not to add it to the artifact file for informational purposes.

We won't be able to show it in the UI or use the information in navigator, but having it in the artifact file would be useful.

If we're really slick about it, we might be able to add it in ansible-navigator settings format so it could be cut and pasted into a settings file :)

thanks again great idea.
This should follow #577, as it should be able to reuse the generation of structured data from the configuration subsystem 

Fix patch:
diff --git a/src/ansible_navigator/command_runner/command_runner.py b/src/ansible_navigator/command_runner/command_runner.py
index 2a9c22ea6..66ed6919e 100644
--- a/src/ansible_navigator/command_runner/command_runner.py
+++ b/src/ansible_navigator/command_runner/command_runner.py
@@ -27,7 +27,7 @@ class Command:
     identity: str
     command: str
     post_process: Callable
-    return_code: Optional[int] = None
+    return_code: int = 0
     stdout: str = ""
     stderr: str = ""
     details: List = field(default_factory=list)
diff --git a/src/ansible_navigator/configuration_subsystem/configurator.py b/src/ansible_navigator/configuration_subsystem/configurator.py
index 677987d2a..141ff885b 100644
--- a/src/ansible_navigator/configuration_subsystem/configurator.py
+++ b/src/ansible_navigator/configuration_subsystem/configurator.py
@@ -32,6 +32,7 @@ def __init__(
         params: List[str],
         application_configuration: ApplicationConfiguration,
         apply_previous_cli_entries: Union[List, C] = C.NONE,
+        skip_roll_back: bool = False,
     ):
         """Initialize the configuration variables.
 
@@ -40,6 +41,7 @@ def __init__(
         :param apply_previous_cli_entries: Apply previous USER_CLI values where the current value
                                            is not a USER_CLI sourced value, a list of entry names
                                            ['all'] will apply all previous
+        :param skip_roll_back: Skip roll back on error
         """
         self._apply_previous_cli_entries = apply_previous_cli_entries
         self._config = application_configuration
@@ -47,6 +49,7 @@ def __init__(
         self._messages: List[LogMessage] = []
         self._params = params
         self._sanity_check()
+        self._skip_rollback = skip_roll_back
         self._unaltered_entries = deepcopy(self._config.entries)
 
     def _sanity_check(self) -> None:
@@ -58,6 +61,8 @@ def _sanity_check(self) -> None:
 
     def _roll_back(self) -> None:
         """In the case of a rollback, log the configuration state prior to roll back."""
+        if self._skip_rollback:
+            return
         message = "Configuration errors encountered, rolling back to previous configuration."
         self._messages.append(LogMessage(level=logging.WARNING, message=message))
         for entry in self._config.entries:
diff --git a/src/ansible_navigator/diagnostics.py b/src/ansible_navigator/diagnostics.py
new file mode 100644
index 000000000..da773c881
--- /dev/null
+++ b/src/ansible_navigator/diagnostics.py
@@ -0,0 +1,385 @@
+"""Produce a diagnostics report in json format."""
+import datetime
+import json
+import sys
+import traceback
+
+from dataclasses import asdict
+from dataclasses import dataclass
+from pathlib import Path
+from sys import stdout
+from typing import Callable
+from typing import Dict
+from typing import Iterator
+from typing import List
+from typing import Union
+
+from pkg_resources import working_set
+
+from .command_runner import Command
+from .command_runner import CommandRunner
+from .configuration_subsystem import ApplicationConfiguration
+from .configuration_subsystem import Constants
+from .configuration_subsystem import to_effective
+from .configuration_subsystem import to_sources
+from .image_manager import introspect
+from .image_manager import introspector
+from .utils import ansi
+from .utils.functions import ExitMessage
+from .utils.functions import LogMessage
+from .utils.functions import now_iso
+from .utils.functions import shlex_join
+from .utils.serialize import Loader
+from .utils.serialize import yaml
+
+
+JSONTypes = Union[bool, int, str, Dict, List]
+
+
+@dataclass
+class Collector:
+    """Data class for a collector."""
+
+    name: str
+
+    def start(self, color: bool):
+        """Output start information to the console.
+
+        :param color: Whether to color the message
+        """
+        message = f"Collecting {self.name} information"
+        information = f"{message:.<60}"
+        ansi.working(color=color, message=information)
+
+    def fail(self, color: bool, duration: float) -> None:
+        """Output fail information to the console.
+
+        :param color: Whether to color the message
+        :param duration: The duration of the collection
+        """
+        message = f"{self.name.capitalize()} information collection failed"
+        information = f"{message:.<60}{duration:.2f}s"
+        ansi.failed(color=color, message=information)
+
+    def finish(self, color: bool, duration: float) -> None:
+        """Output finish information to the console.
+
+        :param color: Whether to color the message
+        :param duration: The duration of the collection
+        """
+        message = f"{self.name.capitalize()} information collected"
+        information = f"{message:.<60}{duration:.2f}s"
+
+        # information = f"{message:<.50}{duration_message:>.7}"
+        ansi.success(color=color, message=information)
+
+
+@dataclass
+class Diagnostics:
+    """Data class for the diagnostics."""
+
+    # pylint: disable=too-many-instance-attributes
+
+    __WARNING__: Dict[str, JSONTypes]  # pylint: disable=invalid-name
+    basics: Dict[str, JSONTypes]
+    container_engines: Dict[str, JSONTypes]
+    execution_environment: Dict[str, JSONTypes]
+    initialization: Dict[str, JSONTypes]
+    local_system: Dict[str, JSONTypes]
+    logs: Dict[str, JSONTypes]
+    python_packages: Dict[str, JSONTypes]
+    settings: Dict[str, JSONTypes]
+    settings_file: Dict[str, JSONTypes]
+
+
+def register(collector: Collector):
+    """Register a collector.
+
+    :param collector: The collector to register
+    :returns: The decorator
+    """
+
+    def decorator(func):
+        """Add the dunder collector to the func.
+
+        :param func: The function to decorate
+        :returns: The decorated function
+        """
+        func.__collector__ = collector
+        return func
+
+    return decorator
+
+
+DIAGNOSTIC_FAILURES = 0
+
+
+class FailedCollection(Exception):
+    """Exception for a failed collection."""
+
+    def __init__(self, errors):
+        """Initialize the exception.
+
+        :param errors: The errors
+        """
+        super().__init__()
+        self.errors = errors
+
+
+def diagnostic_runner(func):
+    """Wrap and run a collector.
+
+    :param func: The function to wrap
+    :returns: The decorator
+    """
+
+    def wrapper(*args, **kwargs):
+        """Wrap and run the collector.
+
+        :param args: The positional arguments
+        :param kwargs: The keyword arguments
+        :returns: The result of the function with elapsed or error information
+        """
+        global DIAGNOSTIC_FAILURES  # pylint: disable=global-statement
+        start = datetime.datetime.now()
+        color = args[0].color
+        collector = func.__collector__
+        collector.start(color=color)
+        try:
+            result = func(*args, **kwargs)
+            duration = (datetime.datetime.now() - start).total_seconds()
+            collector.finish(color=color, duration=duration)
+        except FailedCollection as error:
+            # A collector exception, has data
+            result = error.errors
+            duration = (datetime.datetime.now() - start).total_seconds()
+            collector.fail(color=color, duration=duration)
+            DIAGNOSTIC_FAILURES += 1
+        except Exception as error:  # pylint: disable=broad-except
+            # Any other exception, has no data
+            result = {"error": str(error) + "\n" + traceback.format_exc()}
+            duration = (datetime.datetime.now() - start).total_seconds()
+            collector.fail(color=color, duration=duration)
+            DIAGNOSTIC_FAILURES += 1
+        result["duration"] = round(duration)
+        return result
+
+    return wrapper
+
+
+class DiagnosticsCollector:
+    """The diagnostics collector."""
+
+    WARNING = "The following output may contain sensitive data, please review it carefully."
+
+    def __init__(
+        self,
+        args: ApplicationConfiguration,
+        messages: List[LogMessage],
+        exit_messages: List[ExitMessage],
+    ):
+        """Initialize the ShowTech class.
+
+        :param args: The current settings
+        :param messages: The messages to log
+        :param exit_messages: The exit messages to log
+        """
+        self._args = args
+        self.color = args.display_color and stdout.isatty()
+        self._messages = messages
+        self._exit_messages = exit_messages
+
+    @property
+    def registered(self) -> Iterator[Callable]:
+        """Return the registered diagnostics.
+
+        :returns: The registered diagnostics
+        """
+        return (getattr(self, f) for f in dir(self) if hasattr(getattr(self, f), "registration"))
+
+    def run(self) -> None:
+        """Collect as much information as possible about everything and dump to a json file."""
+        ansi.warning(color=self.color, message=self.WARNING)
+        diagnostics = Diagnostics(
+            __WARNING__=self._warning(),
+            basics=self._basics(),
+            container_engines=self._container_engines(),
+            execution_environment=self._execution_environment(),
+            initialization=self._initialization(),
+            local_system=self._local_system(),
+            logs=self._log_collector(),
+            python_packages=self._python_packages(),
+            settings=self._settings(),
+            settings_file=self._settings_file(),
+        )
+        time = now_iso("local")
+        path = Path(f"diagnostics-{time}.json")
+        path.write_text(json.dumps(asdict(diagnostics), indent=4, sort_keys=True), encoding="utf-8")
+        message = f"\nDiagnostics written to: {path.resolve()}"
+        if DIAGNOSTIC_FAILURES > 0:
+            ansi.warning(color=self.color, message=message)
+        else:
+            ansi.success(color=self.color, message=message)
+        sys.exit(0)
+
+    @diagnostic_runner
+    @register(Collector(name="warning"))
+    def _warning(self) -> Dict[str, JSONTypes]:
+        """Add a warning.
+
+        :returns: The warning
+        """
+        return {"message": self.WARNING}
+
+    @diagnostic_runner
+    @register(Collector(name="basic"))
+    def _basics(self) -> Dict[str, JSONTypes]:
+        """Add basic information.
+
+        :returns: The basic information
+        """
+        return {
+            "application_name": self._args.application_name,
+            "application_version": str(self._args.application_version),
+            "action_packages": list(self._args.internals.action_packages),
+            "original_command": shlex_join(sys.argv),
+            "settings_file_path": str(self._args.internals.settings_file_path),
+            "settings_source": str(self._args.internals.settings_source),
+            "share_directory": self._args.internals.share_directory,
+        }
+
+    @diagnostic_runner
+    @register(Collector(name="container engines"))
+    def _container_engines(self) -> Dict[str, JSONTypes]:
+        """Add container engines.
+
+        :returns: The container engines
+        """
+        commands = [
+            Command(identity="podman", command="podman --version", post_process=lambda c: c),
+            Command(identity="docker", command="docker --version", post_process=lambda c: c),
+        ]
+        CommandRunner().run_single_proccess(commands)
+        engines: Dict[str, JSONTypes] = {}
+        for command in commands:
+            engines[command.identity] = {
+                "return_code": command.return_code,
+                "selected": bool(self._args.container_engine == command.identity),
+                "stdout": command.stdout,
+                "stderr": command.stderr,
+            }
+        return engines
+
+    @diagnostic_runner
+    @register(Collector(name="execution environment"))
+    def _execution_environment(self) -> Dict[str, JSONTypes]:
+        """Add execution environment information.
+
+        :raises FailedCollection: If the collection process fails
+        :returns: The execution environment information
+        """
+        if self._args.entry("container_engine").value.source is Constants.DEFAULT_CFG:
+            return {"errors": "No container engine available or found"}
+
+        details, errors, return_code = introspector.run(
+            image_name=self._args.execution_environment_image,
+            container_engine=self._args.container_engine,
+        )
+        details = {"details": details, "errors": errors, "return_code": return_code}
+        if errors or not details:
+            raise FailedCollection(details)
+        return details
+
+    @diagnostic_runner
+    @register(Collector(name="initialization"))
+    def _initialization(self) -> Dict[str, JSONTypes]:
+        """Add initialization information.
+
+        :returns: The initialization information
+        """
+        return {
+            "messages": [msg.message for msg in self._messages],
+            "exit_messages": [msg.message for msg in self._exit_messages],
+        }
+
+    @diagnostic_runner
+    @register(Collector(name="log"))
+    def _log_collector(self) -> Dict[str, JSONTypes]:
+        """Add log collector information.
+
+        :returns: The log collector information
+        """
+        logs: List[JSONTypes] = []
+        cwd_log = Path("./ansible-navigator.log")
+        if cwd_log.exists():
+            contents = cwd_log.read_text(encoding="utf-8").splitlines()
+            log = {
+                "name": str(cwd_log),
+                "contents": contents,
+                "date": cwd_log.stat().st_mtime,
+            }
+            logs.append(log)
+        settings_log = Path(self._args.log_file)
+        if cwd_log != settings_log and settings_log.exists():
+            contents = settings_log.read_text(encoding="utf-8").splitlines()
+            log = {
+                "name": str(settings_log),
+                "contents": contents,
+                "date": settings_log.stat().st_mtime,
+            }
+            logs.append(log)
+        return {"found": bool(logs), "logs": logs}
+
+    @diagnostic_runner
+    @register(Collector(name="local system"))
+    def _local_system(self) -> Dict[str, JSONTypes]:
+        # pylint:disable=no-self-use
+        """Add local system information.
+
+        :raises FailedCollection: If the collection process fails
+        :returns: The local system information
+        """
+        results = introspect.main(serialize=False)
+        if not results:
+            raise FailedCollection(results)
+        if results.get("errors"):
+            raise FailedCollection(results["errors"])
+        return {"details": results}
+
+    @diagnostic_runner
+    @register(Collector(name="python packages"))
+    def _python_packages(self) -> Dict[str, JSONTypes]:
+        # pylint: disable=no-self-use
+        """Add python packages information.
+
+        :returns: The python packages information
+        """
+        return {
+            i.key: {"location": i.location, "name": i.key, "version": i.version}
+            for i in working_set  # pylint: disable=not-an-iterable
+        }
+
+    @diagnostic_runner
+    @register(Collector(name="settings"))
+    def _settings(self) -> Dict[str, JSONTypes]:
+        """Add settings information.
+
+        :returns: The settings information
+        """
+        return {
+            "effective": to_effective(self._args),
+            "sources": to_sources(self._args),
+        }
+
+    @diagnostic_runner
+    @register(Collector(name="settings file"))
+    def _settings_file(self) -> Dict[str, JSONTypes]:
+        """Add settings file information.
+
+        :returns: The settings file information
+        """
+        contents: Dict[str, JSONTypes] = {}
+        if self._args.internals.settings_file_path:
+            text = Path(self._args.internals.settings_file_path).read_text(encoding="utf-8")
+            contents = yaml.load(text, Loader=Loader)
+        return {"contents": contents}
diff --git a/src/ansible_navigator/image_manager/introspect.py b/src/ansible_navigator/image_manager/introspect.py
new file mode 100644
index 000000000..0955d90b7
--- /dev/null
+++ b/src/ansible_navigator/image_manager/introspect.py
@@ -0,0 +1,412 @@
+"""Introspect an execution environment image."""
+import json
+import os
+import re
+import subprocess
+import sys
+import threading
+
+from queue import Queue
+from types import SimpleNamespace
+from typing import Callable
+from typing import Dict
+from typing import List
+from typing import Optional
+from typing import Union
+
+
+# pylint: disable=broad-except
+
+JSONTypes = Union[bool, int, str, Dict, List]
+
+
+class Command(SimpleNamespace):
+    """Abstraction for a details about a shell command."""
+
+    id: str
+    command: str
+    parse: Callable
+    stdout: str = ""
+    stderr: str = ""
+    details: Union[List, Dict, str] = ""
+    errors: List = []
+
+
+def run_command(command: Command) -> None:
+    """Run a command using subprocess.
+
+    :param command: Details of the command to run
+    """
+    try:
+        proc_out = subprocess.run(
+            command.command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            check=True,
+            universal_newlines=True,
+            shell=True,
+        )
+        command.stdout = proc_out.stdout
+    except subprocess.CalledProcessError as exc:
+        command.stderr = str(exc.stderr)
+        command.errors = [str(exc.stderr)]
+
+
+def worker(pending_queue: Queue, completed_queue: Queue) -> None:
+    """Run a command from pending, parse, and place in completed.
+
+    :param pending_queue: A queue with plugins to process
+    :param completed_queue: The queue in which extracted documentation will be placed
+    """
+    while True:
+        command = pending_queue.get()
+        if command is None:
+            break
+        run_command(command)
+        try:
+            command.parse(command)
+        except Exception as exc:
+            command.errors = command.errors + [str(exc)]
+        completed_queue.put(command)
+
+
+class CommandRunner:
+    """A command runner.
+
+    Run commands using single or multiple processes.
+    """
+
+    def __init__(self):
+        """Initialize the command runner."""
+        self._completed_queue: Optional[Queue] = None
+        self._pending_queue: Optional[Queue] = None
+
+    def run_multi_thread(self, command_classes):
+        """Run commands with multiple threads.
+
+        Workers are started to read from pending queue.
+        Exit when the number of results is equal to the number
+        of commands needing to be run.
+
+        :param command_classes: All command classes to be run
+        :returns: The results from running all commands
+        """
+        if self._completed_queue is None:
+            self._completed_queue = Queue()
+        if self._pending_queue is None:
+            self._pending_queue = Queue()
+        results = {}
+        all_commands = tuple(
+            command for command_class in command_classes for command in command_class.commands
+        )
+        self.start_workers(all_commands)
+        results = []
+        while len(results) != len(all_commands):
+            results.append(self._completed_queue.get())
+        return results
+
+    def start_workers(self, jobs):
+        """Start workers and submit jobs to pending queue.
+
+        :param jobs: The jobs to be run
+        """
+        worker_count = len(jobs)
+        processes = []
+        for _proc in range(worker_count):
+            proc = threading.Thread(
+                target=worker,
+                args=(self._pending_queue, self._completed_queue),
+            )
+            processes.append(proc)
+            proc.start()
+        for job in jobs:
+            self._pending_queue.put(job)
+        for _proc in range(worker_count):
+            self._pending_queue.put(None)
+        for proc in processes:
+            proc.join()
+
+
+class CmdParser:
+    """A base class for command parsers with common parsing functions."""
+
+    @staticmethod
+    def _strip(value: str) -> str:
+        """Remove quotes, leading and trailing whitespace.
+
+        :param value: The string to act on
+        :returns: The string after removing quotes, leading and trailing whitespace
+        """
+        return value.strip('"').strip("'").strip()
+
+    @staticmethod
+    def re_partition(content, separator):
+        """Partition a string using a regular expression.
+
+        :param content: The content to partition
+        :param separator: The separator to use for the partitioning
+        :returns: The first partition, separator, and final partition
+        """
+        separator_match = re.search(separator, content)
+        if not separator_match:
+            return content, "", ""
+        matched_separator = separator_match.group(0)
+        parts = re.split(matched_separator, content, 1)
+        return parts[0], matched_separator, parts[1]
+
+    def splitter(self, lines, delimiter):
+        """Split lines given a delimiter.
+
+        :param lines: The lines to split
+        :param delimiter: The delimiter to use for splitting
+        :returns: All lines split on the delimiter
+        """
+        results = []
+        result = {}
+        while lines:
+            line = lines.pop()
+            left, delim, right = self.re_partition(line, delimiter)
+            right = self._strip(right)
+            if not delim:
+                if result:
+                    results.append(result)
+                    result = {}
+                continue
+            key = left.lower().replace("_", "-").strip()
+            value = right
+            result[key] = value
+        if result:
+            results.append(result)
+        return results
+
+
+class AnsibleCollections(CmdParser):
+    """Available ansible collections collector."""
+
+    @property
+    def commands(self):
+        """Define the ansible-galaxy command to list ansible collections.
+
+        :returns: The defined command
+        """
+        command = "ansible-galaxy collection list"
+        return [
+            Command(
+                id="ansible_collections",
+                command=command,
+                parse=self.parse,
+            ),
+        ]
+
+    @staticmethod
+    def parse(command: Command):
+        """Parse the output of the ansible-galaxy command.
+
+        :param command: The result of running the command
+        """
+        collections = {}
+        for line in command.stdout.splitlines():
+            parts = line.split()
+            if len(parts) == 2 and parts[1][0].isdigit():
+                collections[parts[0].strip()] = parts[1].strip()
+        command.details = collections
+
+
+class AnsibleVersion(CmdParser):
+    """Ansible version collector."""
+
+    @property
+    def commands(self) -> List[Command]:
+        """Define the ansible command to get the version.
+
+        :returns: The defined command
+        """
+        return [Command(id="ansible_version", command="ansible --version", parse=self.parse)]
+
+    @staticmethod
+    def parse(command: Command) -> None:
+        """Parse the output of the ansible command.
+
+        :param command: The result of running the command
+        """
+        version = command.stdout.splitlines()[0].split(" ", 1)[1].strip()[1:-1]
+        command.details = version
+
+
+class OsRelease(CmdParser):
+    """OS release information collector."""
+
+    @property
+    def commands(self) -> List[Command]:
+        """Define the command to collect os release information.
+
+        :returns: The defined command
+        """
+        return [Command(id="os_release", command="cat /etc/os-release", parse=self.parse)]
+
+    def parse(self, command) -> None:
+        """Parse the output of the cat command.
+
+        :param command: The result of running the command
+        """
+        parsed = self.splitter(command.stdout.splitlines(), "=")
+        command.details = parsed
+
+
+class PythonPackages(CmdParser):
+    """Python package collector."""
+
+    @property
+    def commands(self) -> List[Command]:
+        """Define the pip command to list installed pip packages.
+
+        :returns: The defined command
+        """
+        pre = Command(id="pip_freeze", command="python3 -m pip freeze", parse=self.parse_freeze)
+        run_command(pre)
+        pre.parse(pre)
+        pkgs = " ".join(pkg for pkg in pre.details[0])
+        return [
+            Command(id="python_packages", command=f"python3 -m pip show {pkgs}", parse=self.parse),
+        ]
+
+    def parse(self, command):
+        """Parse the output of the pip command.
+
+        :param command: The result of running the command
+        """
+        parsed = self.splitter(command.stdout.splitlines(), ":")
+        for pkg in parsed:
+            for entry in ["required-by", "requires"]:
+                if pkg[entry]:
+                    pkg[entry] = [p.strip() for p in pkg[entry].split(",")]
+                else:
+                    pkg[entry] = []
+        command.details = parsed
+
+    def parse_freeze(self, command):
+        """Parse the output of the pip freeze command, skipping editables.
+
+        :param command: The result of running the command
+        """
+        lines = [line for line in command.stdout.splitlines() if not line.startswith("-e")]
+        parsed = self.splitter(lines, "(==|@)")
+        command.details = parsed
+
+
+class RedhatRelease(CmdParser):
+    """Red Hat release collector."""
+
+    @property
+    def commands(self) -> List[Command]:
+        """Define the command to get the redhat release information.
+
+        :returns: The defined command
+        """
+        return [Command(id="redhat_release", command="cat /etc/redhat-release", parse=self.parse)]
+
+    @staticmethod
+    def parse(command):
+        """Parse the output of the cat redhat release command.
+
+        :param command: The result of running the command
+        """
+        parsed = command.stdout
+        command.details = parsed
+
+
+class SystemPackages(CmdParser):
+    """System packages collector."""
+
+    @property
+    def commands(self) -> List[Command]:
+        """Define the command to list system packages.
+
+        :returns: The defined command
+        """
+        return [Command(id="system_packages", command="rpm -qai", parse=self.parse)]
+
+    def parse(self, command):
+        """Parse the output of the rpm command.
+
+        :param command: The result of running the command
+        """
+        packages = []
+        package = []
+        for line in command.stdout.splitlines():
+            if re.match(r"^Name\s{2,}:", line) and package:
+                packages.append(package)
+                package = [line]
+            else:
+                package.append(line)
+        if package:
+            packages.append(package)
+
+        parsed = []
+        for package in packages:
+            entry = {}
+            while package:
+                line = package.pop(0)
+                left, _delim, right = self.re_partition(line, ":")
+                key = left.lower().replace("_", "-").strip()
+
+                # Description is at the end of the package section
+                # read until package is empty
+                if key == "description":
+                    description = []
+                    while package:
+                        description.append(package.pop(0))
+                    # Normalize the data, in the case description is totally empty
+                    if description:
+                        entry[key] = "\n".join(description)
+                    else:
+                        entry[key] = "No description available"
+                    parsed.append(entry)
+                # other package details are 1 line each
+                else:
+                    value = self._strip(right)
+                    entry[key] = value
+
+        command.details = parsed
+
+
+def main(serialize: bool = True) -> Optional[Dict[str, JSONTypes]]:
+    """Enter the image introspection process.
+
+    :param serialize: Whether to serialize the results
+    :returns: The collected data or none if serialize is False
+    """
+    response: Dict = {"errors": []}
+    response["python_version"] = {"details": {"version": " ".join(sys.version.splitlines())}}
+    response["environment_variables"] = {"details": dict(os.environ)}
+    try:
+        command_runner = CommandRunner()
+        commands = [
+            AnsibleCollections(),
+            AnsibleVersion(),
+            OsRelease(),
+            RedhatRelease(),
+            PythonPackages(),
+            SystemPackages(),
+        ]
+
+        results = command_runner.run_multi_thread(commands)
+
+        for result in results:
+            result_as_dict = vars(result)
+            result_as_dict.pop("parse")
+            for key in list(result_as_dict.keys()):
+                if key not in ["details", "errors"]:
+                    result_as_dict[f"__{key}"] = result_as_dict[key]
+                    result_as_dict.pop(key)
+            response[result_as_dict["__id"]] = result_as_dict
+    except Exception as exc:
+        response["errors"].append(str(exc))
+    if serialize:
+        print(json.dumps(response))
+        return None
+    return response
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/ansible_navigator/image_manager/introspector.py b/src/ansible_navigator/image_manager/introspector.py
new file mode 100644
index 000000000..43aefdb57
--- /dev/null
+++ b/src/ansible_navigator/image_manager/introspector.py
@@ -0,0 +1,71 @@
+"""An image introspector."""
+import inspect
+import json
+import logging
+import tempfile
+
+from pathlib import Path
+from typing import Dict
+from typing import List
+from typing import Tuple
+
+from ..runner import Command
+from . import introspect
+
+
+logger = logging.getLogger(__name__)
+
+
+def run(image_name: str, container_engine: str) -> Tuple[Dict, List[str], int]:
+    """Run runner to collect image details.
+
+    :param image_name: The full image name
+    :param container_engine: The container engine to use
+    :returns: Output, errors and the return code
+    """
+    errors = []
+    python_exec_path = "python3"
+
+    with tempfile.TemporaryDirectory() as tmp_dir_name:
+        introspect_source = inspect.getsource(introspect)
+        file = Path(tmp_dir_name) / "introspect.py"
+        file.write_text(introspect_source)
+
+        _runner = Command(
+            cmdline=[str(file)],
+            container_engine=container_engine,
+            container_volume_mounts=[f"{tmp_dir_name}:{tmp_dir_name}"],
+            executable_cmd=python_exec_path,
+            execution_environment_image=image_name,
+            execution_environment=True,
+            navigator_mode="interactive",
+        )
+        output, error, return_code = _runner.run()
+    if error:
+        errors.append(error)
+    parse_errors, parsed = parse(output)
+    errors.extend(parse_errors)
+    return parsed, errors, return_code
+
+
+def parse(output) -> Tuple[List[str], Dict]:
+    """Load and process the ``json`` output from the image introspection process.
+
+    :param output: The output from the image introspection process
+    :returns: The parsed output
+    """
+    errors = []
+    try:
+        if not output.startswith("{"):
+            _warnings, json_str = output.split("{", 1)
+            json_str = "{" + json_str
+        else:
+            json_str = output
+        parsed = json.loads(json_str)
+    except (json.decoder.JSONDecodeError, ValueError) as exc:
+        errors.append("Unable to extract introspection from stdout")
+        errors.append(f"error json loading output: '{str(exc)}'")
+        errors.append(output)
+        return errors, {}
+
+    return errors, parsed
diff --git a/src/ansible_navigator/initialization.py b/src/ansible_navigator/initialization.py
index 0336bb42b..5c3557423 100644
--- a/src/ansible_navigator/initialization.py
+++ b/src/ansible_navigator/initialization.py
@@ -16,6 +16,7 @@
 from .configuration_subsystem import ApplicationConfiguration
 from .configuration_subsystem import Configurator
 from .configuration_subsystem import Constants as C
+from .diagnostics import DiagnosticsCollector
 from .utils.functions import ExitMessage
 from .utils.functions import ExitPrefix
 from .utils.functions import LogMessage
@@ -141,6 +142,30 @@ def get_and_check_collection_doc_cache(
     return messages, exit_messages, collection_cache
 
 
+# pylint: disable=inconsistent-return-statements
+def _diagnose(
+    args: ApplicationConfiguration,
+    exit_messages: List[ExitMessage],
+    messages: List[LogMessage],
+    should_diagnose: bool,
+):
+    """Direct to the diagnostic information writer or return.
+
+    :param args: Application configuration
+    :param exit_messages: List of exit messages
+    :param messages: List of log messages
+    :param should_diagnose: Whether to show tech info
+    :returns: All messages and exit messages
+    """
+    if should_diagnose:
+        DiagnosticsCollector(args=args, messages=messages, exit_messages=exit_messages).run()
+    else:
+        return messages, exit_messages
+
+
+# pylint: enable=inconsistent-return-statements
+
+
 def parse_and_update(
     params: List,
     args: ApplicationConfiguration,
@@ -160,6 +185,7 @@ def parse_and_update(
     """
     messages: List[LogMessage] = []
     exit_messages: List[ExitMessage] = []
+    should_diagnose = "--diagnostics" in params
 
     (
         new_messages,
@@ -170,12 +196,18 @@ def parse_and_update(
     messages.extend(new_messages)
     exit_messages.extend(new_exit_messages)
     if exit_messages:
-        return messages, exit_messages
+        return _diagnose(
+            args=args,
+            exit_messages=exit_messages,
+            messages=messages,
+            should_diagnose=should_diagnose,
+        )
 
     configurator = Configurator(
         params=params,
         application_configuration=args,
         apply_previous_cli_entries=apply_previous_cli_entries,
+        skip_roll_back=should_diagnose,
     )
 
     new_messages, new_exit_messages = configurator.configure()
@@ -201,7 +233,12 @@ def parse_and_update(
         exit_messages.extend(new_exit_messages)
         if cache is None:
             # There's nothing to be done here, it cannot be attached.
-            return messages, exit_messages
+            return _diagnose(
+                args=args,
+                exit_messages=exit_messages,
+                messages=messages,
+                should_diagnose=should_diagnose,
+            )
         if attach_cdc:
             args.internals.collection_doc_cache = cache
             message = "Collection doc cache attached to args.internals"
@@ -211,11 +248,21 @@ def parse_and_update(
             messages.append(LogMessage(level=logging.DEBUG, message=message))
 
     if exit_messages:
-        return messages, exit_messages
+        return _diagnose(
+            args=args,
+            exit_messages=exit_messages,
+            messages=messages,
+            should_diagnose=should_diagnose,
+        )
 
     for entry in args.entries:
         message = f"Running with {entry.name} as '{entry.value.current}'"
         message += f" ({type(entry.value.current).__name__}/{entry.value.source.value})"
         messages.append(LogMessage(level=logging.DEBUG, message=message))
 
-    return messages, exit_messages
+    return _diagnose(
+        args=args,
+        exit_messages=exit_messages,
+        messages=messages,
+        should_diagnose=should_diagnose,
+    )
diff --git a/src/ansible_navigator/utils/ansi.py b/src/ansible_navigator/utils/ansi.py
new file mode 100644
index 000000000..b57fb378f
--- /dev/null
+++ b/src/ansible_navigator/utils/ansi.py
@@ -0,0 +1,63 @@
+"""ANSI color codes and functions for terminal output."""
+
+
+class Color:
+    """Color constants."""
+
+    GREY = "\033[90m"
+    RED = "\033[91m"
+    GREEN = "\033[92m"
+    YELLOW = "\033[93m"
+    BLUE = "\033[94m"
+    MAGENTA = "\033[95m"
+    CYAN = "\033[96m"
+    WHITE = "\033[97m"
+    END = "\033[0m"
+
+
+def failed(color: bool, message: str):
+    """Output failure information to the console.
+
+    :param color: Whether to color the message
+    :param message: The message to output
+    """
+    if color:
+        print(f"\r{Color.RED}{message}{Color.END}\033[K")
+    else:
+        print(message)
+
+
+def success(color: bool, message: str):
+    """Output success information to the console.
+
+    :param color: Whether to color the message
+    :param message: The message to output
+    """
+    if color:
+        print(f"\r{Color.GREEN}{message}{Color.END}\033[K")
+    else:
+        print(message)
+
+
+def warning(color: bool, message: str):
+    """Output warning information to the console.
+
+    :param color: Whether to color the message
+    :param message: The message to output
+    """
+    if color:
+        print(f"{Color.YELLOW}{message}{Color.END}")
+    else:
+        print(message)
+
+
+def working(color: bool, message: str):
+    """Output working information to the console.
+
+    :param color: Whether to color the message
+    :param message: The message to output
+    """
+    if color:
+        print(f"{Color.GREY}{message}{Color.END}", end="", flush=True)
+    else:
+        print(message)


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index f2e5d014c..479916ddc 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -3,10 +3,22 @@
 import shutil
 import subprocess
 
+from copy import deepcopy
 from pathlib import Path
+from typing import Tuple
 
 import pytest
 
+from ansible_navigator.configuration_subsystem import to_sample
+from ansible_navigator.configuration_subsystem.definitions import SettingsFileType
+from ansible_navigator.configuration_subsystem.navigator_configuration import (
+    NavigatorConfiguration,
+)
+from ansible_navigator.content_defs import ContentView
+from ansible_navigator.content_defs import SerializationFormat
+from ansible_navigator.utils.serialize import Loader
+from ansible_navigator.utils.serialize import serialize_write_file
+from ansible_navigator.utils.serialize import yaml
 from .defaults import PULLABLE_IMAGE
 
 
@@ -58,3 +70,41 @@ def use_venv(monkeypatch: pytest.MonkeyPatch):
         )
     path_prepend = Path.cwd() / venv_path / "bin"
     monkeypatch.setenv("PATH", str(path_prepend), prepend=os.pathsep)
+
+
+@pytest.fixture(name="settings_samples")
+def _settings_samples() -> Tuple[str, str]:
+    """Provide the full settings samples
+
+    :returns: The commented and uncommented samples
+    """
+    settings = deepcopy(NavigatorConfiguration)
+    commented, uncommented = to_sample(settings=settings)
+    return commented, uncommented
+
+
+@pytest.fixture()
+def settings_env_var_to_full(
+    settings_samples: Tuple[str, str],
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> Tuple[Path, SettingsFileType]:
+    """Set the settings environment variable to a full sample settings file in a tmp path."""
+    _commented, uncommented = settings_samples
+    settings_contents = yaml.load(uncommented, Loader=Loader)
+    settings_path = tmp_path / "ansible-navigator.yml"
+    serialize_write_file(
+        content=settings_contents,
+        content_view=ContentView.NORMAL,
+        file_mode="w",
+        file=settings_path,
+        serialization_format=SerializationFormat.YAML,
+    )
+    # The sample has volume mounts, so let's not exit because they doesn't exit
+    root = settings_contents["ansible-navigator"]
+    ee_root = root["execution-environment"]
+    for volume_mount in ee_root["volume-mounts"]:
+        Path(volume_mount["src"]).mkdir(exist_ok=True)
+
+    monkeypatch.setenv("ANSIBLE_NAVIGATOR_CONFIG", str(settings_path))
+    return settings_path, settings_contents
diff --git a/tests/integration/diagnostics/__init__.py b/tests/integration/diagnostics/__init__.py
new file mode 100644
index 000000000..cb280a76d
--- /dev/null
+++ b/tests/integration/diagnostics/__init__.py
@@ -0,0 +1,1 @@
+"""Tests for diagnostics."""
diff --git a/tests/integration/diagnostics/test_from_cli.py b/tests/integration/diagnostics/test_from_cli.py
new file mode 100644
index 000000000..5abd27029
--- /dev/null
+++ b/tests/integration/diagnostics/test_from_cli.py
@@ -0,0 +1,51 @@
+"""Check diagnostics output."""
+import json
+import subprocess
+
+from pathlib import Path
+from typing import Tuple
+
+import pytest
+
+from ansible_navigator.configuration_subsystem.definitions import SettingsFileType
+from ansible_navigator.utils.functions import remove_ansi
+
+
+@pytest.mark.usefixtures("use_venv")
+def test(
+    monkeypatch: pytest.MonkeyPatch,
+    settings_env_var_to_full: Tuple[Path, SettingsFileType],
+    tmp_path: Path,
+):
+    """Test diagnostics generation.
+
+    :param monkeypatch: Fixture for patching
+    :param settings_env_var_to_full: The pytest subtest fixture
+    :param tmp_path: The pytest tmp_path fixture
+
+    :raises AssertionError: When tests fails
+    """
+    monkeypatch.chdir(tmp_path)
+    settings_path, settings_file = settings_env_var_to_full
+
+    proc_out = subprocess.run(
+        "ansible-navigator --diagnostics",
+        check=False,
+        shell=True,
+        stderr=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        universal_newlines=True,
+    )
+    stdout_lines = proc_out.stdout.splitlines()
+    assert "Diagnostics written to: " in stdout_lines[-1]
+
+    file_name = remove_ansi(stdout_lines[-1].split(":", 1)[-1].strip())
+    full_path = tmp_path / file_name
+    contents = full_path.read_text()
+    diagnostics = json.loads(contents)
+
+    assert diagnostics["settings_file"]["contents"] == settings_file
+    local_env_vars = diagnostics["local_system"]["details"]["environment_variables"]
+    assert local_env_vars["details"]["ANSIBLE_NAVIGATOR_CONFIG"] == str(settings_path)
+    for _section_name, section in diagnostics.items():
+        assert not section.get("errors")
diff --git a/tests/unit/configuration_subsystem/conftest.py b/tests/unit/configuration_subsystem/conftest.py
index 1f2bcf2b0..80d928a84 100644
--- a/tests/unit/configuration_subsystem/conftest.py
+++ b/tests/unit/configuration_subsystem/conftest.py
@@ -3,33 +3,26 @@
 import os
 
 from copy import deepcopy
-from pathlib import Path
 from typing import Dict
 from typing import List
 from typing import NamedTuple
-from typing import Tuple
 
 import pytest
 
 from ansible_navigator.command_runner import Command
 from ansible_navigator.command_runner.command_runner import run_command
-from ansible_navigator.configuration_subsystem import to_sample
 from ansible_navigator.configuration_subsystem import to_schema
 from ansible_navigator.configuration_subsystem.configurator import Configurator
 from ansible_navigator.configuration_subsystem.definitions import (
     ApplicationConfiguration,
 )
-from ansible_navigator.configuration_subsystem.definitions import SettingsFileType
 from ansible_navigator.configuration_subsystem.definitions import SettingsSchemaType
 from ansible_navigator.configuration_subsystem.navigator_configuration import (
     NavigatorConfiguration,
 )
-from ansible_navigator.content_defs import ContentView
-from ansible_navigator.content_defs import SerializationFormat
 from ansible_navigator.utils.functions import ExitMessage
 from ansible_navigator.utils.functions import LogMessage
 from ansible_navigator.utils.serialize import Loader
-from ansible_navigator.utils.serialize import serialize_write_file
 from ansible_navigator.utils.serialize import yaml
 from ...defaults import FIXTURES_DIR
 
@@ -137,41 +130,3 @@ def property_dive(subschema: Dict):
 
     property_dive(schema_dict)
     return schema_dict
-
-
-@pytest.fixture(name="settings_samples")
-def _settings_samples() -> Tuple[str, str]:
-    """Provide the full settings samples
-
-    :returns: The commented and uncommented samples
-    """
-    settings = deepcopy(NavigatorConfiguration)
-    commented, uncommented = to_sample(settings=settings)
-    return commented, uncommented
-
-
-@pytest.fixture()
-def settings_env_var_to_full(
-    settings_samples: Tuple[str, str],
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> Tuple[Path, SettingsFileType]:
-    """Set the settings environment variable to a full sample settings file in a tmp path."""
-    _commented, uncommented = settings_samples
-    settings_contents = yaml.load(uncommented, Loader=Loader)
-    settings_path = tmp_path / "ansible-navigator.yml"
-    serialize_write_file(
-        content=settings_contents,
-        content_view=ContentView.NORMAL,
-        file_mode="w",
-        file=settings_path,
-        serialization_format=SerializationFormat.YAML,
-    )
-    # The sample has volume mounts, so let's not exit because they doesn't exit
-    root = settings_contents["ansible-navigator"]
-    ee_root = root["execution-environment"]
-    for volume_mount in ee_root["volume-mounts"]:
-        Path(volume_mount["src"]).mkdir(exist_ok=True)
-
-    monkeypatch.setenv("ANSIBLE_NAVIGATOR_CONFIG", str(settings_path))
-    return settings_path, settings_contents