# 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 typing import List from knack.log import get_logger from rich.padding import Padding from ...common import DEFAULT_DATAFLOW_PROFILE, CheckTaskStatus, ResourceState from ..base import get_namespaced_pods_by_prefix from ..edge_api.dataflow import DATAFLOW_API_V1, DataflowResourceKinds from ..support.dataflow import DATAFLOW_NAME_LABEL, DATAFLOW_OPERATOR_PREFIX, DATAFLOW_PROFILE_POD_PREFIX from .base import ( CheckManager, check_post_deployment, get_resources_by_name, get_resources_grouped_by_namespace, ) from .base.display import basic_property_display, colorize_string from .base.pod import evaluate_pod_health from .base.resource import filter_resources_by_name from .common import ( DEFAULT_PADDING, DEFAULT_PROPERTY_DISPLAY_COLOR, PADDING_SIZE, CoreServiceResourceKinds, DataFlowEndpointAuthenticationType, DataflowEndpointType, DataflowOperationType, ResourceOutputDetailLevel, ) logger = get_logger(__name__) PADDING = DEFAULT_PADDING INNER_PADDING = PADDING + PADDING_SIZE dataflow_api_check_name = "enumerateDataflowApi" dataflow_api_check_desc = "Enumerate Dataflow API resources" dataflow_runtime_check_name = "evalCoreServiceRuntime" dataflow_runtime_check_desc = "Evaluate Dataflow core service" dataflows_check_name = "evalDataflows" dataflows_check_desc = "Evaluate Dataflows" dataflow_endpoint_check_name = "evalDataflowEndpoints" dataflow_endpoint_check_desc = "Evaluate Dataflow Endpoints" dataflow_profile_check_name = "evalDataflowProfiles" dataflow_profile_check_desc = "Evaluate Dataflow Profiles" dataflow_target = "dataflows.connectivity.iotoperations.azure.com" dataflow_endpoint_target = "dataflowendpoints.connectivity.iotoperations.azure.com" dataflow_profile_target = "dataflowprofiles.connectivity.iotoperations.azure.com" valid_source_endpoint_types = [DataflowEndpointType.kafka.value, DataflowEndpointType.mqtt.value] def _process_dataflow_resource_status( check_manager: CheckManager, target_name: str, namespace: str, status: dict, resource_name: str, resource_kind: str, detail_level: int, padding: int, ): # status is overall optional, no evals if no status if not status: return inner_padding = padding + PADDING_SIZE provisioning_status_enum = CheckTaskStatus.skipped runtime_status_enum = CheckTaskStatus.skipped # provisioning Status (always optional) provisioning_status = status.get("provisioningStatus") if provisioning_status: # status - summary provisioning_status_status = provisioning_status.get("status") # override enum value with errors if they exist provisioning_status_error = provisioning_status.get("error") provisioning_status_failure_cause = provisioning_status.get("failureCause") provisioning_status_enum = ( CheckTaskStatus.error if provisioning_status_error or provisioning_status_failure_cause else ResourceState.map_to_status(provisioning_status_status) ) provisioning_status_display = f"Provisioning Status: {{{colorize_string(color=provisioning_status_enum.color, value=provisioning_status_status)}}}" # if not summary, display output / message if detail_level > ResourceOutputDetailLevel.summary.value: provisioning_status_output = provisioning_status.get("output", {}) # output object has no schema, but expecting "message" property to have details provisioning_status_output_message = provisioning_status_output.get("message") # if no "message" property, just use the entire object as output provisioning_status_output_display = provisioning_status_output_message or provisioning_status_output provisioning_status_display += f" - {colorize_string(provisioning_status_output_display)}" # status display check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(provisioning_status_display, (0, 0, 0, padding)), ) # display error, failure cause if provisioning_status_error: message = provisioning_status_error.get("message") code = provisioning_status_error.get("code") error_text = f"{code}: {message}" error_display = f"Error: {{{colorize_string(color=CheckTaskStatus.error.color, value=error_text)}}}" check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(error_display, (0, 0, 0, padding)) ) # show failure cause if provisioning_status_failure_cause: check_manager.add_display( target_name=target_name, namespace=namespace, display=basic_property_display( label="Failure Cause", value=provisioning_status_failure_cause, padding=inner_padding ), ) # logErrors on detail level 1 provisioning_status_log_errors = provisioning_status.get("logErrors") if detail_level > ResourceOutputDetailLevel.summary.value and provisioning_status_log_errors: check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding("Log Errors:", (0, 0, 0, inner_padding)), ) # runtime status (required for profiles) runtime_status = status.get("runtimeStatus", {}) if runtime_status: # calculate status runtime_level = runtime_status.get("level") runtime_status_enum = ResourceState.map_to_status(runtime_level) description = runtime_status.get("description") # create display runtime_status_display = ( f"Runtime Status: {{{colorize_string(color=runtime_status_enum.color, value=runtime_status.get('level'))}}}" ) if description and detail_level > ResourceOutputDetailLevel.summary.value: runtime_status_display += f" - {colorize_string(description)}" check_manager.add_display( target_name=target_name, namespace=namespace, display=Padding(runtime_status_display, (0, 0, 0, padding)), ) elif resource_kind == DataflowResourceKinds.DATAFLOWPROFILE.value: runtime_status_enum = CheckTaskStatus.error # add evals if necessary if provisioning_status_enum != CheckTaskStatus.skipped: # todo - add error object to eval value? status_obj = {"status.provisioningStatus.status": provisioning_status_status} if provisioning_status_error: status_obj["status.provisioningStatus.error"] = provisioning_status_error check_manager.add_target_eval( target_name=target_name, namespace=namespace, status=provisioning_status_enum.value, resource_name=resource_name, resource_kind=resource_kind, value=status_obj, ) if runtime_status_enum != CheckTaskStatus.skipped: check_manager.add_target_eval( target_name=target_name, namespace=namespace, status=runtime_status_enum.value, resource_name=resource_name, resource_kind=resource_kind, value={"status.runtimeStatus.level": runtime_status.get("level")}, ) def _process_dataflow_sourcesettings( check_manager: CheckManager, target: str, namespace: str, dataflow_name: str, endpoints: List[dict], operation: dict, detail_level: int, padding: int, ): inner_padding = padding + PADDING_SIZE settings = operation.get("sourceSettings", {}) # show endpoint ref # TODO - lots of shared code for validating source/dest endpoints, consider refactoring endpoint_ref = settings.get("endpointRef") # currently we are only looking for endpoint references in the same namespace # duplicate names should not exist, so check the first endpoint that matches the name ref endpoint_ref_string = "not found" endpoint_ref_status = endpoint_type_status = CheckTaskStatus.error endpoint_type_status_string = "invalid" found_endpoint = next( (endpoint for endpoint in endpoints if "name" in endpoint and endpoint["name"] == endpoint_ref), None ) endpoint_type = found_endpoint["type"] if found_endpoint and "type" in found_endpoint else None if found_endpoint: endpoint_ref_status = CheckTaskStatus.success endpoint_ref_string = "detected" endpoint_type_valid = endpoint_type and endpoint_type.lower() in valid_source_endpoint_types endpoint_type_status = CheckTaskStatus.success if endpoint_type_valid else CheckTaskStatus.error endpoint_type_status_string = "valid" if endpoint_type_valid else f"has invalid type: {endpoint_type}" endpoint_ref_display = colorize_string(value=endpoint_ref_string, color=endpoint_ref_status.color) endpoint_validity_display = colorize_string( color=endpoint_type_status.color, value=endpoint_type_status_string ) # valid endpoint ref eval check_manager.add_target_eval( target_name=target, namespace=namespace, status=endpoint_ref_status.value, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"spec.operations[*].sourceSettings.endpointRef": endpoint_ref}, ) # valid source endpoint type eval check_manager.add_target_eval( target_name=target, namespace=namespace, status=endpoint_type_status.value, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"ref(spec.operations[*].sourceSettings.endpointRef).endpointType": endpoint_type}, ) if detail_level > ResourceOutputDetailLevel.summary.value: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("\nSource:", (0, 0, 0, padding)) ) endpoint_name_display = f"{{{colorize_string(value=endpoint_ref)}}}" check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"Dataflow Endpoint {endpoint_name_display} {endpoint_ref_display}, {endpoint_validity_display}", (0, 0, 0, padding + PADDING_SIZE), ), ) elif not found_endpoint or not endpoint_type_valid: check_manager.add_display( target_name=target, namespace=namespace, display=Padding( "[red]Invalid source endpoint reference[/red]", (0, 0, 0, padding - PADDING_SIZE) ), ) if detail_level > ResourceOutputDetailLevel.detail.value: for label, key in [ # TODO - validate asset ref / colorize ("DeviceRegistry Asset Reference", "assetRef"), ("Schema Reference", "schemaRef"), ("Serialization Format", "serializationFormat"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=inner_padding), ) # data source strings - not on summary if detail_level > ResourceOutputDetailLevel.summary.value: data_sources = settings.get("dataSources", []) if data_sources: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Data Sources:", (0, 0, 0, inner_padding)), ) for data_source in data_sources: check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"- {colorize_string(data_source)}", (0, 0, 0, inner_padding + 2)), ) def _process_dataflow_transformationsettings( check_manager: CheckManager, target: str, namespace: str, resource: dict, detail_level: int, padding: int ): settings = resource.get("builtInTransformationSettings", {}) # only show details on non-summary if detail_level > ResourceOutputDetailLevel.summary.value: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("\nBuilt-In Transformation:", (0, 0, 0, padding)), ) padding += PADDING_SIZE inner_padding = padding + PADDING_SIZE def _process_inputs(inputs: List[str]): if inputs: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Inputs:", (0, 0, 0, inner_padding)), ) for input in inputs: check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"- {colorize_string(input)}", (0, 0, 0, inner_padding + 2)), ) # extra properties for datasets_label, key in [ ("Schema Reference", "schemaRef"), ("Serialization Format", "serializationFormat"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=datasets_label, value=val, padding=padding), ) # only show datasets, filters, maps on verbose if detail_level > ResourceOutputDetailLevel.detail.value: # datasets datasets = settings.get("datasets", []) if datasets: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Datasets:", (0, 0, 0, padding)) ) for dataset in datasets: for label, key in [ ("Description", "description"), ("Key", "key"), ("Expression", "expression"), ("Schema", "schemaRef"), ]: val = dataset.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=inner_padding), ) inputs = dataset.get("inputs", []) _process_inputs(inputs) # filters filters = settings.get("filter", []) if filters: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Filters:", (0, 0, 0, padding)) ) for filter in filters: for datasets_label, key in [ ("Description", "description"), ("Expression", "expression"), ("Operation Type", "type"), ]: val = filter.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=datasets_label, value=val, padding=padding), ) inputs = filter.get("inputs", []) _process_inputs(inputs) # maps maps = settings.get("map", []) if maps: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Maps:", (0, 0, 0, padding)) ) for map in maps: for label, key in [ ("Description", "description"), ("Expression", "expression"), ("Output", "output"), ("Transformation Type", "type"), ]: val = map.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=inner_padding), ) inputs = map.get("inputs", []) _process_inputs(inputs) def _process_dataflow_destinationsettings( check_manager: CheckManager, target: str, namespace: str, dataflow_name: str, endpoints: List[dict], operation: dict, detail_level: int, padding: int, ): settings = operation.get("destinationSettings", {}) if detail_level > ResourceOutputDetailLevel.summary.value: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("\nDestination:", (0, 0, 0, padding)) ) endpoint_ref = settings.get("endpointRef") # currently we are only looking for endpoint references in the same namespace # duplicate names should not exist, so check the first endpoint that matches the name ref endpoint_match = next( (endpoint for endpoint in endpoints if "name" in endpoint and endpoint["name"] == endpoint_ref), None ) endpoint_validity = "valid" endpoint_status = CheckTaskStatus.success if not endpoint_match: endpoint_validity = "not found" endpoint_status = CheckTaskStatus.error # valid endpoint ref eval check_manager.add_target_eval( target_name=target, namespace=namespace, status=endpoint_status.value, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"spec.operations[*].destinationSettings.endpointRef": endpoint_ref}, ) # show dataflow endpoint ref on detail if detail_level > ResourceOutputDetailLevel.summary.value: padding += PADDING_SIZE endpoint_name_display = f"{{{colorize_string(value=endpoint_ref)}}}" endpoint_validity_display = colorize_string(color=endpoint_status.color, value=endpoint_validity) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"Dataflow Endpoint {endpoint_name_display} {endpoint_validity_display}", (0, 0, 0, padding), ), ) elif not endpoint_match: check_manager.add_display( target_name=target, namespace=namespace, display=Padding( "[red]Invalid destination endpoint reference[/red]", (0, 0, 0, padding - PADDING_SIZE) ), ) # only show destination on verbose if detail_level > ResourceOutputDetailLevel.detail.value: for label, key in [ ("Data Destination", "dataDestination"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) def _process_endpoint_authentication( endpoint_settings: dict, check_manager: CheckManager, target: str, namespace: str, padding: int, detail_level: int ) -> None: auth_property_dict = { DataFlowEndpointAuthenticationType.access_token.value: { "key": "accessTokenSettings", "displays": [ ("Secret Reference", "secretRef"), ], }, DataFlowEndpointAuthenticationType.system_assigned.value: { "key": "systemAssignedManagedIdentitySettings", "displays": [ ("Audience", "audience"), ], }, DataFlowEndpointAuthenticationType.user_assigned.value: { "key": "userAssignedManagedIdentitySettings", "displays": [ ("Client ID", "clientId"), ("Scope", "scope"), ("Tenant ID", "tenantId"), ], }, DataFlowEndpointAuthenticationType.x509.value: { "key": "x509CertificateSettings", "displays": [ ("Secret Ref", "secretRef"), ], }, DataFlowEndpointAuthenticationType.service_account_token.value: { "key": "serviceAccountTokenSettings", "displays": [ ("Audience", "audience"), ], }, DataFlowEndpointAuthenticationType.sasl.value: { "key": "saslSettings", "displays": [ ("Type", "saslType"), ("Secret Ref", "secretRef"), ], }, DataFlowEndpointAuthenticationType.anonymous.value: { "key": "anonymousSettings", "displays": [], }, } auth: dict = endpoint_settings.get("authentication", {}) auth_method = auth.get("method") # display unkown auth method if auth_method not in auth_property_dict: check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"[red]Unknown authentication method: {auth_method}", (0, 0, 0, padding)), ) return # display auth method check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label="Authentication Method", value=auth_method, padding=padding), ) # show details for various auth methods if detail_level > ResourceOutputDetailLevel.detail.value: auth_properties: dict = auth_property_dict.get(auth_method, {}) auth_settings_key = auth_properties.get("key") auth_obj = auth.get(auth_settings_key) if auth_obj: for label, key in auth_properties.get("displays", []): val = auth_obj.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding + PADDING_SIZE), ) def _process_endpoint_TLS( tls_settings: dict, check_manager: CheckManager, target: str, namespace: str, padding: int ) -> None: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("TLS:", (0, 0, 0, padding)), ) for label, key in [ ("Mode", "mode"), ("Trusted CA ConfigMap", "trustedCaCertificateConfigMapRef"), ]: # TODO - validate ref? val = tls_settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=(padding + PADDING_SIZE)), ) def _process_endpoint_mqttsettings( check_manager: CheckManager, target: str, namespace: str, spec: dict, detail_level: int, padding: int ) -> None: settings = spec.get("mqttSettings", {}) for label, key in [ ("MQTT Host", "host"), ("Protocol", "protocol"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # endpoint authentication details _process_endpoint_authentication( endpoint_settings=settings, check_manager=check_manager, target=target, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) if detail_level > ResourceOutputDetailLevel.detail.value: for label, key in [ ("Cloud Event Attributes", "cloudEventAttributes"), ("Client ID Prefix", "clientIdPrefix"), ("Keep Alive (s)", "keepAliveSeconds"), ("Max Inflight Messages", "maxInflightMessages"), ("QOS", "qos"), ("Retain", "retain"), ("Session Expiry (s)", "sessionExpirySeconds"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # TLS tls = settings.get("tls", {}) if tls: _process_endpoint_TLS( tls_settings=tls, check_manager=check_manager, target=target, namespace=namespace, padding=padding, ) def _process_endpoint_kafkasettings( check_manager: CheckManager, target: str, namespace: str, spec: dict, detail_level: int, padding: int ) -> None: inner_padding = padding + PADDING_SIZE settings = spec.get("kafkaSettings", {}) for label, key in [ ("Kafka Host", "host"), ("Consumer Group ID", "consumerGroupId"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # endpoint authentication details _process_endpoint_authentication( endpoint_settings=settings, check_manager=check_manager, target=target, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) if detail_level > ResourceOutputDetailLevel.detail.value: # extra properties for label, key in [ ("Cloud Event Attributes", "cloudEventAttributes"), ("Compression", "compression"), ("Copy MQTT Properties", "copyMqttProperties"), ("Acks", "kafkaAcks"), ("Partition Strategy", "partitionStrategy"), ]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # TLS tls = settings.get("tls", {}) if tls: _process_endpoint_TLS( tls_settings=tls, check_manager=check_manager, target=target, namespace=namespace, padding=padding, ) # batching batching = settings.get("batching", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Batching:", (0, 0, 0, padding)), ) for label, key in [ ("Latency (ms)", "latencyMs"), ("Max Bytes", "maxBytes"), ("Max Messages", "maxMessages"), ("Mode", "mode"), ]: val = batching.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=inner_padding), ) def _process_endpoint_fabriconelakesettings( check_manager: CheckManager, target: str, namespace: str, spec: dict, detail_level: int, padding: int ) -> None: settings = spec.get("fabricOneLakeSettings", {}) for label, key in [("Fabric Host", "host"), ("Path Type", "oneLakePathType")]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # endpoint authentication details _process_endpoint_authentication( endpoint_settings=settings, check_manager=check_manager, target=target, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) if detail_level > ResourceOutputDetailLevel.detail.value: names = settings.get("names", {}) for label, key in [ ("Lakehouse Name", "lakehouseName"), ("Workspace Name", "workspaceName"), ]: val = names.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) batching = settings.get("batching", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Batching:", (0, 0, 0, padding)), ) padding += PADDING_SIZE for label, key in [ ("Latency (s)", "latencySeconds"), ("Max Messages", "maxMessages"), ]: val = batching.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) def _process_endpoint_datalakestoragesettings( check_manager: CheckManager, target: str, namespace: str, spec: dict, detail_level: int, padding: int ) -> None: settings = spec.get("datalakeStorageSettings", {}) for label, key in [("DataLake Host", "host")]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # endpoint authentication details _process_endpoint_authentication( endpoint_settings=settings, check_manager=check_manager, target=target, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) if detail_level > ResourceOutputDetailLevel.detail.value: batching = settings.get("batching", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Batching:", (0, 0, 0, padding)), ) padding += PADDING_SIZE for label, key in [ ("Latency (s)", "latencySeconds"), ("Max Messages", "maxMessages"), ]: val = batching.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) def _process_endpoint_dataexplorersettings( check_manager: CheckManager, target: str, namespace: str, spec: dict, detail_level: int, padding: int ) -> None: settings = spec.get("dataExplorerSettings", {}) for label, key in [("Database Name", "database"), ("Data Explorer Host", "host")]: val = settings.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) # endpoint authentication details _process_endpoint_authentication( endpoint_settings=settings, check_manager=check_manager, target=target, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) if detail_level > ResourceOutputDetailLevel.detail.value: batching = settings.get("batching", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Batching:", (0, 0, 0, padding)), ) padding += PADDING_SIZE for label, key in [ ("Latency (s)", "latencySeconds"), ("Max Messages", "maxMessages"), ]: val = batching.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display(label=label, value=val, padding=padding), ) def _process_endpoint_localstoragesettings( check_manager: CheckManager, target: str, namespace: str, spec: dict, detail_level: int, padding: int ) -> None: # TODO - validate reference settings = spec.get("localStorageSettings", {}) persistent_volume_claim = settings.get("persistentVolumeClaimRef") check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"Persistent Volume Claim: {persistent_volume_claim}", (0, 0, 0, padding)), ) # endpoint authentication details _process_endpoint_authentication( endpoint_settings=settings, check_manager=check_manager, target=target, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) def check_dataflows_deployment( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_kinds: List[str] = None, resource_name: str = None, ) -> List[dict]: evaluate_funcs = { CoreServiceResourceKinds.RUNTIME_RESOURCE: evaluate_core_service_runtime, DataflowResourceKinds.DATAFLOWPROFILE: evaluate_dataflow_profiles, DataflowResourceKinds.DATAFLOWENDPOINT: evaluate_dataflow_endpoints, DataflowResourceKinds.DATAFLOW: evaluate_dataflows, } return check_post_deployment( api_info=DATAFLOW_API_V1, check_name=dataflow_api_check_name, check_desc=dataflow_api_check_desc, evaluate_funcs=evaluate_funcs, as_list=as_list, detail_level=detail_level, resource_kinds=resource_kinds, resource_name=resource_name, ) def evaluate_core_service_runtime( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_name: str = None, ): check_manager = CheckManager( check_name=dataflow_runtime_check_name, check_desc=dataflow_runtime_check_desc, ) operators = get_namespaced_pods_by_prefix( prefix=DATAFLOW_OPERATOR_PREFIX, namespace="", label_selector=DATAFLOW_NAME_LABEL, ) if resource_name: operators = filter_resources_by_name( resources=operators, resource_name=resource_name, ) if not operators: check_manager.add_target(target_name=CoreServiceResourceKinds.RUNTIME_RESOURCE.value) check_manager.add_display( target_name=CoreServiceResourceKinds.RUNTIME_RESOURCE.value, display=Padding("Unable to fetch pods.", (0, 0, 0, PADDING)), ) for namespace, pods in get_resources_grouped_by_namespace(operators): check_manager.add_target( target_name=CoreServiceResourceKinds.RUNTIME_RESOURCE.value, namespace=namespace, ) check_manager.add_display( target_name=CoreServiceResourceKinds.RUNTIME_RESOURCE.value, namespace=namespace, display=Padding( f"Dataflow operator in namespace {{[purple]{namespace}[/purple]}}", (0, 0, 0, PADDING), ), ) evaluate_pod_health( check_manager=check_manager, target=CoreServiceResourceKinds.RUNTIME_RESOURCE.value, pods=pods, namespace=namespace, padding=PADDING + 2, detail_level=detail_level, ) return check_manager.as_dict(as_list) def evaluate_dataflows( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_name: str = None, ): check_manager = CheckManager( check_name=dataflows_check_name, check_desc=dataflows_check_desc, ) all_dataflows = get_resources_by_name( api_info=DATAFLOW_API_V1, kind=DataflowResourceKinds.DATAFLOW, resource_name=resource_name, ) target = dataflow_target # No dataflows - skip if not all_dataflows: no_dataflows_text = "No Dataflow resources detected in any namespace." check_manager.add_target(target_name=target) check_manager.add_target_eval( target_name=target, status=CheckTaskStatus.skipped.value, value={"dataflows": no_dataflows_text} ) check_manager.add_display( target_name=target, display=Padding(no_dataflows_text, (0, 0, 0, PADDING)), ) return check_manager.as_dict(as_list=as_list) for namespace, dataflows in get_resources_grouped_by_namespace(all_dataflows): check_manager.add_target(target_name=target, namespace=namespace) check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"Dataflows in namespace {{[purple]{namespace}[/purple]}}", (0, 0, 0, PADDING)), ) # conditions check_manager.add_target_conditions( target_name=target, namespace=namespace, conditions=[ # at least a source and destination operation "len(spec.operations)<=3", # valid source endpoint "spec.operations[*].sourceSettings.endpointRef", "ref(spec.operations[*].sourceSettings.endpointRef).endpointType in ('kafka','mqtt')", # valid destination endpoint "spec.operations[*].destinationSettings.endpointRef", # single source/destination "len(spec.operations[*].sourceSettings)==1", "len(spec.operations[*].destinationSettings)==1", ], ) # profile names for reference lookup all_profiles = get_resources_by_name( api_info=DATAFLOW_API_V1, kind=DataflowResourceKinds.DATAFLOWPROFILE, namespace=namespace, resource_name=None, ) profile_names = {profile.get("metadata", {}).get("name") for profile in all_profiles} all_endpoints = get_resources_by_name( api_info=DATAFLOW_API_V1, kind=DataflowResourceKinds.DATAFLOWENDPOINT, namespace=namespace, resource_name=None, ) endpoints = [ { "name": endpoint.get("metadata", {}).get("name"), "type": endpoint.get("spec", {}).get("endpointType"), } for endpoint in all_endpoints ] for dataflow in list(dataflows): spec = dataflow.get("spec", {}) dataflow_name = dataflow.get("metadata", {}).get("name") mode = spec.get("mode") mode_lower = str(mode).lower() if mode else "unknown" dataflow_enabled = mode_lower == "enabled" mode_display = colorize_string(value=mode_lower) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"\n- Dataflow {{{colorize_string(value=dataflow_name)}}} is {mode_display}", (0, 0, 0, PADDING), ), ) # if dataflow is disabled, skip evaluations and displays if not dataflow_enabled: check_manager.add_target_eval( target_name=target, namespace=namespace, status=CheckTaskStatus.skipped.value, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"spec.mode": mode}, ) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( colorize_string( value=f"{CheckTaskStatus.skipped.emoji} Skipping evaluation of disabled dataflow", color=CheckTaskStatus.skipped.color, ), (0, 0, 0, PADDING + 2), ), ) continue status = dataflow.get("status", {}) _process_dataflow_resource_status( check_manager=check_manager, target_name=target, namespace=namespace, status=status, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, detail_level=detail_level, padding=INNER_PADDING, ) profile_ref = spec.get("profileRef") # profileRef is optional, only show an error if it exists but is invalid if profile_ref: profile_ref_status = ( CheckTaskStatus.error if profile_ref not in profile_names else CheckTaskStatus.success ) # valid profileRef eval check_manager.add_target_eval( target_name=target, namespace=namespace, status=profile_ref_status.value, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"spec.profileRef": profile_ref}, ) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"Dataflow Profile: {{{colorize_string(color=profile_ref_status.color, value=profile_ref)}}}", (0, 0, 0, INNER_PADDING), ), ) if profile_ref_status == CheckTaskStatus.error: check_manager.add_display( target_name=target, namespace=namespace, display=Padding( colorize_string( color=profile_ref_status.color, value="Invalid Dataflow Profile reference" ), (0, 0, 0, INNER_PADDING), ), ) operations = spec.get("operations", []) # check operations count operations_status = CheckTaskStatus.success.value if not operations or not (2 <= len(operations) <= 3): operations_status = CheckTaskStatus.error.value check_manager.add_target_eval( target_name=target, namespace=namespace, status=operations_status, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"len(operations)": len(operations)}, ) if operations and detail_level > ResourceOutputDetailLevel.summary.value: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Operations:", (0, 0, 0, INNER_PADDING)), ) operation_padding = INNER_PADDING + PADDING_SIZE sources = destinations = 0 for operation in operations: op_type = operation.get("operationType", "").lower() if op_type == DataflowOperationType.source.value: sources += 1 _process_dataflow_sourcesettings( check_manager=check_manager, target=target, namespace=namespace, dataflow_name=dataflow_name, endpoints=endpoints, operation=operation, detail_level=detail_level, padding=operation_padding, ) elif op_type == DataflowOperationType.builtin_transformation.value: _process_dataflow_transformationsettings( check_manager=check_manager, target=target, namespace=namespace, resource=operation, detail_level=detail_level, padding=operation_padding, ) elif op_type == DataflowOperationType.destination.value: destinations += 1 _process_dataflow_destinationsettings( check_manager=check_manager, target=target, namespace=namespace, dataflow_name=dataflow_name, endpoints=endpoints, operation=operation, detail_level=detail_level, padding=operation_padding, ) # eval source amount (1) sources_status = destinations_status = CheckTaskStatus.success.value if sources != 1: sources_status = CheckTaskStatus.error.value message = ( "Missing source operation" if sources == 0 else f"Too many source operations: {sources}" ) check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"[red]{message}[/red]", (0, 0, 0, INNER_PADDING)), ) check_manager.add_target_eval( target_name=target, namespace=namespace, status=sources_status, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"len(spec.operations[*].sourceSettings)": sources}, ) if destinations != 1: destinations_status = CheckTaskStatus.error.value message = ( "Missing destination operation" if destinations == 0 else f"Too many destination operations: {destinations}" ) check_manager.add_display( target_name=target, namespace=namespace, display=Padding(f"[red]{message}[/red]", (0, 0, 0, INNER_PADDING)), ) check_manager.add_target_eval( target_name=target, namespace=namespace, status=destinations_status, resource_name=dataflow_name, resource_kind=DataflowResourceKinds.DATAFLOW.value, value={"len(spec.operations[*].destinationSettings)": destinations}, ) return check_manager.as_dict(as_list=as_list) def evaluate_dataflow_endpoints( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_name: str = None, ): check_manager = CheckManager( check_name=dataflow_endpoint_check_name, check_desc=dataflow_endpoint_check_desc, ) all_endpoints = get_resources_by_name( api_info=DATAFLOW_API_V1, kind=DataflowResourceKinds.DATAFLOWENDPOINT, resource_name=resource_name, ) target = dataflow_endpoint_target if not all_endpoints: no_endpoints_text = "No Dataflow Endpoints detected in any namespace." check_manager.add_target(target_name=target) check_manager.add_target_eval( target_name=target, status=CheckTaskStatus.skipped.value, value={"endpoints": no_endpoints_text} ) check_manager.add_display( target_name=target, display=Padding(no_endpoints_text, (0, 0, 0, PADDING)), ) return check_manager.as_dict(as_list=as_list) for namespace, endpoints in get_resources_grouped_by_namespace(all_endpoints): check_manager.add_target(target_name=target, namespace=namespace, conditions=["spec.endpointType"]) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"Dataflow Endpoints in namespace {{[purple]{namespace}[/purple]}}", (0, 0, 0, PADDING) ), ) for endpoint in list(endpoints): spec = endpoint.get("spec", {}) endpoint_name = endpoint.get("metadata", {}).get("name") endpoint_type = spec.get("endpointType") valid_endpoint_type = endpoint_type and endpoint_type.lower() in DataflowEndpointType.list() check_manager.add_target_eval( target_name=target, namespace=namespace, status=CheckTaskStatus.success.value if valid_endpoint_type else CheckTaskStatus.error.value, resource_name=endpoint_name, resource_kind=DataflowResourceKinds.DATAFLOWENDPOINT.value, value={"spec.endpointType": endpoint_type}, ) endpoint_string = f"Endpoint {{{colorize_string(value=endpoint_name)}}}" detected_string = colorize_string(color="green", value="detected") type_string = f"type: {colorize_string(color=DEFAULT_PROPERTY_DISPLAY_COLOR if valid_endpoint_type else 'red', value=endpoint_type)}" check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"\n- {endpoint_string} {detected_string}, {type_string}", (0, 0, 0, PADDING), ), ) status = endpoint.get("status", {}) _process_dataflow_resource_status( check_manager=check_manager, target_name=target, namespace=namespace, status=status, resource_name=endpoint_name, resource_kind=DataflowResourceKinds.DATAFLOWENDPOINT.value, detail_level=detail_level, padding=INNER_PADDING, ) # endpoint details at higher detail levels if detail_level > ResourceOutputDetailLevel.summary.value: endpoint_processor_dict = { DataflowEndpointType.mqtt.value: _process_endpoint_mqttsettings, DataflowEndpointType.kafka.value: _process_endpoint_kafkasettings, DataflowEndpointType.fabric_onelake.value: _process_endpoint_fabriconelakesettings, DataflowEndpointType.datalake.value: _process_endpoint_datalakestoragesettings, DataflowEndpointType.data_explorer.value: _process_endpoint_dataexplorersettings, DataflowEndpointType.local_storage.value: _process_endpoint_localstoragesettings, } # process endpoint settings if endpoint_type and endpoint_type.lower() in endpoint_processor_dict: endpoint_processor_dict[endpoint_type.lower()]( check_manager=check_manager, target=target, namespace=namespace, spec=spec, detail_level=detail_level, padding=INNER_PADDING, ) else: check_manager.add_display( target_name=target, namespace=namespace, display=Padding( colorize_string(color="red", value=f"Unknown endpoint type: {endpoint_type}"), (0, 0, 0, INNER_PADDING), ), ) return check_manager.as_dict(as_list=as_list) def evaluate_dataflow_profiles( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_name: str = None, ): check_manager = CheckManager( check_name=dataflow_profile_check_name, check_desc=dataflow_profile_check_desc, ) target = dataflow_profile_target all_profiles = get_resources_by_name( api_info=DATAFLOW_API_V1, kind=DataflowResourceKinds.DATAFLOWPROFILE, resource_name=resource_name, ) if not all_profiles: no_profiles_text = "No Dataflow Profiles detected in any namespace." check_manager.add_target(target_name=target) # if we may have manually filtered out the default profile by input, skip instead of warn default_profile_status = CheckTaskStatus.skipped if resource_name else CheckTaskStatus.warning check_manager.add_target_eval( target_name=target, status=default_profile_status.value, value={"profiles": no_profiles_text} ) check_manager.add_display( target_name=target, display=Padding(no_profiles_text, (0, 0, 0, PADDING)), ) return check_manager.as_dict(as_list=as_list) for namespace, profiles in get_resources_grouped_by_namespace(all_profiles): check_manager.add_target( target_name=target, namespace=namespace, conditions=["spec.instanceCount", f"[*].metadata.name=='{DEFAULT_DATAFLOW_PROFILE}'"], ) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"Dataflow Profiles in namespace {{[purple]{namespace}[/purple]}}", (0, 0, 0, PADDING) ), ) # warn if no default dataflow profile (unless possibly filtered) default_profile_status = CheckTaskStatus.skipped if resource_name else CheckTaskStatus.warning for profile in list(profiles): profile_name = profile.get("metadata", {}).get("name") # check for default dataflow profile if profile_name == DEFAULT_DATAFLOW_PROFILE: default_profile_status = CheckTaskStatus.success spec = profile.get("spec", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding( f"\n- Profile {{{colorize_string(value=profile_name)}}} {colorize_string(color='green', value='detected')}", (0, 0, 0, PADDING), ), ) # evaluate status status = profile.get("status", {}) _process_dataflow_resource_status( check_manager=check_manager, target_name=target, namespace=namespace, status=status, resource_name=profile_name, resource_kind=DataflowResourceKinds.DATAFLOWPROFILE.value, detail_level=detail_level, padding=INNER_PADDING, ) # instance count instance_count = spec.get("instanceCount") has_instances = instance_count is not None and int(instance_count) >= 0 instance_status = CheckTaskStatus.success if has_instances else CheckTaskStatus.error check_manager.add_target_eval( target_name=target, namespace=namespace, status=instance_status.value, resource_name=profile_name, resource_kind=DataflowResourceKinds.DATAFLOWPROFILE.value, value={"spec.instanceCount": instance_count}, ) if has_instances: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display( label="Instance count", value=instance_count, color=instance_status.color, padding=INNER_PADDING, ), ) else: check_manager.add_display( target_name=target, namespace=namespace, display=Padding("[red]No instance count set[/red]", (0, 0, 0, INNER_PADDING)), ) # diagnostics on higher detail levels if detail_level > ResourceOutputDetailLevel.summary.value: log_padding = PADDING + PADDING_SIZE log_inner_padding = log_padding + PADDING_SIZE diagnostics = spec.get("diagnostics", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Diagnostic Logs:", (0, 0, 0, log_padding)), ) # diagnostic logs diagnostic_logs = diagnostics.get("logs", {}) diagnostic_log_level = diagnostic_logs.get("level") check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display( label="Log Level", value=diagnostic_log_level, padding=log_inner_padding ), ) if detail_level > ResourceOutputDetailLevel.detail.value: diagnostic_log_otelconfig = diagnostic_logs.get("openTelemetryExportConfig", {}) if diagnostic_log_otelconfig: for label, key in [ ("Endpoint", "otlpGrpcEndpoint"), ("Interval (s)", "intervalSeconds"), ("Level", "level"), ]: val = diagnostic_log_otelconfig.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display( label=label, value=val, padding=log_inner_padding ), ) # diagnostic metrics diagnostic_metrics = diagnostics.get("metrics", {}) check_manager.add_display( target_name=target, namespace=namespace, display=Padding("Diagnostic Metrics:", (0, 0, 0, log_padding)), ) diagnostic_metrics_prometheusPort = diagnostic_metrics.get("prometheusPort") check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display( label="Prometheus Port", value=diagnostic_metrics_prometheusPort, padding=log_inner_padding, ), ) diagnostic_metrics_otelconfig = diagnostic_metrics.get("openTelemetryExportConfig", {}) if diagnostic_metrics_otelconfig: for label, key in [ ("Endpoint", "otlpGrpcEndpoint"), ("Interval (s)", "intervalSeconds"), ]: val = diagnostic_metrics_otelconfig.get(key) if val: check_manager.add_display( target_name=target, namespace=namespace, display=basic_property_display( label=label, value=val, padding=log_inner_padding ), ) # pod health - trailing `-` is important in case profiles have similar prefixes pod_prefix = f"{DATAFLOW_PROFILE_POD_PREFIX}{profile_name}-" profile_pods = get_namespaced_pods_by_prefix( prefix=pod_prefix, namespace=namespace, label_selector=DATAFLOW_NAME_LABEL, ) # only show pods if they exist if profile_pods: evaluate_pod_health( check_manager=check_manager, target=target, pods=profile_pods, namespace=namespace, padding=INNER_PADDING, detail_level=detail_level, ) # default dataflow profile status, display warning if not success check_manager.add_target_eval( target_name=target, namespace=namespace, status=default_profile_status.value, resource_kind=DataflowResourceKinds.DATAFLOWPROFILE.value, resource_name=DEFAULT_DATAFLOW_PROFILE, value={f"[*].metadata.name=='{DEFAULT_DATAFLOW_PROFILE}'": default_profile_status.value}, ) if default_profile_status not in [CheckTaskStatus.success, CheckTaskStatus.skipped]: check_manager.add_display( target_name=target, namespace=namespace, display=Padding( colorize_string( color=default_profile_status.color, value=f"\nDefault Dataflow Profile '{DEFAULT_DATAFLOW_PROFILE}' not found in namespace '{namespace}'", ), (0, 0, 0, PADDING), ), ) return check_manager.as_dict(as_list=as_list)