Extend hidden_fields to allow more complicated field definitions (#872)

SUMMARY
This allows us to ignore e.g. the last-applied-configuration annotation by specifying
metadata.annotations[kubectl.kubernetes.io/last-applied-configuration]
ISSUE TYPE

Feature Pull Request

COMPONENT NAME
hidden_fields
This replaces #643 as I no longer have permissions to push to branches in this repo

Reviewed-by: Bikouo Aubin
Reviewed-by: Helen Bailey <hebailey@redhat.com>
Reviewed-by: GomathiselviS <gomathiselvi@gmail.com>
Reviewed-by: Alina Buzachis

changelogs/fragments/643-extend-hidden-fields.yaml

@@ -0,0 +1,4 @@
+---
+minor_changes:
+- k8s - Extend hidden_fields to allow the expression of more complex field types to be hidden (https://github.com/ansible-collections/kubernetes.core/pull/872)
+- k8s_info - Extend hidden_fields to allow the expression of more complex field types to be hidden (https://github.com/ansible-collections/kubernetes.core/pull/872)

plugins/module_utils/k8s/service.py

@@ -4,7 +4,7 @@
 import copy
 from json import loads
 from re import compile
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from ansible.module_utils.common.dict_transformations import dict_merge
 from ansible_collections.kubernetes.core.plugins.module_utils.hashes import (
@@ -501,47 +501,107 @@ def diff_objects(
     result["before"] = diff[0]
     result["after"] = diff[1]
 
-    if list(result["after"].keys()) != ["metadata"] or list(
+    if list(result["after"].keys()) == ["metadata"] and list(
         result["before"].keys()
-    ) != ["metadata"]:
+    ) == ["metadata"]:
-        return False, result
+        # If only metadata.generation and metadata.resourceVersion changed, ignore it
+        ignored_keys = set(["generation", "resourceVersion"])
 
-    # If only metadata.generation and metadata.resourceVersion changed, ignore it
+        if set(result["after"]["metadata"].keys()).issubset(ignored_keys) and set(
-    ignored_keys = set(["generation", "resourceVersion"])
+            result["before"]["metadata"].keys()
-
+        ).issubset(ignored_keys):
-    if not set(result["after"]["metadata"].keys()).issubset(ignored_keys):
+            return True, result
-        return False, result
-    if not set(result["before"]["metadata"].keys()).issubset(ignored_keys):
-        return False, result
 
     result["before"] = hide_fields(result["before"], hidden_fields)
     result["after"] = hide_fields(result["after"], hidden_fields)
 
-    return True, result
+    return False, result
 
 
-def hide_fields(definition: dict, hidden_fields: Optional[list]) -> dict:
+def hide_field_tree(hidden_field: str) -> List[str]:
-    if not hidden_fields:
+    result = []
-        return definition
+    key, rest = hide_field_split2(hidden_field)
-    result = copy.deepcopy(definition)
+    result.append(key)
-    for hidden_field in hidden_fields:
+    while rest:
-        result = hide_field(result, hidden_field)
+        key, rest = hide_field_split2(rest)
+        result.append(key)
+
     return result
 
 
-# hide_field is not hugely sophisticated and designed to cope
+def build_hidden_field_tree(hidden_fields: List[str]) -> Dict[str, Any]:
-# with e.g. status or metadata.managedFields rather than e.g.
+    """Group hidden field targeting the same json key
-# spec.template.spec.containers[0].env[3].value
+    Example:
-def hide_field(definition: dict, hidden_field: str) -> dict:
+        Input: ['env[3]', 'env[0]']
-    split = hidden_field.split(".", 1)
+        Output: {'env': [0, 3]}
-    if split[0] in definition:
+    """
-        if len(split) == 2:
+    output = {}
-            definition[split[0]] = hide_field(definition[split[0]], split[1])
+    for hidden_field in hidden_fields:
-        else:
+        current = output
-            del definition[split[0]]
+        tree = hide_field_tree(hidden_field)
+        for idx, key in enumerate(tree):
+            if current.get(key, "") is None:
+                break
+            if idx == (len(tree) - 1):
+                current[key] = None
+            elif key not in current:
+                current[key] = {}
+            current = current[key]
+    return output
+
+
+# hide_field should be able to cope with simple or more complicated
+# field definitions
+# e.g. status or metadata.managedFields or
+# spec.template.spec.containers[0].env[3].value or
+# metadata.annotations[kubectl.kubernetes.io/last-applied-configuration]
+def hide_field(
+    definition: Union[Dict[str, Any], List[Any]], hidden_field: Dict[str, Any]
+) -> Dict[str, Any]:
+    def dict_contains_key(obj: Dict[str, Any], key: str) -> bool:
+        return key in obj
+
+    def list_contains_key(obj: List[Any], key: str) -> bool:
+        return int(key) < len(obj)
+
+    hidden_keys = list(hidden_field.keys())
+    field_contains_key = dict_contains_key
+    field_get_key = str
+    if isinstance(definition, list):
+        # Sort with reverse=true so that when we delete an item from the list, the order is not changed
+        hidden_keys = sorted(
+            [k for k in hidden_field.keys() if k.isdecimal()], reverse=True
+        )
+        field_contains_key = list_contains_key
+        field_get_key = int
+
+    for key in hidden_keys:
+        if field_contains_key(definition, key):
+            value = hidden_field.get(key)
+            convert_key = field_get_key(key)
+            if value is None:
+                del definition[convert_key]
+            else:
+                definition[convert_key] = hide_field(definition[convert_key], value)
+                if (
+                    definition[convert_key] == dict()
+                    or definition[convert_key] == list()
+                ):
+                    del definition[convert_key]
+
     return definition
 
 
+def hide_fields(
+    definition: Dict[str, Any], hidden_fields: Optional[List[str]]
+) -> Dict[str, Any]:
+    if not hidden_fields:
+        return definition
+    result = copy.deepcopy(definition)
+    hidden_field_tree = build_hidden_field_tree(hidden_fields)
+    return hide_field(result, hidden_field_tree)
+
+
 def decode_response(resp) -> Tuple[Dict, List[str]]:
     """
     This function decodes unserialized responses from the Kubernetes python
@@ -620,3 +680,35 @@ def parse_quoted_string(quoted_string: str) -> Tuple[str, str]:
         raise ValueError("invalid quoted string: missing closing quote")
 
     return "".join(result), remainder
+
+
+# hide_field_split2 returns the first key in hidden_field and the rest of the hidden_field
+# We expect the first key to either be in brackets, to be terminated by the start of a left
+# bracket, or to be terminated by a dot.
+
+# examples would be:
+# field.another.next -> (field, another.next)
+# field[key].value -> (field, [key].value)
+# [key].value -> (key, value)
+# [one][two] -> (one, [two])
+
+
+def hide_field_split2(hidden_field: str) -> Tuple[str, str]:
+    lbracket = hidden_field.find("[")
+    rbracket = hidden_field.find("]")
+    dot = hidden_field.find(".")
+
+    if lbracket == 0:
+        # skip past right bracket and any following dot
+        rest = hidden_field[rbracket + 1 :]  # noqa: E203
+        if rest and rest[0] == ".":
+            rest = rest[1:]
+        return (hidden_field[lbracket + 1 : rbracket], rest)  # noqa: E203
+
+    if lbracket != -1 and (dot == -1 or lbracket < dot):
+        return (hidden_field[:lbracket], hidden_field[lbracket:])
+
+    split = hidden_field.split(".", 1)
+    if len(split) == 1:
+        return split[0], ""
+    return split

plugins/modules/k8s.py

@@ -188,7 +188,8 @@ options:
     description:
       - Hide fields matching this option in the result
       - An example might be C(hidden_fields=[metadata.managedFields])
-      - Only field definitions that don't reference list items are supported (so V(spec.containers[0]) would not work)
+        or V(hidden_fields=[spec.containers[0].env[3].value])
+        or V(hidden_fields=[metadata.annotations[kubectl.kubernetes.io/last-applied-configuration]])
     type: list
     elements: str
     version_added: 3.0.0

plugins/modules/k8s_info.py

@@ -48,7 +48,8 @@ options:
     description:
       - Hide fields matching any of the field definitions in the result
       - An example might be C(hidden_fields=[metadata.managedFields])
-      - Only field definitions that don't reference list items are supported (so V(spec.containers[0]) would not work)
+        or V(hidden_fields=[spec.containers[0].env[3].value])
+        or V(hidden_fields=[metadata.annotations[kubectl.kubernetes.io/last-applied-configuration]])
     type: list
     elements: str
     version_added: 3.0.0

tests/integration/targets/k8s_hide_fields/tasks/main.yml

@@ -77,6 +77,7 @@
         definition: "{{ hide_fields_base_configmap | combine({'data':{'anew':'value'}}) }}"
         hidden_fields:
           - data
+          - metadata.annotations[kubectl.kubernetes.io/last-applied-configuration]
         apply: true
       register: hf6
       diff: true
@@ -86,6 +87,22 @@
         that:
           - hf6.changed
 
+    - name: Ensure hidden fields are not present
+      assert:
+        that:
+          - >-
+            'annotations' not in hf6.result.metadata or
+            'kubectl.kubernetes.io/last-applied-configuration'
+            not in hf6.result.metadata.annotations
+          - >-
+            'annotations' not in hf6.diff.before.metadata or
+            'kubectl.kubernetes.io/last-applied-configuration'
+            not in hf6.diff.before.metadata.annotations
+          - >-
+            'annotations' not in hf6.diff.after.metadata or
+            'kubectl.kubernetes.io/last-applied-configuration'
+            not in hf6.diff.after.metadata.annotations
+
     - name: Hidden field should not show up in deletion
       k8s:
         definition: "{{ hide_fields_base_configmap}}"

tests/unit/module_utils/test_hide_fields.py

@@ -0,0 +1,264 @@
+# Copyright [2025] [Red Hat, Inc.]
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import pytest
+from ansible_collections.kubernetes.core.plugins.module_utils.k8s.service import (
+    build_hidden_field_tree,
+    hide_fields,
+)
+
+
+def test_hiding_missing_field_does_nothing():
+    output = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=dict(one="1", two="2")
+    )
+    hidden_fields = ["doesnotexist"]
+    assert hide_fields(output, hidden_fields) == output
+
+
+def test_hiding_simple_field():
+    output = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=dict(one="1", two="2")
+    )
+    hidden_fields = ["metadata"]
+    expected = dict(kind="ConfigMap", data=dict(one="1", two="2"))
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_only_key_in_dict_removes_dict():
+    output = dict(kind="ConfigMap", metadata=dict(name="foo"), data=dict(one="1"))
+    hidden_fields = ["data.one"]
+    expected = dict(kind="ConfigMap", metadata=dict(name="foo"))
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_all_keys_in_dict_removes_dict():
+    output = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=dict(one="1", two="2")
+    )
+    hidden_fields = ["data.one", "data.two"]
+    expected = dict(kind="ConfigMap", metadata=dict(name="foo"))
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_multiple_fields():
+    output = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=dict(one="1", two="2")
+    )
+    hidden_fields = ["metadata", "data.one"]
+    expected = dict(kind="ConfigMap", data=dict(two="2"))
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_dict_key():
+    output = dict(
+        kind="ConfigMap",
+        metadata=dict(
+            name="foo",
+            annotations={
+                "kubectl.kubernetes.io/last-applied-configuration": '{"testvalue"}'
+            },
+        ),
+        data=dict(one="1", two="2"),
+    )
+    hidden_fields = [
+        "metadata.annotations[kubectl.kubernetes.io/last-applied-configuration]",
+    ]
+    expected = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=dict(one="1", two="2")
+    )
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_list_value_key():
+    output = dict(
+        kind="Pod",
+        metadata=dict(name="foo"),
+        spec=dict(
+            containers=[
+                dict(
+                    name="containers",
+                    image="busybox",
+                    env=[
+                        dict(name="ENV1", value="env1"),
+                        dict(name="ENV2", value="env2"),
+                        dict(name="ENV3", value="env3"),
+                    ],
+                )
+            ]
+        ),
+    )
+    hidden_fields = ["spec.containers[0].env[1].value"]
+    expected = dict(
+        kind="Pod",
+        metadata=dict(name="foo"),
+        spec=dict(
+            containers=[
+                dict(
+                    name="containers",
+                    image="busybox",
+                    env=[
+                        dict(name="ENV1", value="env1"),
+                        dict(name="ENV2"),
+                        dict(name="ENV3", value="env3"),
+                    ],
+                )
+            ]
+        ),
+    )
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_last_list_item():
+    output = dict(
+        kind="Pod",
+        metadata=dict(name="foo"),
+        spec=dict(
+            containers=[
+                dict(
+                    name="containers",
+                    image="busybox",
+                    env=[
+                        dict(name="ENV1", value="env1"),
+                    ],
+                )
+            ]
+        ),
+    )
+    hidden_fields = ["spec.containers[0].env[0]"]
+    expected = dict(
+        kind="Pod",
+        metadata=dict(name="foo"),
+        spec=dict(
+            containers=[
+                dict(
+                    name="containers",
+                    image="busybox",
+                )
+            ]
+        ),
+    )
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hiding_nested_dicts_using_brackets():
+    output = dict(
+        kind="Pod",
+        metadata=dict(name="foo"),
+        spec=dict(
+            containers=[
+                dict(
+                    name="containers",
+                    image="busybox",
+                    securityContext=dict(runAsUser=101),
+                )
+            ]
+        ),
+    )
+    hidden_fields = ["spec.containers[0][securityContext][runAsUser]"]
+    expected = dict(
+        kind="Pod",
+        metadata=dict(name="foo"),
+        spec=dict(
+            containers=[
+                dict(
+                    name="containers",
+                    image="busybox",
+                )
+            ]
+        ),
+    )
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_using_jinja_syntax():
+    output = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=["0", "1", "2", "3"]
+    )
+    hidden_fields = ["data.2"]
+    expected = dict(kind="ConfigMap", metadata=dict(name="foo"), data=["0", "1", "3"])
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_remove_multiple_items_from_list():
+    output = dict(
+        kind="ConfigMap", metadata=dict(name="foo"), data=["0", "1", "2", "3"]
+    )
+    hidden_fields = ["data[0]", "data[2]"]
+    expected = dict(kind="ConfigMap", metadata=dict(name="foo"), data=["1", "3"])
+    assert hide_fields(output, hidden_fields) == expected
+
+
+def test_hide_dict_and_nested_dict():
+    output = {
+        "kind": "Pod",
+        "metadata": {
+            "labels": {
+                "control-plane": "controller-manager",
+                "pod-template-hash": "687b856498",
+            },
+            "annotations": {
+                "kubectl.kubernetes.io/default-container": "awx-manager",
+                "creationTimestamp": "2025-01-16T12:40:43Z",
+            },
+        },
+    }
+    hidden_fields = ["metadata.labels.pod-template-hash", "metadata.labels"]
+    expected = {
+        "kind": "Pod",
+        "metadata": {
+            "annotations": {
+                "kubectl.kubernetes.io/default-container": "awx-manager",
+                "creationTimestamp": "2025-01-16T12:40:43Z",
+            }
+        },
+    }
+    assert hide_fields(output, hidden_fields) == expected
+
+
+@pytest.mark.parametrize(
+    "hidden_fields,expected",
+    [
+        (
+            [
+                "data[0]",
+                "data[1]",
+                "metadata.annotation",
+                "metadata.annotation[0].name",
+            ],
+            {"data": {"0": None, "1": None}, "metadata": {"annotation": None}},
+        ),
+        (
+            [
+                "data[0]",
+                "data[1]",
+                "metadata.annotation[0].name",
+                "metadata.annotation",
+            ],
+            {"data": {"0": None, "1": None}, "metadata": {"annotation": None}},
+        ),
+        (
+            [
+                "data[0]",
+                "data[1]",
+                "data",
+                "metadata.annotation[0].name",
+                "metadata.annotation",
+            ],
+            {"data": None, "metadata": {"annotation": None}},
+        ),
+    ],
+)
+def test_build_hidden_field_tree(hidden_fields, expected):
+    assert build_hidden_field_tree(hidden_fields) == expected