chore: merge dev into feature/0.7.0 (#321)

azext_edge/edge/_help.py

@@ -95,6 +95,7 @@ def load_iotops_help():
             - {COMPAT_DEVICEREGISTRY_APIS.as_str()}
             - {COMPAT_MQTT_BROKER_APIS.as_str()}
             - {COMPAT_OPCUA_APIS.as_str()}
+            - {COMPAT_DATAFLOW_APIS.as_str()}
 
             For more information on cluster requirements, please check https://aka.ms/iot-ops-cluster-requirements
 

azext_edge/edge/common.py

@@ -157,6 +157,7 @@ def list_check_services(cls):
             cls.opcua.value,
             cls.akri.value,
             cls.deviceregistry.value,
+            cls.dataflow.value,
         ]
 
 
@@ -246,6 +247,9 @@ class BundleResourceKind(Enum):
 METRICS_SERVICE_API_PORT = 9600
 PROTOBUF_SERVICE_API_PORT = 9800
 
+# Dataflow constants
+DEFAULT_DATAFLOW_PROFILE = "profile"
+
 # Init Env Control
 
 INIT_NO_PREFLIGHT_ENV_KEY = "AIO_CLI_INIT_PREFLIGHT_DISABLED"

azext_edge/edge/params.py

@@ -15,6 +15,8 @@
 )
 from knack.arguments import CaseInsensitiveList
 
+from azext_edge.edge.providers.edge_api.dataflow import DataflowResourceKinds
+
 from ._validators import validate_namespace, validate_resource_name
 from .common import FileType, OpsServiceType
 from .providers.check.common import ResourceOutputDetailLevel
@@ -170,6 +172,9 @@ def load_iotops_arguments(self, _):
                         OpcuaResourceKinds.ASSET_TYPE.value,
                         AkriResourceKinds.CONFIGURATION.value,
                         AkriResourceKinds.INSTANCE.value,
+                        DataflowResourceKinds.DATAFLOW.value,
+                        DataflowResourceKinds.DATAFLOWENDPOINT.value,
+                        DataflowResourceKinds.DATAFLOWPROFILE.value,
                     ]
                 )
             ),

azext_edge/edge/providers/check/base/check_manager.py

@@ -68,14 +68,14 @@ def __init__(self, check_name: str, check_desc: str):
         self.check_desc = check_desc
         self.targets = {}
         self.target_displays = {}
-        self.worst_status = CheckTaskStatus.success.value
+        self.worst_status = CheckTaskStatus.skipped.value
 
     def add_target(
         self,
         target_name: str,
         namespace: str = ALL_NAMESPACES_TARGET,
         conditions: List[str] = None,
-        description: str = None
+        description: str = None,
     ) -> None:
         # TODO: maybe make a singular taget into a class for consistent structure?
         if target_name not in self.targets:
@@ -85,7 +85,7 @@ def add_target(
             self.targets[target_name][namespace] = {}
         self.targets[target_name][namespace]["conditions"] = conditions
         self.targets[target_name][namespace]["evaluations"] = []
-        self.targets[target_name][namespace]["status"] = CheckTaskStatus.success.value
+        self.targets[target_name][namespace]["status"] = CheckTaskStatus.skipped.value
         if description:
             self.targets[target_name][namespace]["description"] = description
 
@@ -124,20 +124,23 @@ def add_target_eval(
         self._process_status(target_name, status, namespace)
 
     def _process_status(self, target_name: str, status: str, namespace: str = ALL_NAMESPACES_TARGET) -> None:
-        existing_status = self.targets[target_name].get("status", CheckTaskStatus.success.value)
-        if existing_status != status:
-            if existing_status == CheckTaskStatus.success.value and status in [
-                CheckTaskStatus.warning.value,
-                CheckTaskStatus.error.value,
-                CheckTaskStatus.skipped.value,
-            ]:
+        namespace_status = self.targets[target_name][namespace].get("status")
+        # success only overrides skipped status (default)
+        if status == CheckTaskStatus.success.value:
+            if namespace_status == CheckTaskStatus.skipped.value:
                 self.targets[target_name][namespace]["status"] = status
+            if self.worst_status == CheckTaskStatus.skipped.value:
                 self.worst_status = status
-            elif existing_status in [
-                CheckTaskStatus.warning.value, CheckTaskStatus.skipped.value
-            ] and status in [CheckTaskStatus.error.value]:
+        # warning overrides any state that is not "error"
+        elif status == CheckTaskStatus.warning.value:
+            if namespace_status != CheckTaskStatus.error.value:
                 self.targets[target_name][namespace]["status"] = status
+            if self.worst_status != CheckTaskStatus.error.value:
                 self.worst_status = status
+        # error overrides any state
+        elif status == CheckTaskStatus.error.value:
+            self.targets[target_name][namespace]["status"] = status
+            self.worst_status = status
 
     def add_display(self, target_name: str, display: Any, namespace: str = ALL_NAMESPACES_TARGET) -> None:
         if target_name not in self.target_displays:

azext_edge/edge/providers/check/base/display.py

@@ -10,7 +10,7 @@
 from typing import Any, Dict, List, Optional, Tuple
 
 from .check_manager import CheckManager
-from ..common import ALL_NAMESPACES_TARGET
+from ..common import ALL_NAMESPACES_TARGET, COLOR_STR_FORMAT, DEFAULT_PADDING, DEFAULT_PROPERTY_DISPLAY_COLOR
 from ....common import CheckTaskStatus
 
 logger = get_logger(__name__)
@@ -142,3 +142,21 @@ def process_value_color(
         )
         return f"[red]{value}[/red]"
     return f"[cyan]{value}[/cyan]"
+
+
+def colorize_string(value: str, color: Optional[str] = DEFAULT_PROPERTY_DISPLAY_COLOR) -> str:
+    color = color or DEFAULT_PROPERTY_DISPLAY_COLOR
+    return COLOR_STR_FORMAT.format(value=value, color=color)
+
+
+def basic_property_display(
+    label: str,
+    value: str,
+    color: Optional[str] = DEFAULT_PROPERTY_DISPLAY_COLOR,
+    padding: Optional[int] = DEFAULT_PADDING
+) -> Padding:
+    padding = padding or DEFAULT_PADDING
+    return Padding(
+        f"{label}: {colorize_string(value=value, color=color)}",
+        (0, 0, 0, padding)
+    )

azext_edge/edge/providers/check/base/node.py

@@ -58,6 +58,7 @@ def check_nodes(as_list: bool = False) -> Dict[str, Any]:
             check_manager.add_display(target_name=target, display=target_display)
             return check_manager.as_dict()
 
+        check_manager.add_target_eval(target_name=target, status=CheckTaskStatus.success.value, value={"len(cluster/nodes)": len(nodes.items)})
         table = _generate_node_table(check_manager, nodes)
         check_manager.add_display(target_name=target, display=Padding(table, padding))
 

azext_edge/edge/providers/check/common.py

@@ -103,6 +103,30 @@ class CoreServiceResourceKinds(Enum):
     RUNTIME_RESOURCE = "coreServiceRuntimeResource"
 
 
+# Dataflow properties
+class DataflowOperationType(ListableEnum):
+    """
+    Dataflow Profile Operation Type:
+    """
+
+    source = "source"
+    destination = "destination"
+    builtin_transformation = "builtintransformation"
+
+
+class DataflowEndpointType(ListableEnum):
+    """
+    Dataflow Endpoint Type:
+    """
+
+    data_explorer = "dataexplorer"
+    datalake = "datalakestorage"
+    fabric_onelake = "fabriconelake"
+    kafka = "kafka"
+    local_storage = "localstorage"
+    mqtt = "mqtt"
+
+
 # Akri runtime attributes
 AKRI_PREFIX = "aio-akri-"
 
@@ -132,6 +156,8 @@ class CoreServiceResourceKinds(Enum):
 DISPLAY_BYTES_PER_GIGABYTE = 10 ** 9
 
 # UI constants
+DEFAULT_PADDING = 8
 PADDING_SIZE = 4
+DEFAULT_PROPERTY_DISPLAY_COLOR = "cyan"
 
 COLOR_STR_FORMAT = "[{color}]{value}[/{color}]"