in azext_edge/edge/providers/orchestration/work.py [0:0]
def _do_work(self):
from .host import verify_cli_client_connections
from .permissions import verify_write_permission_against_rg
from .rp_namespace import register_providers
try:
# Ensure connection to ARM if needed. Show remediation error message otherwise.
self._render_display()
verify_cli_client_connections()
self._process_connected_cluster()
# Pre-Flight workflow
if self._pre_flight:
# WorkStepKey.REG_RP
self._render_display(category=WorkCategoryKey.PRE_FLIGHT, active_step=WorkStepKey.REG_RP)
register_providers(self.subscription_id)
self._complete_step(
category=WorkCategoryKey.PRE_FLIGHT,
completed_step=WorkStepKey.REG_RP,
active_step=WorkStepKey.ENUMERATE_PRE_FLIGHT,
)
# WorkStepKey.ENUMERATE_PRE_FLIGHT
if self._targets.deploy_resource_sync_rules and self._targets.instance_name:
# TODO - @digimaun use permission manager after fixing check access issue
verify_write_permission_against_rg(
subscription_id=self.subscription_id, resource_group_name=self._targets.resource_group_name
)
self._complete_step(
category=WorkCategoryKey.PRE_FLIGHT,
completed_step=WorkStepKey.ENUMERATE_PRE_FLIGHT,
)
# Enable IoT Ops workflow
if self._apply_foundation:
enablement_work_name = self._work_format_str.format(op="enablement")
self._render_display(
category=WorkCategoryKey.ENABLE_IOT_OPS, active_step=WorkStepKey.WHAT_IF_ENABLEMENT
)
enablement_content, enablement_parameters = self._targets.get_ops_enablement_template()
self._deploy_template(
content=enablement_content,
parameters=enablement_parameters,
deployment_name=enablement_work_name,
what_if=True,
)
self._complete_step(
category=WorkCategoryKey.ENABLE_IOT_OPS,
completed_step=WorkStepKey.WHAT_IF_ENABLEMENT,
active_step=WorkStepKey.DEPLOY_ENABLEMENT,
)
enablement_poller = self._deploy_template(
content=enablement_content,
parameters=enablement_parameters,
deployment_name=enablement_work_name,
)
# Pattern needs work, it is this way to dynamically update UI
self._display.categories[WorkCategoryKey.ENABLE_IOT_OPS][0].title = (
f"[link={self._get_deployment_link(enablement_work_name)}]"
f"{self._display.categories[WorkCategoryKey.ENABLE_IOT_OPS][0].title}[/link]"
)
self._render_display(category=WorkCategoryKey.ENABLE_IOT_OPS)
_ = wait_for_terminal_state(enablement_poller)
self._complete_step(
category=WorkCategoryKey.ENABLE_IOT_OPS, completed_step=WorkStepKey.DEPLOY_ENABLEMENT
)
# Deploy IoT Ops workflow
if self._targets.instance_name:
# Ensure schema registry exists.
self.resource_client.resources.get_by_id(
resource_id=self._targets.schema_registry_resource_id,
api_version=REGISTRY_PREVIEW_API_VERSION,
)
self._process_extension_dependencies()
dependency_ext_ids = [
self.ops_extension_dependencies[ext]["id"] for ext in [EXTENSION_TYPE_PLATFORM, EXTENSION_TYPE_SSC]
]
self._render_display(category=WorkCategoryKey.DEPLOY_IOT_OPS, active_step=WorkStepKey.DEPLOY_INSTANCE)
self._create_or_update_custom_location(
extension_ids=[self.ops_extension_dependencies[EXTENSION_TYPE_PLATFORM]["id"]]
)
instance_content, instance_parameters = self._targets.get_ops_instance_template(
cl_extension_ids=dependency_ext_ids,
phase=InstancePhase.EXT,
)
instance_work_name = self._work_format_str.format(op="extension")
wait_for_terminal_state(
self._deploy_template(
content=instance_content,
parameters=instance_parameters,
deployment_name=instance_work_name,
)
)
self._create_or_update_custom_location(extension_ids=dependency_ext_ids + [self.ops_extension["id"]])
sleep(CATCH_UP_SEC)
instance_work_name = self._work_format_str.format(op="instance")
instance_content, instance_parameters = self._targets.get_ops_instance_template(
cl_extension_ids=dependency_ext_ids,
phase=InstancePhase.INSTANCE,
)
wait_for_terminal_state(
self._deploy_template(
content=instance_content,
parameters=instance_parameters,
deployment_name=instance_work_name,
)
)
sleep(CATCH_UP_SEC)
self._complete_step(
category=WorkCategoryKey.DEPLOY_IOT_OPS,
completed_step=WorkStepKey.DEPLOY_INSTANCE,
active_step=WorkStepKey.DEPLOY_RESOURCES,
)
instance_content, instance_parameters = self._targets.get_ops_instance_template(
cl_extension_ids=dependency_ext_ids, phase=InstancePhase.RESOURCES
)
instance_work_name = self._work_format_str.format(op="resources")
instance_poller = self._deploy_template(
content=instance_content,
parameters=instance_parameters,
deployment_name=instance_work_name,
)
# Pattern needs work, it is this way to dynamically update UI
self._display.categories[WorkCategoryKey.DEPLOY_IOT_OPS][0].title = (
f"[link={self._get_deployment_link(instance_work_name)}]"
f"{self._display.categories[WorkCategoryKey.DEPLOY_IOT_OPS][0].title}[/link]"
)
self._render_display(category=WorkCategoryKey.DEPLOY_IOT_OPS)
wait_for_terminal_state(instance_poller)
self._apply_sr_role_assignment()
self._complete_step(
category=WorkCategoryKey.DEPLOY_IOT_OPS,
completed_step=WorkStepKey.DEPLOY_INSTANCE,
)
return self._get_user_result()
except HttpResponseError as e:
# TODO: repeated error messages.
raise AzureResponseError(e.message)
except KeyboardInterrupt:
return
finally:
self._stop_display()