internet/kubernetes.core
3
0
Fork 0
mirror of https://github.com/ansible-collections/kubernetes.core.git synced 2026-03-26 21:33:02 +00:00
Code Issues Projects Releases Wiki Activity

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

Browse Source 
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
This commit is contained in:
Will Thames
2025-03-20 10:35:51 +00:00
committed by GitHub
parent 7cdf0d03f5
commit 9ec6912325
6 changed files with 409 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
changelogs/fragments/643-extend-hidden-fields.yaml Normal file
View File

@@ -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)

148
plugins/module_utils/k8s/service.py
View File

@@ -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"]:
return False, result
) == ["metadata"]:
# 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
ignored_keys = set(["generation", "resourceVersion"])
if not set(result["after"]["metadata"].keys()).issubset(ignored_keys):
return False, result
if not set(result["before"]["metadata"].keys()).issubset(ignored_keys):
return False, result
if set(result["after"]["metadata"].keys()).issubset(ignored_keys) and set(
result["before"]["metadata"].keys()
).issubset(ignored_keys):
return True, 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:
if not hidden_fields:
return definition
result = copy.deepcopy(definition)
for hidden_field in hidden_fields:
result = hide_field(result, hidden_field)
def hide_field_tree(hidden_field: str) -> List[str]:
result = []
key, rest = hide_field_split2(hidden_field)
result.append(key)
while rest:
key, rest = hide_field_split2(rest)
result.append(key)
return result
# hide_field is not hugely sophisticated and designed to cope
# with e.g. status or metadata.managedFields rather than e.g.
# spec.template.spec.containers[0].env[3].value
def hide_field(definition: dict, hidden_field: str) -> dict:
split = hidden_field.split(".", 1)
if split[0] in definition:
if len(split) == 2:
definition[split[0]] = hide_field(definition[split[0]], split[1])
else:
del definition[split[0]]
def build_hidden_field_tree(hidden_fields: List[str]) -> Dict[str, Any]:
"""Group hidden field targeting the same json key
Example:
Input: ['env[3]', 'env[0]']
Output: {'env': [0, 3]}
"""
output = {}
for hidden_field in hidden_fields:
current = output
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

3
plugins/modules/k8s.py
View File

@@ -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

3
plugins/modules/k8s_info.py
View File

@@ -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

17
tests/integration/targets/k8s_hide_fields/tasks/main.yml
View File

@@ -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}}"

264
tests/unit/module_utils/test_hide_fields.py Normal file
View File

@@ -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