我们希望听到您的意见！帮助我们深入了解 Ansible 生态系统的现状。
参与 2024 年 Ansible 项目调查

使用过滤器进行 Ansible 数据操作

来源

今年的峰会上，一位参会者提出了一个关于如何在 Ansible 中处理设置事实和更改数据的问题。很多时候，我们会遇到人们使用一个接一个的任务来操作数据，将项目转换为列表，过滤选项，尝试进行大量数据操作，以及将数据从一个来源转换为另一个来源。尝试使用角色和剧本中的 YAML 和 Jinja 的混合来进行这些编程更改本身就是一件令人头疼的事情。虽然许多这些选项都能奏效，但它们效率不高，也不易于实现。Ansible Playbook 从未打算用于编程。

一个通常被忽视的解决方案是在模块或过滤器内部使用 Python 进行操作。本文将详细介绍如何创建过滤器来操作数据。此外，本文中引用的所有代码的存储库已创建。

此示例最初是作为模块开发的。但是，经过审查后，确定这些数据转换最好作为过滤器来完成。过滤器可以接收多个数据输入，执行编程操作，然后可以在将其用作输入或设置为事实的地方内联使用。此外，这在本地运行，而不是在主机级别运行，因此它可以更快并且避免不必要的连接。

起点

首先，我们需要一个要处理的数据集。为此，我们使用了来自自动化控制器 API 的工作流数据；它提供了每个工作流中节点的嵌套数据，以便循环遍历。在这种情况下使用的变量文件可以在存储库中找到。

目标是找到自动化控制器在循环遍历嵌套列表时正在使用什么。虽然这不是一个非常实用的例子，但它确实为创建过滤器来操作任何数据集提供了一个起点。

过滤器基础

此过滤器的骨架取自ansible.netcommon.pop_ace。每个过滤器的开头都有一些必需的选项，例如 FilterModule，此外 AnsibleFilterError 对于故障排除很有用。

from ansible.errors import AnsibleFilterError

类调用将代码设置为过滤器，并调用要用于过滤器的函数。这将设置在剧本中称为“used”的过滤器，以及要调用的函数。请注意，函数和过滤器名称不需要匹配。

class FilterModule(object):
    def filters(self):
        return {"example_filter": self.workflow_manip}

然后是文档部分：这可以包含输入、示例和其他元数据。这也是 ansible-docs 的填充方式。

EXAMPLES = r"""
    - name: Transform Data
        ansible.builtin.set_fact:
        data_out: "{{ workflow_job_templates | example_filter }}"
    """

在大多数情况下，这应该是标准信息。虽然文档字段对于过滤器不是必需的，但最好包含它。虽然此处未显示，但链接示例还包含注释掉的预期输出，这对于将来回溯和进行故障排除非常有用。

设置

首先要为传入的数据设置过滤器参数。在我们的例子中，变量 data_in，并且输入类型为 dict。最好在此处将返回值设置为为空，以及需要定义的任何其他默认值。

def example_filter(self, data_in: dict):
        workflow_data = {}
        workflow_data["workflows"] = []
        workflow_data["job_templates"] = []
        workflow_data["inventory_sources"] = []
        workflow_data["approval_nodes"] = []

下一步是进行实际的数据操作。

深入细节

在这里，我们可以开始做我们真正想做的事情，从源中获取数据，循环遍历它，并提取所需的数据。由于数据包含在嵌套列表中，因此需要一个内部循环和一个外部循环来遍历。

for workflow in data_in:
        workflow_data["workflows"].append(workflow["name"])
        for node in workflow["related"]["workflow_nodes"]:
            if node["unified_job_template"]["type"] == "inventory_source":
                workflow_data["inventory_sources"].append(
                        node["unified_job_template"]["name"]
                )
            elif node["unified_job_template"]["type"] == "job_template":
                workflow_data["job_templates"].append(
                    node["unified_job_template"]["name"]
                )
            elif node["unified_job_template"]["type"] == "workflow_approval":
                workflow_data["approval_nodes"].append(
                    node["unified_job_template"]["name"]
                )
            else:
                raise AnsibleFilterError(
                    "Failed to find valid node: {0}".format(workflow)
                )

第一个循环是查找工作流名称字段并将其追加到工作流列表中。下一个循环遍历每个工作流节点，查找其类型，并将其追加到相应的列表中。

最后是错误消息，对于有效数据不应该触发，但是当构建或调试模块时，它是一段有用的代码，可以插入到其他地方，以便强制输出到控制台以找出发生了什么。在操作结束时，使用结果变量返回。另一种方法是使用三个任务，其中两个使用循环来实现相同的结果。通过使用实际的编程语言、可用的库和内部化的循环，它简化了剧本，并提供了比仅使用 YAML 和 Jinja2 拼凑在一起更好的逻辑。

总结

希望本文能为创建过滤器和简化剧本中的任务提供一个起点。就像 Ansible 中的所有内容一样，没有唯一的解决方案，有 10 个选项可供选择。并非每个解决方案都适合手头的状况。希望这能提供另一个更好的工作选项。

由 提供支持