ansible-lint speed improvement research

Project Discussions
1

The slowest part of our CI is ansible-lint, we have a modest ansible repo (36 roles, 265 playbooks), the ansible-lint execution on our CI takes between 7 and 12 minutes.

I got my hands on a Claude subscription and I was able to get it to identify two bottlenecks that improved runtime on our repo by about 40% (7m12 → 4m16s).

First one is get_deps_versions getting called for every template render. Adding @functools.cache drops 947 calls in our repo.

Second one is ansible-playbook --syntax-check invocation. Rewritting runner to call PlaybookCLI.cli_executor instead of spawning a subprocess while serializing calls to avoid messing ansible’s global state since it’s not thread safe. This is the most major improvement: faster, lighter (28% lower cpu time).

I’m curious if others can reproduce those results of if it’s just very specific to us.

Here is a sitecustomize that can be dropped in your python path to check those findings.

import contextlib
import functools
import importlib.abc
import importlib.util
import io
import os
import subprocess
import sys
import threading
import warnings

_syntax_check_lock = threading.Lock()


class _CompletedInProcess:
    def __init__(self, returncode, stdout, stderr):
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr


def _run_syntax_check_in_process(cmd, env):
    from ansible.cli.playbook import PlaybookCLI

    stdout = io.StringIO()
    stderr = io.StringIO()
    with _syntax_check_lock:
        overrides = {k: v for k, v in env.items() if os.environ.get(k) != v}
        previous = {k: os.environ.get(k) for k in overrides}
        os.environ.update(overrides)
        try:
            with (
                warnings.catch_warnings(),
                contextlib.redirect_stdout(stdout),
                contextlib.redirect_stderr(stderr),
            ):
                warnings.simplefilter("ignore")
                try:
                    PlaybookCLI.cli_executor(list(cmd))
                    returncode = 0
                except SystemExit as exc:
                    returncode = exc.code if isinstance(exc.code, int) else 1
        finally:
            for key, value in previous.items():
                if value is None:
                    os.environ.pop(key, None)
                else:
                    os.environ[key] = value
    return _CompletedInProcess(returncode, stdout.getvalue(), stderr.getvalue())


class _SyntaxCheckSubprocess:
    @staticmethod
    def run(cmd, *args, env=None, **kwargs):
        if (
            isinstance(cmd, (list, tuple))
            and len(cmd) >= 2
            and cmd[0] == "ansible-playbook"
            and "--syntax-check" in cmd
        ):
            return _run_syntax_check_in_process(cmd, env or {})
        return subprocess.run(cmd, *args, env=env, **kwargs)

    def __getattr__(self, name):
        return getattr(subprocess, name)


class _RunnerPatcher(importlib.abc.MetaPathFinder):
    _target = "ansiblelint.runner"

    def find_spec(self, fullname, path, target=None):
        if fullname != self._target:
            return None
        sys.meta_path.remove(self)
        try:
            spec = importlib.util.find_spec(fullname)
        finally:
            sys.meta_path.insert(0, self)
        if spec is None or spec.loader is None:
            return None
        original_exec_module = spec.loader.exec_module

        def exec_module(module):
            original_exec_module(module)
            # actual patching happens here
            module.subprocess = _SyntaxCheckSubprocess()
            with contextlib.suppress(ValueError):
                sys.meta_path.remove(self)

        spec.loader.exec_module = exec_module
        return spec


if not any(isinstance(f, _RunnerPatcher) for f in sys.meta_path):
    sys.meta_path.insert(0, _RunnerPatcher())


class _DepsVersionsCachePatcher(importlib.abc.MetaPathFinder):
    _target = "ansiblelint.config"

    def find_spec(self, fullname, path, target=None):
        if fullname != self._target:
            return None
        sys.meta_path.remove(self)
        try:
            spec = importlib.util.find_spec(fullname)
        finally:
            sys.meta_path.insert(0, self)
        if spec is None or spec.loader is None:
            return None
        original_exec_module = spec.loader.exec_module

        def exec_module(module):
            original_exec_module(module)
            # actual patching happens here
            module.get_deps_versions = functools.cache(module.get_deps_versions)
            with contextlib.suppress(ValueError):
                sys.meta_path.remove(self)

        spec.loader.exec_module = exec_module
        return spec


if not any(isinstance(f, _DepsVersionsCachePatcher) for f in sys.meta_path):
    sys.meta_path.insert(0, _DepsVersionsCachePatcher())