feat: integrates AIO version 0.3.0-preview + refactors + other improv…

…ements (#132)

* feat: integrates AIO version 0.3.0-preview
* refactor: removes user `--cluster-location`. The target cluster location will always be queried.
* refactor: orchestration host module connectivity checks for more re-usable code.
* refactor: permission check (which currently only evaluates rg role-write permissions) runs if resource sync is enabled (default)
* introduces a template optimizer python script to trim or modify attributes of the deployment template.

azext_edge/constants.py

@@ -7,7 +7,7 @@
 
 import os
 
-VERSION = "0.2.0b4"
+VERSION = "0.3.0b1"
 EXTENSION_NAME = "azure-iot-ops"
 EXTENSION_ROOT = os.path.dirname(os.path.abspath(__file__))
 USER_AGENT = "IotOperationsCliExtension/{}".format(VERSION)

azext_edge/edge/commands_edge.py

@@ -92,7 +92,6 @@ def init(
     cluster_name: str,
     resource_group_name: str,
     cluster_namespace: str = DEFAULT_NAMESPACE,
-    cluster_location: Optional[str] = None,
     keyvault_sat_secret_name: str = DEFAULT_NAMESPACE,
     custom_location_namespace: Optional[str] = None,
     custom_location_name: Optional[str] = None,
@@ -135,6 +134,7 @@ def init(
     no_deploy: Optional[bool] = None,
     no_tls: Optional[bool] = None,
     no_preflight: Optional[bool] = None,
+    disable_rsync_rules: Optional[bool] = None,
     context_name: Optional[str] = None,
 ) -> Union[Dict[str, Any], None]:
     from .providers.orchestration import deploy
@@ -206,7 +206,7 @@ def init(
         cmd=cmd,
         cluster_name=cluster_name,
         cluster_namespace=cluster_namespace,
-        cluster_location=cluster_location,
+        cluster_location=None,  # Effectively always fetch connected cluster location
         custom_location_name=custom_location_name,
         custom_location_namespace=custom_location_namespace,
         resource_group_name=resource_group_name,
@@ -219,6 +219,7 @@ def init(
         no_tls=no_tls,
         no_preflight=no_preflight,
         no_deploy=no_deploy,
+        disable_rsync_rules=disable_rsync_rules,
         dp_instance_name=dp_instance_name,
         dp_reader_workers=int(dp_reader_workers),
         dp_runner_workers=int(dp_runner_workers),

azext_edge/edge/params.py

@@ -230,13 +230,8 @@ def load_iotops_arguments(self, _):
         context.argument(
             "location",
             options_list=["--location"],
-            help="The ARM location that will be used for provisioned ARM collateral. "
-            "If not provided the resource group location will be used.",
-        )
-        context.argument(
-            "cluster_location",
-            options_list=["--cluster-location"],
-            help="The cluster ARM location.",
+            help="The ARM location that will be used for provisioned RPSaaS collateral. "
+            "If not provided the connected cluster location will be used.",
         )
         context.argument(
             "show_template",
@@ -275,6 +270,12 @@ def load_iotops_arguments(self, _):
             arg_type=get_three_state_flag(),
             help="The pre-flight workflow will be skipped.",
         )
+        context.argument(
+            "disable_rsync_rules",
+            options_list=["--disable-rsync-rules"],
+            arg_type=get_three_state_flag(),
+            help="Resource sync rules will not be included in the deployment.",
+        )
         # Akri
         context.argument(
             "opcua_discovery_endpoint",

azext_edge/edge/providers/orchestration/base.py

@@ -461,18 +461,26 @@ def deploy_template(
     return result, deployment
 
 
-def verify_connect_mgmt_plane(cmd):
-    from .host import check_connectivity, get_connectivity_error, ARM_ENDPOINT
-
-    try_arm_endpoint = ARM_ENDPOINT
-    try:
-        try_arm_endpoint = cmd.cli_ctx.cloud.endpoints.resource_manager
-    except AttributeError:
-        pass
-
-    connect_result = check_connectivity(try_arm_endpoint, http_verb="HEAD")
-    if not connect_result:
-        raise ValidationError(get_connectivity_error(try_arm_endpoint, include_cluster=False))
+def process_default_location(kwargs: dict):
+    # TODO: use intermediate object to store KPIs / refactor out of kwargs
+    cluster_location = kwargs["cluster_location"]
+    location = kwargs["location"]
+    cluster_name = kwargs["cluster_name"]
+    subscription_id = kwargs["subscription_id"]
+    resource_group_name = kwargs["resource_group_name"]
+
+    if not cluster_location or not location:
+        from .connected_cluster import ConnectedCluster
+
+        connected_cluster = ConnectedCluster(
+            subscription_id=subscription_id, cluster_name=cluster_name, resource_group_name=resource_group_name
+        )
+        connected_cluster_location = connected_cluster.location
+
+        if not cluster_location:
+            kwargs["cluster_location"] = connected_cluster_location
+        if not location:
+            kwargs["location"] = connected_cluster_location
 
 
 def wait_for_terminal_state(poller: "LROPoller") -> "GenericResource":

azext_edge/edge/providers/orchestration/connected_cluster.py

@@ -0,0 +1,29 @@
+# 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 ...util.az_client import get_resource_client
+
+CONNECTED_CLUSTER_API_VERSION = "2022-10-01-preview"
+
+
+class ConnectedCluster:
+    def __init__(self, subscription_id: str, cluster_name: str, resource_group_name: str):
+        self.subscription_id = subscription_id
+        self.cluster_name = cluster_name
+        self.resource_group_name = resource_group_name
+        self.resource_client = get_resource_client(self.subscription_id)
+
+    @property
+    def resource(self) -> dict:
+        return self.resource_client.resources.get_by_id(
+            resource_id=f"/subscriptions/{self.subscription_id}/resourceGroups/{self.resource_group_name}"
+            f"/providers/Microsoft.Kubernetes/connectedClusters/{self.cluster_name}",
+            api_version=CONNECTED_CLUSTER_API_VERSION,
+        ).as_dict()
+
+    @property
+    def location(self) -> str:
+        return self.resource["location"]

azext_edge/edge/providers/orchestration/host.py

@@ -6,7 +6,7 @@
 
 
 from platform import system
-from typing import List, Optional, Tuple
+from typing import Dict, List, NamedTuple, Optional, Tuple
 
 import requests
 from azure.cli.core.azclierror import ValidationError
@@ -20,28 +20,44 @@
 from ...util.common import run_host_command
 
 ARM_ENDPOINT = "https://management.azure.com/"
-MCR_ENDPOINT = "https://mcr.microsoft.com/api/v1/"
+MCR_ENDPOINT = "https://mcr.microsoft.com/"
+GRAPH_ENDPOINT = "https://graph.microsoft.com/"
 
 NFS_COMMON_ALIAS = "nfs-common"
 
 logger = get_logger(__name__)
 console = Console(width=88)
 
 
+class EndpointConnections(NamedTuple):
+    connect_map: Dict[str, bool]
+
+    @property
+    def failed_connections(self):
+        fc = []
+        for endpoint in self.connect_map:
+            if not self.connect_map[endpoint]:
+                fc.append(endpoint)
+        return fc
+
+    def throw_if_failure(self, include_cluster: bool = True):
+        failed_conns = self.failed_connections
+        if failed_conns:
+            raise ValidationError(get_connectivity_error(failed_conns, include_cluster=include_cluster))
+
+
 def run_host_verify(render_progress: Optional[bool] = True, confirm_yes: Optional[bool] = False):
     if not render_progress:
         console.quiet = True
-    connect_tuples = [(ARM_ENDPOINT, "HEAD"), (MCR_ENDPOINT, "GET")]
-    connect_result_map = {}
+    connect_endpoints = [ARM_ENDPOINT, MCR_ENDPOINT]
     console.print()
 
-    with console.status(status="Analyzing host...", refresh_per_second=12.5) as status_render:
+    with console.status(status="Analyzing host...") as status_render:
         console.print("[bold]Connectivity[/bold] to:")
-        for connect_tuple in connect_tuples:
-            connect_result_map[connect_tuple] = check_connectivity(url=connect_tuple[0], http_verb=connect_tuple[1])
-            console.print(f"- {connect_tuple[0]} ", "...", connect_result_map[connect_tuple])
-            if not connect_result_map[connect_tuple]:
-                raise ValidationError(get_connectivity_error(connect_tuple[0]))
+        endpoint_connections = preflight_http_connections(connect_endpoints)
+        for endpoint in endpoint_connections.connect_map:
+            console.print(f"- {endpoint} ", "...", endpoint_connections.connect_map[endpoint])
+        endpoint_connections.throw_if_failure()
 
         if not is_windows():
             if is_ubuntu_distro():
@@ -92,17 +108,13 @@ def run_host_verify(render_progress: Optional[bool] = True, confirm_yes: Optiona
     console.print()
 
 
-def check_connectivity(url: str, timeout: int = 20, http_verb: str = "GET"):
+def check_connectivity(url: str, timeout: int = 20):
     try:
-        http_verb = http_verb.lower()
-        if http_verb == "get":
-            req = requests.get(url, timeout=timeout)
-        if http_verb == "head":
-            req = requests.head(url, timeout=timeout)
-
+        req = requests.head(url=url, timeout=timeout)
         req.raise_for_status()
         return True
     except requests.HTTPError:
+        # HTTPError implies an http server response
         return True
     except requests.ConnectionError:
         return False
@@ -173,10 +185,19 @@ def _get_eval_result_display(eval_result: Optional[bool]) -> str:
 
 
 def get_connectivity_error(
-    endpoint: str, protocol: str = "https", direction: str = "outbound", include_cluster: bool = True
+    endpoints: List[str], protocol: str = "https", direction: str = "outbound", include_cluster: bool = True
 ):
+    todo_endpoints = []
+    if endpoints:
+        todo_endpoints.extend(endpoints)
+
+    endpoints_list_format = ""
+    for ep in todo_endpoints:
+        endpoints_list_format += f"* {ep}\n"
+
     connectivity_error = (
-        f"\nUnable to verify {direction} {protocol} connectivity to {endpoint}.\n"
+        f"\nUnable to verify {direction} {protocol} connectivity to:\n"
+        f"\n{endpoints_list_format}\n"
         "Ensure host, proxy and/or firewall config allows connection.\n"
         "\nThe 'HTTP_PROXY' and 'HTTPS_PROXY' environment variables can be used for the CLI client.\n"
     )
@@ -190,3 +211,25 @@ def get_connectivity_error(
     )
 
     return connectivity_error
+
+
+def preflight_http_connections(endpoints: List[str]) -> EndpointConnections:
+    """
+    Tests connectivity for each endpoint in the provided list.
+    """
+    todo_connect_endpoints = []
+    if endpoints:
+        todo_connect_endpoints.extend(endpoints)
+
+    endpoint_connect_map = {}
+    for endpoint in todo_connect_endpoints:
+        endpoint_connect_map[endpoint] = check_connectivity(url=endpoint)
+
+    return EndpointConnections(connect_map=endpoint_connect_map)
+
+
+def verify_cli_client_connections(include_graph: bool):
+    test_endpoints = [ARM_ENDPOINT]
+    if include_graph:
+        test_endpoints.append(GRAPH_ENDPOINT)
+    preflight_http_connections(test_endpoints).throw_if_failure(include_cluster=False)

azext_edge/edge/providers/orchestration/permissions.py

@@ -15,7 +15,7 @@
 
 
 # TODO: one-off for time, make generic
-def verify_write_permission_against_rg(subscription_id: str, resource_group_name: str, **kwargs) -> bool:
+def verify_write_permission_against_rg(subscription_id: str, resource_group_name: str, **kwargs):
     for permission in get_principal_permissions_for_group(
         subscription_id=subscription_id, resource_group_name=resource_group_name
     ):