fix: make az iot ops check find nested errors (#156)

azext_edge/edge/providers/check/akri.py

@@ -15,7 +15,7 @@
     AkriResourceKinds,
 )
 
-from ..support.akri import AKRI_PREFIX
+from ..support.akri import AKRI_PREFIX, AKRI_INSTANCE_LABEL, AKRI_APP_LABEL
 
 from .base import (
     CheckManager,
@@ -67,11 +67,16 @@ def evaluate_core_service_runtime(
     check_manager = CheckManager(check_name="evalCoreServiceRuntime", check_desc="Evaluate Akri core service")
 
     padding = 6
-    akri_runtime_resources = get_namespaced_pods_by_prefix(
-        prefix=AKRI_PREFIX,
-        namespace="",
-        label_selector="",
-    )
+    akri_runtime_resources: List[dict] = []
+
+    for label in [AKRI_INSTANCE_LABEL, AKRI_APP_LABEL]:
+        akri_runtime_resources.extend(
+            get_namespaced_pods_by_prefix(
+                prefix=AKRI_PREFIX,
+                namespace="",
+                label_selector=label,
+            )
+        )
 
     for (namespace, pods) in pods_grouped_by_namespace(akri_runtime_resources):
         check_manager.add_target(target_name=CoreServiceResourceKinds.RUNTIME_RESOURCE.value, namespace=namespace)

azext_edge/edge/providers/check/base.py

@@ -785,29 +785,43 @@ def process_dict_resource(
                 padding=padding + PADDING_SIZE
             )
         else:
-            display_text = ""
+            display_text = f"{key}: "
+            value_padding = padding
             if isinstance(value, str) and len(value) > 50:
-                display_text = f"{key}:"
-                check_manager.add_display(
-                    target_name=target_name,
-                    namespace=namespace,
-                    display=Padding(display_text, (0, 0, 0, padding))
-                )
-                display_text = f"[cyan]{value}[/cyan]"
-                check_manager.add_display(
-                    target_name=target_name,
-                    namespace=namespace,
-                    display=Padding(display_text, (0, 0, 0, padding + PADDING_SIZE))
-                )
-            else:
-                # replace empty string with N/A
-                value = value if value else "N/A"
-                display_text = f"{key}: [cyan]{value}[/cyan]"
                 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_value_color(
+    check_manager: CheckManager,
+    target_name: str,
+    key: Any,
+    value: Any,
+) -> str:
+    value = value if value else "N/A"
+    if "error" in str(key).lower() and str(value).lower() not in ["null", "n/a", "none", "noerror"]:
+        check_manager.set_target_status(
+            target_name=target_name,
+            status=CheckTaskStatus.error.value
+        )
+        return f"[red]{value}[/red]"
+    return f"[cyan]{value}[/cyan]"
 
 
 def process_list_resource(

azext_edge/tests/edge/checks/test_base_helpers_unit.py

@@ -0,0 +1,41 @@
+# 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.
+# ----------------------------------------------------------------------------------------------
+
+import pytest
+from azext_edge.edge.common import CheckTaskStatus
+
+from ...generators import generate_generic_id
+
+
+@pytest.fixture
+def mock_check_manager(mocker):
+    mocked_manager = mocker.patch("azext_edge.edge.providers.check.base.CheckManager", autospec=True)
+    yield mocked_manager
+
+
+@pytest.mark.parametrize("key", ["discoveryError", "discovery", "error", 0])
+@pytest.mark.parametrize("value", ["Null", "", "None", "NoError", "Everything is ok~", 100])
+def test_process_value_color(mock_check_manager, key, value):
+    from azext_edge.edge.providers.check.base import process_value_color
+    target_name = generate_generic_id()
+    result = process_value_color(
+        check_manager=mock_check_manager, target_name=target_name, key=key, value=value
+    )
+    if not value:
+        value = "N/A"
+    assert str(value) in result
+
+    if all([
+        "error" in str(key).lower(),
+        str(value).lower() not in ["null", "n/a", "none", "noerror"]
+    ]):
+        assert result.startswith("[red]")
+        mock_check_manager.set_target_status.assert_called_once_with(
+            target_name=target_name,
+            status=CheckTaskStatus.error.value
+        )
+    else:
+        assert result.startswith("[cyan]")