refactor: Incremental changeset for init and deployment mechanisms

azext_edge/edge/_help.py

@@ -150,7 +150,7 @@ def load_iotops_help():
           text: >
             az iot ops broker stats --raw
 
-        - name: Fetch all available mq traces from the diagnostics Protobuf endpoint.
+        - name: Fetch all available mqtt broker traces from the diagnostics Protobuf endpoint.
                 This will produce a `.zip` with both `Otel` and Grafana `tempo` file formats.
                 A trace files last modified attribute will match the trace timestamp.
           text: >

azext_edge/edge/providers/orchestration/base.py

@@ -9,12 +9,6 @@
 from azure.cli.core.azclierror import ValidationError
 from knack.log import get_logger
 
-from ...util import (
-    get_timestamp_now_utc,
-)
-from ...util.az_client import (
-    get_resource_client,
-)
 from ..k8s.cluster_role_binding import get_bindings
 from ..k8s.config_map import get_config_map
 from .common import (
@@ -28,125 +22,6 @@
 
 logger = get_logger(__name__)
 
-IOT_OPERATIONS_EXTENSION_PREFIX = "microsoft.iotoperations"
-
-
-# TODO - @digimaun - potentially can reuse
-# def configure_cluster_secrets(
-#     cluster_namespace: str,
-#     cluster_secret_ref: str,
-#     cluster_akv_secret_class_name: str,
-#     keyvault_spc_secret_name: str,
-#     keyvault_resource_id: str,
-#     sp_record: ServicePrincipal,
-#     **kwargs,
-# ):
-#     if not KEYVAULT_API_V1.is_deployed():
-#         raise ValidationError(
-#             f"The API {KEYVAULT_API_V1.as_str()} "
-#             "is not available on the cluster the local kubeconfig is configured for.\n"
-#             "Please ensure the local kubeconfig matches the target cluster intended for deployment."
-#         )
-
-#     if not get_cluster_namespace(namespace=cluster_namespace):
-#         create_cluster_namespace(namespace=cluster_namespace)
-
-#     create_namespaced_secret(
-#         secret_name=cluster_secret_ref,
-#         namespace=cluster_namespace,
-#         data={"clientid": sp_record.client_id, "clientsecret": sp_record.secret},
-#         labels={"secrets-store.csi.k8s.io/used": "true"},
-#         delete_first=True,
-#     )
-
-#     yaml_configs = []
-#     keyvault_split = keyvault_resource_id.split("/")
-#     keyvault_name = keyvault_split[-1]
-
-#     for secret_class in [
-#         cluster_akv_secret_class_name,
-#         "aio-opc-ua-broker-client-certificate",
-#         "aio-opc-ua-broker-user-authentication",
-#         "aio-opc-ua-broker-trust-list",
-#         "aio-opc-ua-broker-issuer-list",
-#     ]:
-#         yaml_configs.append(
-#             get_kv_secret_store_yaml(
-#                 name=secret_class,
-#                 namespace=cluster_namespace,
-#                 keyvault_name=keyvault_name,
-#                 secret_name=keyvault_spc_secret_name,
-#                 tenantId=sp_record.tenant_id,
-#             )
-#         )
-
-#     create_namespaced_custom_objects(
-#         group=KEYVAULT_API_V1.group,
-#         version=KEYVAULT_API_V1.version,
-#         plural="secretproviderclasses",  # TODO
-#         namespace=cluster_namespace,
-#         yaml_objects=yaml_configs,
-#         delete_first=True,
-#     )
-
-
-def deploy_template(
-    template: dict,
-    parameters: dict,
-    subscription_id: str,
-    resource_group_name: str,
-    deployment_name: str,
-    cluster_name: str,
-    cluster_namespace: str,
-    instance_name: str,
-    pre_flight: bool = False,
-    **kwargs,
-) -> Tuple[dict, dict]:
-    resource_client = get_resource_client(subscription_id=subscription_id)
-
-    deployment_params = {"properties": {"mode": "Incremental", "template": template, "parameters": parameters}}
-    if pre_flight:
-        return {}, resource_client.deployments.begin_what_if(
-            resource_group_name=resource_group_name,
-            deployment_name=deployment_name,
-            parameters=deployment_params,
-        )
-
-    deployment = resource_client.deployments.begin_create_or_update(
-        resource_group_name=resource_group_name,
-        deployment_name=deployment_name,
-        parameters=deployment_params,
-    )
-
-    deploy_link = (
-        "https://portal.azure.com/#blade/HubsExtension/DeploymentDetailsBlade/id/"
-        f"%2Fsubscriptions%2F{subscription_id}%2FresourceGroups%2F{resource_group_name}"
-        f"%2Fproviders%2FMicrosoft.Resources%2Fdeployments%2F{deployment_name}"
-    )
-
-    result = {
-        "clusterName": cluster_name,
-        "clusterNamespace": cluster_namespace,
-        "deploymentLink": deploy_link,
-        "deploymentName": deployment_name,
-        "deploymentState": {"timestampUtc": {"started": get_timestamp_now_utc()}, "status": deployment.status()},
-        "instanceName": instance_name,
-        "resourceGroup": resource_group_name,
-        "subscriptionId": subscription_id,
-    }
-    return result, deployment
-
-
-def throw_if_iotops_deployed(connected_cluster: ConnectedCluster):
-    connected_cluster_extensions = connected_cluster.extensions
-    for extension in connected_cluster_extensions:
-        if "properties" in extension and "extensionType" in extension["properties"]:
-            if extension["properties"]["extensionType"].lower().startswith(IOT_OPERATIONS_EXTENSION_PREFIX):
-                raise ValidationError(
-                    "Detected existing IoT Operations deployment. "
-                    "Remove IoT Operations or use a different connected cluster to continue.\n"
-                )
-
 
 def verify_custom_locations_enabled(cmd):
     from azure.cli.core.util import send_raw_request

azext_edge/edge/providers/orchestration/permissions.py

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

azext_edge/edge/providers/orchestration/resources/instances.py

@@ -55,7 +55,7 @@ def _show_tree(self, instance: dict):
     def _get_associated_cl(self, instance: dict) -> dict:
         return self.resource_client.resources.get_by_id(
             resource_id=instance["extendedLocation"]["name"], api_version=CUSTOM_LOCATIONS_API_VERSION
-        ).as_dict()
+        )
 
     def get_resource_map(self, instance: dict) -> IoTOperationsResourceMap:
         custom_location = self._get_associated_cl(instance)

azext_edge/edge/providers/orchestration/rp_namespace.py

@@ -21,14 +21,13 @@
 )
 
 
-def register_providers(subscription_id: str, **kwargs):
+def register_providers(subscription_id: str):
     resource_client = get_resource_client(subscription_id=subscription_id)
     providers_list = resource_client.providers.list()
     for provider in providers_list:
-        provider_dict = provider.as_dict()
-        if "namespace" in provider_dict and provider_dict["namespace"] in RP_NAMESPACE_SET:
-            if provider_dict["registration_state"] == "Registered":
-                logger.debug("RP %s is already registered.", provider_dict["namespace"])
+        if "namespace" in provider and provider["namespace"] in RP_NAMESPACE_SET:
+            if provider["registrationState"] == "Registered":
+                logger.debug("RP %s is already registered.", provider["namespace"])
                 continue
-            logger.debug("Registering RP %s.", provider_dict["namespace"])
-            resource_client.providers.register(provider_dict["namespace"])
+            logger.debug("Registering RP %s.", provider["namespace"])
+            resource_client.providers.register(provider["namespace"])

azext_edge/edge/providers/orchestration/targets.py

@@ -0,0 +1,126 @@
+# 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, Optional, Tuple
+
+from ...util import url_safe_random_chars
+from .template import (
+    M2_ENABLEMENT_TEMPLATE,
+    M2_INSTANCE_TEMPLATE,
+    TemplateVer,
+    get_basic_dataflow_profile,
+)
+
+
+class InitTargets:
+    def __init__(
+        self,
+        cluster_name: str,
+        resource_group_name: str,
+        cluster_namespace: str = "azure-iot-operations",
+        location: Optional[str] = None,
+        custom_location_name: Optional[str] = None,
+        disable_rsync_rules: Optional[bool] = None,
+        instance_name: Optional[str] = None,
+        instance_description: Optional[str] = None,
+        enable_fault_tolerance: Optional[bool] = None,
+        dataflow_profile_instances: int = 1,
+        broker_config: Optional[dict] = None,
+        add_insecure_listener: Optional[bool] = None,
+        **_,
+    ):
+        self.cluster_name = cluster_name
+        self.safe_cluster_name = self._sanitize_k8s_name(self.cluster_name)
+        self.resource_group_name = resource_group_name
+        self.cluster_namespace = self._sanitize_k8s_name(cluster_namespace)
+        self.location = location
+        self.custom_location_name = (
+            self._sanitize_k8s_name(custom_location_name)
+            or f"{self.safe_cluster_name}-{url_safe_random_chars(3).lower()}-ops-cl"
+        )
+        self.deploy_resource_sync_rules: bool = not disable_rsync_rules
+        self.instance_name = self._sanitize_k8s_name(instance_name)
+        self.instance_description = instance_description
+        self.enable_fault_tolerance = enable_fault_tolerance
+        self.dataflow_profile_instances = dataflow_profile_instances
+        self.broker_config = broker_config
+        self.add_insecure_listener = add_insecure_listener
+
+    def _sanitize_k8s_name(self, name: str) -> str:
+        if not name:
+            return name
+        sanitized = str(name)
+        sanitized = sanitized.lower()
+        sanitized = sanitized.replace("_", "-")
+        return sanitized
+
+    def _handle_apply_targets(
+        self, param_to_target: dict, template_blueprint: TemplateVer
+    ) -> Tuple[TemplateVer, dict]:
+        template_copy = template_blueprint.copy()
+        built_in_template_params = template_copy.parameters
+
+        deploy_params = {}
+
+        for param in param_to_target:
+            if param in built_in_template_params and param_to_target[param]:
+                deploy_params[param] = {"value": param_to_target[param]}
+
+        return template_copy, deploy_params
+
+    def get_ops_enablement_template(
+        self,
+    ) -> Tuple[dict, dict]:
+        template, parameters = self._handle_apply_targets(
+            param_to_target={
+                "clusterName": self.cluster_name,
+                "kubernetesDistro": "",
+                "containerRuntimeSocket": "",
+                "trustSource": "",
+                # "trustBundleSettings": ""
+                "schemaRegistryId": "",
+            },
+            template_blueprint=M2_ENABLEMENT_TEMPLATE,
+        )
+        return template.content, parameters
+
+    def get_ops_instance_template(
+        self,
+    ) -> Tuple[dict, dict]:
+        template, parameters = self._handle_apply_targets(
+            param_to_target={
+                "clusterName": self.cluster_name,
+                "kubernetesDistro": "",
+                "containerRuntimeSocket": "",
+                "trustSource": "",
+                # "trustBundleSettings": ""
+                "schemaRegistryId": "",
+            },
+            template_blueprint=M2_INSTANCE_TEMPLATE,
+        )
+
+        content = template.content
+        deploy_resources: List[dict] = content.get("resources", [])
+        df_profile_instances = self.dataflow_profile_instances
+        deploy_resources.append(get_basic_dataflow_profile(instance_count=df_profile_instances))
+
+        if self.broker_config:
+            broker_config = self.broker_config
+            if "properties" in broker_config:
+                broker_config = broker_config["properties"]
+            broker: dict = template.get_resource_defs("Microsoft.IoTOperations/instances/brokers")
+            broker["properties"] = broker_config
+
+        if self.add_insecure_listener:
+            # This solution entirely relies on the form of the "standard" template.
+            # TODO - @digimaun - default resource names
+            # TODO - @digimaun - new listener
+            default_listener = template.get_resource_defs("Microsoft.IoTOperations/instances/brokers/listeners")
+            if default_listener:
+                ports: list = default_listener["properties"]["ports"]
+                ports.append({"port": 1883})
+
+        return content, parameters