如何在Ansible Python API中传递--limit参数运行Playbook

问题场景 #

现有inventory配置文件hosts内容如下：

[rhosts]
host_a
host_b
host_c

Playbook文件ping.yml指定目标主机组为rhosts，但需要通过Python API调用时限制仅在host_a和host_c上执行，当前代码未实现该限制，且不确定如何传递对应于--limit的参数。

当前运行Playbook的代码：

from ansible import context
from ansible.cli import CLI
from ansible.module_utils.common.collections import ImmutableDict
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.parsing.dataloader import DataLoader
from ansible.inventory.manager import InventoryManager
from ansible.vars.manager import VariableManager

INVENTORY_FILE = 'hosts'

loader = DataLoader()

context.CLIARGS = ImmutableDict(tags={}, listtags=False, listtasks=False, 
                    listhosts=False, syntax=False, connection='ssh',
                    module_path=None, forks=100, private_key_file=None,
                    ssh_common_args=None, ssh_extra_args=None, sftp_extra_args=None, 
                    scp_extra_args=None, become=True, become_method='sudo', become_user='root', 
                    verbosity=True, check=False, start_at_task=None,)

inventory = InventoryManager(loader=loader, sources=[INVENTORY_FILE])

variable_manager = VariableManager(loader=loader, inventory=inventory, version_info=CLI.version_info(gitinfo=False))

pbex = PlaybookExecutor(playbooks=['ping.yml'], 
                inventory=inventory, variable_manager=variable_manager, 
                loader=loader, passwords={})

results = pbex.run()

用户还找到Ansible内部的get_host_list函数，但不确定其用法：

def get_host_list(inventory, subset, pattern='all'):
        no_hosts = False
        if len(inventory.list_hosts()) == 0:
            # Empty inventory
            if C.LOCALHOST_WARNING and pattern not in C.LOCALHOST:
                display.warning("provided hosts list is empty, only localhost is available. Note that the implicit localhost does not match 'all'")
            no_hosts = True

        inventory.subset(subset)

        hosts = inventory.list_hosts(pattern)
        if not hosts and no_hosts is False:
            raise AnsibleError("Specified hosts and/or --limit does not match any hosts")

        return hosts

解决方案 #

Ansible命令行的--limit参数对应API中的subset选项，有两种实现方式：

方式1：通过context.CLIARGS传递subset参数 #

直接在ImmutableDict中添加subset='host_a,host_c'，效果等同于命令行执行ansible-playbook ping.yml --limit host_a,host_c：

context.CLIARGS = ImmutableDict(tags={}, listtags=False, listtasks=False, 
                    listhosts=False, syntax=False, connection='ssh',
                    module_path=None, forks=100, private_key_file=None,
                    ssh_common_args=None, ssh_extra_args=None, sftp_extra_args=None, 
                    scp_extra_args=None, become=True, become_method='sudo', become_user='root', 
                    verbosity=True, check=False, start_at_task=None,
                    subset='host_a,host_c')  # 新增主机限制参数

方式2：直接调用inventory.subset()方法 #

在初始化InventoryManager之后，手动调用inventory.subset()传入限制规则：

inventory = InventoryManager(loader=loader, sources=[INVENTORY_FILE])
inventory.subset('host_a,host_c')  # 添加主机限制

完整修改后的代码示例（以方式2为例） #

from ansible import context
from ansible.cli import CLI
from ansible.module_utils.common.collections import ImmutableDict
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.parsing.dataloader import DataLoader
from ansible.inventory.manager import InventoryManager
from ansible.vars.manager import VariableManager

INVENTORY_FILE = 'hosts'

loader = DataLoader()

context.CLIARGS = ImmutableDict(tags={}, listtags=False, listtasks=False, 
                    listhosts=False, syntax=False, connection='ssh',
                    module_path=None, forks=100, private_key_file=None,
                    ssh_common_args=None, ssh_extra_args=None, sftp_extra_args=None, 
                    scp_extra_args=None, become=True, become_method='sudo', become_user='root', 
                    verbosity=True, check=False, start_at_task=None,)

inventory = InventoryManager(loader=loader, sources=[INVENTORY_FILE])
inventory.subset('host_a,host_c')  # 添加主机限制

variable_manager = VariableManager(loader=loader, inventory=inventory, version_info=CLI.version_info(gitinfo=False))

pbex = PlaybookExecutor(playbooks=['ping.yml'], 
                inventory=inventory, variable_manager=variable_manager, 
                loader=loader, passwords={})

results = pbex.run()

原理说明 #

你找到的get_host_list函数核心逻辑就是调用inventory.subset(subset)过滤主机，这也是Ansible处理--limit参数的底层实现。不管是通过CLIARGS传递参数还是直接调用方法，最终都会触发inventory.subset筛选符合条件的主机列表，从而限制Playbook的执行范围。

内容的提问来源于stack exchange，提问作者Maciej Golkowski