# coding=utf-8 # ---------------------------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License file in the project root for license information. # ---------------------------------------------------------------------------------------------- from enum import Enum from itertools import groupby from knack.log import get_logger from kubernetes.client.models import ( V1APIResource, V1APIResourceList, ) from rich.padding import Padding from typing import Any, Dict, List, Optional, Tuple, Union from azext_edge.edge.providers.k8s.config_map import get_config_map from .check_manager import CheckManager from .display import process_value_color from ..common import COLOR_STR_FORMAT, PADDING_SIZE, ResourceOutputDetailLevel, ValidationResourceType from ...base import get_cluster_custom_api, get_namespaced_secret from ...edge_api import EdgeResourceApi from ....common import CheckTaskStatus, ResourceState # TODO: refactor logger = get_logger(__name__) def decorate_resource_status(status: str) -> str: from ....common import ResourceState return COLOR_STR_FORMAT.format(color=ResourceState.map_to_color(status), value=status) def enumerate_ops_service_resources( api_info: EdgeResourceApi, check_name: str, check_desc: str, as_list: bool = False, excluded_resources: Optional[List[str]] = None, ) -> Tuple[dict, dict]: resource_kind_map = {} target_api = api_info.as_str() check_manager = CheckManager(check_name=check_name, check_desc=check_desc) check_manager.add_target(target_name=target_api) api_resources: V1APIResourceList = get_cluster_custom_api(group=api_info.group, version=api_info.version) if not api_resources: check_manager.add_target_eval(target_name=target_api, status=CheckTaskStatus.error.value) missing_api_text = f"[bright_blue]{target_api}[/bright_blue] API resources [red]not[/red] detected." check_manager.add_display(target_name=target_api, display=Padding(missing_api_text, (0, 0, 0, 8))) return check_manager.as_dict(as_list), resource_kind_map api_header_display = Padding(f"[bright_blue]{target_api}[/bright_blue] API resources", (0, 0, 0, 8)) check_manager.add_display(target_name=target_api, display=api_header_display) for resource in api_resources.resources: r: V1APIResource = resource if excluded_resources and r.name in excluded_resources: continue if r.kind not in resource_kind_map: resource_kind_map[r.kind] = True check_manager.add_display( target_name=target_api, display=Padding(f"[cyan]{r.kind}[/cyan]", (0, 0, 0, 12)), ) check_manager.add_target_eval( target_name=target_api, status=CheckTaskStatus.success.value, value=list(resource_kind_map.keys()), ) return check_manager.as_dict(as_list), resource_kind_map def filter_resources_by_name( resources: List[dict], resource_name: str, ) -> List[dict]: from fnmatch import fnmatch if not resource_name: return resources resource_name = resource_name.lower() resources = [ resource for resource in resources if fnmatch(get_resource_metadata_property(resource, prop_name="name"), resource_name) ] return resources def filter_resources_by_namespace(resources: List[dict], namespace: str) -> List[dict]: return [resource for resource in resources if _get_namespace(resource) == namespace] def generate_target_resource_name(api_info: EdgeResourceApi, resource_kind: str) -> str: resource_plural = api_info._kinds[resource_kind] if api_info._kinds else f"{resource_kind}s" return f"{resource_plural}.{api_info.group}" def _get_namespace(resource): return get_resource_metadata_property(resource, prop_name="namespace") def get_resources_by_name( api_info: EdgeResourceApi, kind: Union[str, Enum], resource_name: str, namespace: str = None, ) -> List[dict]: resources: list = api_info.get_resources(kind=kind, namespace=namespace).get("items", []) resources = filter_resources_by_name(resources, resource_name) return resources def get_resources_grouped_by_namespace(resources: List[dict]): resources.sort(key=_get_namespace) return groupby(resources, key=_get_namespace) # get either name or namespace from resource that might be a object or a dict def get_resource_metadata_property(resource: Union[dict, Any], prop_name: str) -> Union[str, None]: if isinstance(resource, dict): return resource.get("metadata", {}).get(prop_name) return getattr(resource.metadata, prop_name, None) if hasattr(resource, "metadata") else None def process_dict_resource( check_manager: CheckManager, target_name: str, resource: dict, namespace: str, padding: int, prop_name: Optional[str] = None, ) -> None: if prop_name: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(f"{prop_name}:", (0, 0, 0, padding)) ) padding += PADDING_SIZE for key, value in resource.items(): if isinstance(value, dict): check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(f"{key}:", (0, 0, 0, padding)) ) process_dict_resource( check_manager=check_manager, target_name=target_name, resource=value, namespace=namespace, padding=padding + PADDING_SIZE, ) elif isinstance(value, list): if len(value) == 0: continue display_text = f"{key}:" check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(display_text, (0, 0, 0, padding)), ) process_list_resource( check_manager=check_manager, target_name=target_name, resource=value, namespace=namespace, padding=padding + PADDING_SIZE, ) else: display_text = f"{key}: " value_padding = padding if isinstance(value, str) and len(value) > 50: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(display_text, (0, 0, 0, padding)), ) value_padding += PADDING_SIZE display_text = "" display_text += process_value_color( check_manager=check_manager, target_name=target_name, key=key, value=value ) check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(display_text, (0, 0, 0, value_padding)), ) def process_list_resource( check_manager: CheckManager, target_name: str, resource: List[dict], namespace: str, padding: int ) -> None: for item in resource: name = "" if isinstance(item, dict): name = item.pop("name", None) # when name property exists, use name as header; if not, use property type and index as header if name: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(f"- name: [cyan]{name}[/cyan]", (0, 0, 0, padding)), ) else: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(f"- item {resource.index(item) + 1}", (0, 0, 0, padding)), ) if isinstance(item, dict): process_dict_resource( check_manager=check_manager, target_name=target_name, resource=item, namespace=namespace, padding=padding + 2, ) elif isinstance(item, str): check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(f"[cyan]{item}[/cyan]", (0, 0, 0, padding + 2)), ) def process_resource_properties( check_manager: CheckManager, detail_level: int, target_name: str, prop_value: Dict[str, Any], properties: Dict[str, Any], namespace: str, padding: tuple, ) -> None: if not prop_value: return for prop, display_name, verbose_only in properties: keys = prop.split(".") value = prop_value for key in keys: value = value.get(key) if value is None: break if prop == "descriptor": value = value if detail_level == ResourceOutputDetailLevel.verbose.value else value[:10] + "..." if verbose_only and detail_level != ResourceOutputDetailLevel.verbose.value: continue process_resource_property_by_type( check_manager=check_manager, target_name=target_name, properties=value, display_name=display_name, namespace=namespace, padding=padding, ) def process_resource_property_by_type( check_manager: CheckManager, target_name: str, properties: Any, display_name: str, namespace: str, padding: tuple, ) -> None: padding_left = padding[3] if isinstance(properties, list): if len(properties) == 0: return display_text = f"{display_name}:" check_manager.add_display(target_name=target_name, namespace=namespace, display=Padding(display_text, padding)) for property in properties: display_text = f"- {display_name} {properties.index(property) + 1}" check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(display_text, (0, 0, 0, padding_left + 2)), ) for prop, value in property.items(): display_text = f"{prop}: [cyan]{value}[/cyan]" check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(display_text, (0, 0, 0, padding_left + PADDING_SIZE)), ) elif isinstance(properties, str) or isinstance(properties, bool) or isinstance(properties, int): properties = str(properties) if properties else "undefined" if len(properties) < 50: display_text = f"{display_name}: [cyan]{properties}[/cyan]" else: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(f"{display_name}:", padding) ) display_text = f"[cyan]{properties}[/cyan]" padding = (0, 0, 0, padding_left + 4) check_manager.add_display(target_name=target_name, namespace=namespace, display=Padding(display_text, padding)) elif isinstance(properties, dict): display_text = f"{display_name}:" check_manager.add_display(target_name=target_name, namespace=namespace, display=Padding(display_text, padding)) for prop, value in properties.items(): display_text = f"{prop}: [cyan]{value}[/cyan]" check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(display_text, (0, 0, 0, padding_left + 2)), ) def validate_one_of_conditions( conditions: List[tuple], check_manager: CheckManager, eval_value: dict, namespace: str, target_name: str, padding: int, resource_name: Optional[str] = None, ) -> None: if len(conditions) == 1: return non_empty_conditions_count = len([condition for condition in conditions if condition[1]]) eval_status = CheckTaskStatus.success.value conditions_names = ", ".join([f"'{condition[0]}'" for condition in conditions]) if non_empty_conditions_count == 0: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding( f"One of {conditions_names} should be specified", (0, 0, 0, padding), ), ) eval_status = CheckTaskStatus.error.value elif non_empty_conditions_count > 1: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding( f"Only one of {conditions_names} should be specified", (0, 0, 0, padding), ), ) eval_status = CheckTaskStatus.error.value one_of_condition = f"oneOf({conditions_names})" check_manager.add_target_conditions(target_name=target_name, namespace=namespace, conditions=[one_of_condition]) check_manager.add_target_eval( target_name=target_name, namespace=namespace, status=eval_status, value=eval_value, resource_name=resource_name, ) def combine_statuses(status_list: List[str]): # lower case status list status_list = [status.lower() for status in status_list] final_status = "success" for status in status_list: if final_status == "success" and status not in ["running", "succeeded", "ok"]: final_status = status elif final_status in ["warning", "skipped", "warn", "starting", "recovering"] and status == "error": final_status = status return final_status def calculate_status(resource_state: str) -> str: return ResourceState.map_to_status(resource_state).value def process_custom_resource_status( check_manager: CheckManager, status: dict, target_name: str, namespace: str, resource_name: str, padding: int, detail_level: int = ResourceOutputDetailLevel.summary.value, ) -> None: runtime_status = status.get("runtimeStatus", {}) provisioning_status = status.get("provisioningStatus", {}) check_manager.add_target_conditions( target_name=target_name, conditions=["status"], namespace=namespace, ) if not runtime_status and not provisioning_status: # if no status for both runtime and provisioning, set status to error check_manager.add_target_eval( target_name=target_name, namespace=namespace, status=CheckTaskStatus.error.value, value={"status": status}, resource_name=resource_name, ) check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding("Status [red]not found[/red].", (0, 0, 0, padding)), ) return status_eval_value = {"status": status} status_list = [] if runtime_status: runtime_status_state = runtime_status.get("status") status_list.append(calculate_status(runtime_status_state)) if provisioning_status: provisioning_status_state = provisioning_status.get("status") status_list.append(calculate_status(provisioning_status_state)) # combine runtime and provisioning status status_eval_status = combine_statuses(status_list) check_manager.add_target_eval( target_name=target_name, namespace=namespace, status=status_eval_status, value=status_eval_value, resource_name=resource_name, ) if detail_level == ResourceOutputDetailLevel.summary.value: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding( f"Status {{{decorate_resource_status(status_eval_status)}}}.", (0, 0, 0, padding), ), ) else: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding("Status:", (0, 0, 0, padding)), ) for prop_name, prop_value in { "Provisioning Status": provisioning_status, "Runtime Status": runtime_status, }.items(): if prop_value: status_text = f"{prop_name} {{{decorate_resource_status(prop_value.get('status'))}}}." if detail_level == ResourceOutputDetailLevel.verbose.value: status_description = prop_value.get("description") or prop_value.get("output", {}).get("message") if status_description: status_text = status_text.replace(".", f", [cyan]{status_description}[/cyan].") check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(status_text, (0, 0, 0, padding + 4)), ) def validate_runtime_resource_ref(name: str, namespace: str, ref_type: ValidationResourceType) -> bool: ref_obj = None if ref_type == ValidationResourceType.secret: ref_obj = get_namespaced_secret(secret_name=name, namespace=namespace) elif ref_type == ValidationResourceType.configmap: ref_obj = get_config_map(name=name, namespace=namespace) else: raise ValueError(f"Unsupported ref type: {ref_type}") return bool(ref_obj) def get_valid_resource_names( api: EdgeResourceApi, kind: Union[Enum, str], namespace: Optional[str] = None ) -> List[str]: custom_objects = api.get_resources(kind=kind, namespace=namespace) if custom_objects: objects: List[dict] = custom_objects.get("items", []) if objects: return [get_resource_metadata_property(object, prop_name="name") for object in objects] return []