We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] Failed workspace deployment: only one positional argument may be specified, the installation name, but multiple were received
**Describe the bug**
Workspace deployment failed with:

```
only one positional argument may be specified, the installation name, but multiple were received [mrtre17-9c1d X secure project]
```

**Steps to reproduce**

Deployed a workspace with following values:

```
"azure_location":"westeurope",
"workspace_id":"9c1d",
"tre_id":"mrtre17",
"address_space":"192.168.10.0/24",
"display_name":"Project X",
"description":"Very secure project"
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `resource_processor/vmss_porter/runner.py`
Content:
```
1 import os
2 import sys
3 import json
4 import socket
5 import asyncio
6 import logging
7 
8 from shared.logging import disable_unwanted_loggers, initialize_logging, get_message_id_logger  # pylint: disable=import-error # noqa
9 from resources import strings  # pylint: disable=import-error # noqa
10 from contextlib import asynccontextmanager
11 from azure.servicebus import ServiceBusMessage
12 from azure.servicebus.aio import ServiceBusClient, AutoLockRenewer
13 from azure.identity.aio import DefaultAzureCredential
14 
15 logger_adapter = initialize_logging(logging.INFO, socket.gethostname())
16 disable_unwanted_loggers()
17 
18 
19 @asynccontextmanager
20 async def default_credentials(msi_id):
21     """
22     Context manager which yields the default credentials.
23     """
24     credential = DefaultAzureCredential(managed_identity_client_id=msi_id) if msi_id else DefaultAzureCredential()
25     yield credential
26     await credential.close()
27 
28 
29 async def receive_message(env_vars, service_bus_client):
30     """
31     This method is an async generator which receives messages from service bus
32     and yields those messages. If the yielded function return True the message is
33     marked complete.
34     """
35     async with service_bus_client:
36         q_name = env_vars["resource_request_queue"]
37         renewer = AutoLockRenewer(max_lock_renewal_duration=1800)
38         receiver = service_bus_client.get_queue_receiver(queue_name=q_name, auto_lock_renewer=renewer)
39 
40         async with receiver:
41             received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5)
42 
43             for msg in received_msgs:
44                 result = True
45                 message = ""
46 
47                 try:
48                     message = json.loads(str(msg))
49                     result = (yield message)
50                 except (json.JSONDecodeError) as e:
51                     logging.error(f"Received bad service bus resource request message: {e}")
52                 if result:
53                     logging.info(f"Resource request for {message} is complete")
54                 else:
55                     logging.error('Message processing failed!')
56                 logger_adapter.info(f"Message with id = {message['id']} processed as {result} and marked complete.")
57                 await receiver.complete_message(msg)
58 
59 
60 def azure_login_command(env_vars):
61     local_login = f"az login --service-principal --username {env_vars['arm_client_id']} --password {env_vars['arm_client_secret']} --tenant {env_vars['arm_tenant_id']}"
62     vmss_login = f"az login --identity -u {env_vars['vmss_msi_id']}"
63     command = vmss_login if env_vars['vmss_msi_id'] else local_login
64     return command
65 
66 
67 async def filter_parameters_not_needed_by_porter(msg_body, env_vars):
68     parameters = msg_body["parameters"]
69     command = [f"{azure_login_command(env_vars)} >/dev/null && \
70         az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} >/dev/null && \
71         porter explain --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']} -ojson"]
72     proc = await asyncio.create_subprocess_shell(
73         ''.join(command),
74         stdout=asyncio.subprocess.PIPE,
75         stderr=asyncio.subprocess.PIPE,
76         env=porter_envs(env_vars))
77 
78     stdout, stderr = await proc.communicate()
79     logging.info(f'[{command!r} exited with {proc.returncode}]')
80     result_stdout = None
81     result_stderr = None
82     if stdout:
83         result_stdout = stdout.decode()
84         porter_explain_parameters = json.loads(result_stdout)["parameters"]
85         items = [item["name"] for item in porter_explain_parameters]
86         porter_keys = set(items).intersection(set(parameters.keys()))
87         return porter_keys
88     if stderr:
89         result_stderr = stderr.decode()
90         logger_adapter.info('[stderr]')
91         for string in result_stderr.split('\n'):
92             logger_adapter.info(str(string))
93 
94     return parameters.keys()
95 
96 
97 async def build_porter_command(msg_body, env_vars):
98     porter_parameters = ""
99 
100     porter_keys = await filter_parameters_not_needed_by_porter(msg_body, env_vars)
101     for parameter in porter_keys:
102         porter_parameters = porter_parameters + f" --param {parameter}={msg_body['parameters'][parameter]}"
103 
104     installation_id = msg_body['parameters']['tre_id'] + "-" + msg_body['parameters']['workspace_id']
105 
106     porter_parameters = porter_parameters + f" --param tfstate_container_name={env_vars['tfstate_container_name']}"
107     porter_parameters = porter_parameters + f" --param tfstate_resource_group_name={env_vars['tfstate_resource_group_name']}"
108     porter_parameters = porter_parameters + f" --param tfstate_storage_account_name={env_vars['tfstate_storage_account_name']}"
109     porter_parameters = porter_parameters + f" --param arm_use_msi={env_vars['arm_use_msi']}"
110 
111     command_line = [f"{azure_login_command(env_vars)} && az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} && porter "
112                     f"{msg_body['action']} {installation_id} "
113                     f" --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']}"
114                     f" {porter_parameters} --cred ./vmss_porter/azure.json --allow-docker-host-access"
115                     f" && porter show {installation_id}"]
116     return command_line
117 
118 
119 def porter_envs(env_var):
120     porter_env_vars = {}
121     porter_env_vars["HOME"] = os.environ['HOME']
122     porter_env_vars["PATH"] = os.environ['PATH']
123     porter_env_vars["ARM_CLIENT_ID"] = env_var["arm_client_id"]
124     porter_env_vars["ARM_CLIENT_SECRET"] = env_var["arm_client_secret"]
125     porter_env_vars["ARM_SUBSCRIPTION_ID"] = env_var["arm_subscription_id"]
126     porter_env_vars["ARM_TENANT_ID"] = env_var["arm_tenant_id"]
127 
128     return porter_env_vars
129 
130 
131 async def run_porter(command, env_vars):
132     proc = await asyncio.create_subprocess_shell(
133         ''.join(command),
134         stdout=asyncio.subprocess.PIPE,
135         stderr=asyncio.subprocess.PIPE,
136         env=porter_envs(env_vars))
137 
138     stdout, stderr = await proc.communicate()
139     logging.info(f'[{command!r} exited with {proc.returncode}]')
140     result_stdout = None
141     result_stderr = None
142     if stdout:
143         result_stdout = stdout.decode()
144         logger_adapter.info('[stdout]')
145         for string in result_stdout.split('\n'):
146             if len(string) != 0:
147                 logger_adapter.info(str(string))
148     if stderr:
149         result_stderr = stderr.decode()
150         logger_adapter.info('[stderr]')
151         for string in result_stderr.split('\n'):
152             if len(string) != 0:
153                 logger_adapter.info(str(string))
154 
155     return (proc.returncode, result_stdout, result_stderr)
156 
157 
158 def service_bus_message_generator(sb_message, status, deployment_message):
159     installation_id = sb_message['parameters']['tre_id'] + "-" + sb_message['parameters']['workspace_id']
160     resource_request_message = json.dumps({
161         "id": sb_message["id"],
162         "status": status,
163         "message": f"{installation_id}: {deployment_message}"
164     })
165     return resource_request_message
166 
167 
168 async def deploy_porter_bundle(msg_body, sb_client, env_vars, message_logger_adapter):
169     installation_id = msg_body['parameters']['tre_id'] + "-" + msg_body['parameters']['workspace_id']
170     message_logger_adapter.info(f"{installation_id}: Deployment job configuration starting")
171     sb_sender = sb_client.get_queue_sender(queue_name=env_vars["deployment_status_queue"])
172     resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYING, "Deployment job starting")
173     await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
174     porter_command = await build_porter_command(msg_body, env_vars)
175     returncode, _, err = await run_porter(porter_command, env_vars)
176     if returncode != 0:
177         error_message = "Error context message = " + " ".join(err.split('\n'))
178         resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_FAILED, error_message)
179         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
180         message_logger_adapter.info(f"{installation_id}: Deployment job configuration failed error = {error_message}")
181         return False
182     else:
183         success_message = "Workspace was deployed successfully..."
184         resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYED, success_message)
185         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
186         message_logger_adapter.info(f"{installation_id}: {success_message}")
187         return True
188 
189 
190 async def runner(env_vars):
191     msi_id = env_vars["vmss_msi_id"]
192     service_bus_namespace = env_vars["service_bus_namespace"]
193     async with default_credentials(msi_id) as credential:
194         service_bus_client = ServiceBusClient(service_bus_namespace, credential)
195         logger_adapter.info("Starting message receiving loop...")
196         while True:
197             logger_adapter.info("Checking for new messages...")
198             receive_message_gen = receive_message(env_vars, service_bus_client)
199             try:
200                 async for message in receive_message_gen:
201                     logger_adapter.info(f"Message received for id={message['id']}")
202                     message_logger_adapter = get_message_id_logger(message['id'])  # logger includes message id in every entry.
203                     result = await deploy_porter_bundle(message, service_bus_client, env_vars, message_logger_adapter)
204                     await receive_message_gen.asend(result)
205             except StopAsyncIteration:  # the async generator when finished signals end with this exception.
206                 pass
207             logger_adapter.info("All messages done sleeping...")
208             await asyncio.sleep(60)
209 
210 
211 def read_env_vars():
212     env_vars = {
213         # Needed for local dev
214         "app_id": os.environ.get("AZURE_CLIENT_ID", None),
215         "app_password": os.environ.get("AZURE_CLIENT_SECRET", None),
216 
217         "registry_server": os.environ["REGISTRY_SERVER"],
218         "tfstate_container_name": os.environ['TERRAFORM_STATE_CONTAINER_NAME'],
219         "tfstate_resource_group_name": os.environ['MGMT_RESOURCE_GROUP_NAME'],
220         "tfstate_storage_account_name": os.environ['MGMT_STORAGE_ACCOUNT_NAME'],
221         "deployment_status_queue": os.environ['SERVICE_BUS_DEPLOYMENT_STATUS_UPDATE_QUEUE'],
222         "resource_request_queue": os.environ['SERVICE_BUS_RESOURCE_REQUEST_QUEUE'],
223         "service_bus_namespace": os.environ['SERVICE_BUS_FULLY_QUALIFIED_NAMESPACE'],
224         "vmss_msi_id": os.environ.get('VMSS_MSI_ID', None),
225 
226         # Needed for running porter
227         "arm_use_msi": os.environ["ARM_USE_MSI"],
228         "arm_subscription_id": os.environ['ARM_SUBSCRIPTION_ID'],
229         "arm_client_id": os.environ["ARM_CLIENT_ID"],
230         "arm_tenant_id": os.environ["ARM_TENANT_ID"]
231     }
232 
233     env_vars["arm_client_secret"] = os.environ["ARM_CLIENT_SECRET"] if env_vars["arm_use_msi"] == "false" else ""
234 
235     return env_vars
236 
237 
238 if __name__ == "__main__":
239     try:
240         env_vars = read_env_vars()
241     except KeyError as e:
242         logger_adapter.error(f"Environment variable {e} is not set correctly...Exiting")
243         sys.exit(1)
244     logger_adapter.info("Started processor")
245     asyncio.run(runner(env_vars))
246
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/resource_processor/vmss_porter/runner.py b/resource_processor/vmss_porter/runner.py
--- a/resource_processor/vmss_porter/runner.py
+++ b/resource_processor/vmss_porter/runner.py
@@ -99,7 +99,7 @@
 
     porter_keys = await filter_parameters_not_needed_by_porter(msg_body, env_vars)
     for parameter in porter_keys:
-        porter_parameters = porter_parameters + f" --param {parameter}={msg_body['parameters'][parameter]}"
+        porter_parameters = porter_parameters + f" --param {parameter}=\"{msg_body['parameters'][parameter]}\""
 
     installation_id = msg_body['parameters']['tre_id'] + "-" + msg_body['parameters']['workspace_id']
 
@@ -109,7 +109,7 @@
     porter_parameters = porter_parameters + f" --param arm_use_msi={env_vars['arm_use_msi']}"
 
     command_line = [f"{azure_login_command(env_vars)} && az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} && porter "
-                    f"{msg_body['action']} {installation_id} "
+                    f"{msg_body['action']} \"{installation_id}\" "
                     f" --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']}"
                     f" {porter_parameters} --cred ./vmss_porter/azure.json --allow-docker-host-access"
                     f" && porter show {installation_id}"]
@@ -174,7 +174,7 @@
     porter_command = await build_porter_command(msg_body, env_vars)
     returncode, _, err = await run_porter(porter_command, env_vars)
     if returncode != 0:
-        error_message = "Error context message = " + " ".join(err.split('\n'))
+        error_message = "Error context message = " + " ".join(err.split('\n')) + " ; Command executed: ".join(porter_command)
         resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_FAILED, error_message)
         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
         message_logger_adapter.info(f"{installation_id}: Deployment job configuration failed error = {error_message}")

{"golden_diff": "diff --git a/resource_processor/vmss_porter/runner.py b/resource_processor/vmss_porter/runner.py\n--- a/resource_processor/vmss_porter/runner.py\n+++ b/resource_processor/vmss_porter/runner.py\n@@ -99,7 +99,7 @@\n \n     porter_keys = await filter_parameters_not_needed_by_porter(msg_body, env_vars)\n     for parameter in porter_keys:\n-        porter_parameters = porter_parameters + f\" --param {parameter}={msg_body['parameters'][parameter]}\"\n+        porter_parameters = porter_parameters + f\" --param {parameter}=\\\"{msg_body['parameters'][parameter]}\\\"\"\n \n     installation_id = msg_body['parameters']['tre_id'] + \"-\" + msg_body['parameters']['workspace_id']\n \n@@ -109,7 +109,7 @@\n     porter_parameters = porter_parameters + f\" --param arm_use_msi={env_vars['arm_use_msi']}\"\n \n     command_line = [f\"{azure_login_command(env_vars)} && az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} && porter \"\n-                    f\"{msg_body['action']} {installation_id} \"\n+                    f\"{msg_body['action']} \\\"{installation_id}\\\" \"\n                     f\" --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']}\"\n                     f\" {porter_parameters} --cred ./vmss_porter/azure.json --allow-docker-host-access\"\n                     f\" && porter show {installation_id}\"]\n@@ -174,7 +174,7 @@\n     porter_command = await build_porter_command(msg_body, env_vars)\n     returncode, _, err = await run_porter(porter_command, env_vars)\n     if returncode != 0:\n-        error_message = \"Error context message = \" + \" \".join(err.split('\\n'))\n+        error_message = \"Error context message = \" + \" \".join(err.split('\\n')) + \" ; Command executed: \".join(porter_command)\n         resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_FAILED, error_message)\n         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n         message_logger_adapter.info(f\"{installation_id}: Deployment job configuration failed error = {error_message}\")\n", "issue": "[BUG] Failed workspace deployment: only one positional argument may be specified, the installation name, but multiple were received\n**Describe the bug**\r\nWorkspace deployment failed with:\r\n\r\n```\r\nonly one positional argument may be specified, the installation name, but multiple were received [mrtre17-9c1d X secure project]\r\n```\r\n\r\n**Steps to reproduce**\r\n\r\nDeployed a workspace with following values:\r\n\r\n```\r\n\"azure_location\":\"westeurope\",\r\n\"workspace_id\":\"9c1d\",\r\n\"tre_id\":\"mrtre17\",\r\n\"address_space\":\"192.168.10.0/24\",\r\n\"display_name\":\"Project X\",\r\n\"description\":\"Very secure project\"\r\n```\n", "before_files": [{"content": "import os\nimport sys\nimport json\nimport socket\nimport asyncio\nimport logging\n\nfrom shared.logging import disable_unwanted_loggers, initialize_logging, get_message_id_logger  # pylint: disable=import-error # noqa\nfrom resources import strings  # pylint: disable=import-error # noqa\nfrom contextlib import asynccontextmanager\nfrom azure.servicebus import ServiceBusMessage\nfrom azure.servicebus.aio import ServiceBusClient, AutoLockRenewer\nfrom azure.identity.aio import DefaultAzureCredential\n\nlogger_adapter = initialize_logging(logging.INFO, socket.gethostname())\ndisable_unwanted_loggers()\n\n\n@asynccontextmanager\nasync def default_credentials(msi_id):\n    \"\"\"\n    Context manager which yields the default credentials.\n    \"\"\"\n    credential = DefaultAzureCredential(managed_identity_client_id=msi_id) if msi_id else DefaultAzureCredential()\n    yield credential\n    await credential.close()\n\n\nasync def receive_message(env_vars, service_bus_client):\n    \"\"\"\n    This method is an async generator which receives messages from service bus\n    and yields those messages. If the yielded function return True the message is\n    marked complete.\n    \"\"\"\n    async with service_bus_client:\n        q_name = env_vars[\"resource_request_queue\"]\n        renewer = AutoLockRenewer(max_lock_renewal_duration=1800)\n        receiver = service_bus_client.get_queue_receiver(queue_name=q_name, auto_lock_renewer=renewer)\n\n        async with receiver:\n            received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5)\n\n            for msg in received_msgs:\n                result = True\n                message = \"\"\n\n                try:\n                    message = json.loads(str(msg))\n                    result = (yield message)\n                except (json.JSONDecodeError) as e:\n                    logging.error(f\"Received bad service bus resource request message: {e}\")\n                if result:\n                    logging.info(f\"Resource request for {message} is complete\")\n                else:\n                    logging.error('Message processing failed!')\n                logger_adapter.info(f\"Message with id = {message['id']} processed as {result} and marked complete.\")\n                await receiver.complete_message(msg)\n\n\ndef azure_login_command(env_vars):\n    local_login = f\"az login --service-principal --username {env_vars['arm_client_id']} --password {env_vars['arm_client_secret']} --tenant {env_vars['arm_tenant_id']}\"\n    vmss_login = f\"az login --identity -u {env_vars['vmss_msi_id']}\"\n    command = vmss_login if env_vars['vmss_msi_id'] else local_login\n    return command\n\n\nasync def filter_parameters_not_needed_by_porter(msg_body, env_vars):\n    parameters = msg_body[\"parameters\"]\n    command = [f\"{azure_login_command(env_vars)} >/dev/null && \\\n        az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} >/dev/null && \\\n        porter explain --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']} -ojson\"]\n    proc = await asyncio.create_subprocess_shell(\n        ''.join(command),\n        stdout=asyncio.subprocess.PIPE,\n        stderr=asyncio.subprocess.PIPE,\n        env=porter_envs(env_vars))\n\n    stdout, stderr = await proc.communicate()\n    logging.info(f'[{command!r} exited with {proc.returncode}]')\n    result_stdout = None\n    result_stderr = None\n    if stdout:\n        result_stdout = stdout.decode()\n        porter_explain_parameters = json.loads(result_stdout)[\"parameters\"]\n        items = [item[\"name\"] for item in porter_explain_parameters]\n        porter_keys = set(items).intersection(set(parameters.keys()))\n        return porter_keys\n    if stderr:\n        result_stderr = stderr.decode()\n        logger_adapter.info('[stderr]')\n        for string in result_stderr.split('\\n'):\n            logger_adapter.info(str(string))\n\n    return parameters.keys()\n\n\nasync def build_porter_command(msg_body, env_vars):\n    porter_parameters = \"\"\n\n    porter_keys = await filter_parameters_not_needed_by_porter(msg_body, env_vars)\n    for parameter in porter_keys:\n        porter_parameters = porter_parameters + f\" --param {parameter}={msg_body['parameters'][parameter]}\"\n\n    installation_id = msg_body['parameters']['tre_id'] + \"-\" + msg_body['parameters']['workspace_id']\n\n    porter_parameters = porter_parameters + f\" --param tfstate_container_name={env_vars['tfstate_container_name']}\"\n    porter_parameters = porter_parameters + f\" --param tfstate_resource_group_name={env_vars['tfstate_resource_group_name']}\"\n    porter_parameters = porter_parameters + f\" --param tfstate_storage_account_name={env_vars['tfstate_storage_account_name']}\"\n    porter_parameters = porter_parameters + f\" --param arm_use_msi={env_vars['arm_use_msi']}\"\n\n    command_line = [f\"{azure_login_command(env_vars)} && az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} && porter \"\n                    f\"{msg_body['action']} {installation_id} \"\n                    f\" --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']}\"\n                    f\" {porter_parameters} --cred ./vmss_porter/azure.json --allow-docker-host-access\"\n                    f\" && porter show {installation_id}\"]\n    return command_line\n\n\ndef porter_envs(env_var):\n    porter_env_vars = {}\n    porter_env_vars[\"HOME\"] = os.environ['HOME']\n    porter_env_vars[\"PATH\"] = os.environ['PATH']\n    porter_env_vars[\"ARM_CLIENT_ID\"] = env_var[\"arm_client_id\"]\n    porter_env_vars[\"ARM_CLIENT_SECRET\"] = env_var[\"arm_client_secret\"]\n    porter_env_vars[\"ARM_SUBSCRIPTION_ID\"] = env_var[\"arm_subscription_id\"]\n    porter_env_vars[\"ARM_TENANT_ID\"] = env_var[\"arm_tenant_id\"]\n\n    return porter_env_vars\n\n\nasync def run_porter(command, env_vars):\n    proc = await asyncio.create_subprocess_shell(\n        ''.join(command),\n        stdout=asyncio.subprocess.PIPE,\n        stderr=asyncio.subprocess.PIPE,\n        env=porter_envs(env_vars))\n\n    stdout, stderr = await proc.communicate()\n    logging.info(f'[{command!r} exited with {proc.returncode}]')\n    result_stdout = None\n    result_stderr = None\n    if stdout:\n        result_stdout = stdout.decode()\n        logger_adapter.info('[stdout]')\n        for string in result_stdout.split('\\n'):\n            if len(string) != 0:\n                logger_adapter.info(str(string))\n    if stderr:\n        result_stderr = stderr.decode()\n        logger_adapter.info('[stderr]')\n        for string in result_stderr.split('\\n'):\n            if len(string) != 0:\n                logger_adapter.info(str(string))\n\n    return (proc.returncode, result_stdout, result_stderr)\n\n\ndef service_bus_message_generator(sb_message, status, deployment_message):\n    installation_id = sb_message['parameters']['tre_id'] + \"-\" + sb_message['parameters']['workspace_id']\n    resource_request_message = json.dumps({\n        \"id\": sb_message[\"id\"],\n        \"status\": status,\n        \"message\": f\"{installation_id}: {deployment_message}\"\n    })\n    return resource_request_message\n\n\nasync def deploy_porter_bundle(msg_body, sb_client, env_vars, message_logger_adapter):\n    installation_id = msg_body['parameters']['tre_id'] + \"-\" + msg_body['parameters']['workspace_id']\n    message_logger_adapter.info(f\"{installation_id}: Deployment job configuration starting\")\n    sb_sender = sb_client.get_queue_sender(queue_name=env_vars[\"deployment_status_queue\"])\n    resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYING, \"Deployment job starting\")\n    await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n    porter_command = await build_porter_command(msg_body, env_vars)\n    returncode, _, err = await run_porter(porter_command, env_vars)\n    if returncode != 0:\n        error_message = \"Error context message = \" + \" \".join(err.split('\\n'))\n        resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_FAILED, error_message)\n        await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n        message_logger_adapter.info(f\"{installation_id}: Deployment job configuration failed error = {error_message}\")\n        return False\n    else:\n        success_message = \"Workspace was deployed successfully...\"\n        resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYED, success_message)\n        await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n        message_logger_adapter.info(f\"{installation_id}: {success_message}\")\n        return True\n\n\nasync def runner(env_vars):\n    msi_id = env_vars[\"vmss_msi_id\"]\n    service_bus_namespace = env_vars[\"service_bus_namespace\"]\n    async with default_credentials(msi_id) as credential:\n        service_bus_client = ServiceBusClient(service_bus_namespace, credential)\n        logger_adapter.info(\"Starting message receiving loop...\")\n        while True:\n            logger_adapter.info(\"Checking for new messages...\")\n            receive_message_gen = receive_message(env_vars, service_bus_client)\n            try:\n                async for message in receive_message_gen:\n                    logger_adapter.info(f\"Message received for id={message['id']}\")\n                    message_logger_adapter = get_message_id_logger(message['id'])  # logger includes message id in every entry.\n                    result = await deploy_porter_bundle(message, service_bus_client, env_vars, message_logger_adapter)\n                    await receive_message_gen.asend(result)\n            except StopAsyncIteration:  # the async generator when finished signals end with this exception.\n                pass\n            logger_adapter.info(\"All messages done sleeping...\")\n            await asyncio.sleep(60)\n\n\ndef read_env_vars():\n    env_vars = {\n        # Needed for local dev\n        \"app_id\": os.environ.get(\"AZURE_CLIENT_ID\", None),\n        \"app_password\": os.environ.get(\"AZURE_CLIENT_SECRET\", None),\n\n        \"registry_server\": os.environ[\"REGISTRY_SERVER\"],\n        \"tfstate_container_name\": os.environ['TERRAFORM_STATE_CONTAINER_NAME'],\n        \"tfstate_resource_group_name\": os.environ['MGMT_RESOURCE_GROUP_NAME'],\n        \"tfstate_storage_account_name\": os.environ['MGMT_STORAGE_ACCOUNT_NAME'],\n        \"deployment_status_queue\": os.environ['SERVICE_BUS_DEPLOYMENT_STATUS_UPDATE_QUEUE'],\n        \"resource_request_queue\": os.environ['SERVICE_BUS_RESOURCE_REQUEST_QUEUE'],\n        \"service_bus_namespace\": os.environ['SERVICE_BUS_FULLY_QUALIFIED_NAMESPACE'],\n        \"vmss_msi_id\": os.environ.get('VMSS_MSI_ID', None),\n\n        # Needed for running porter\n        \"arm_use_msi\": os.environ[\"ARM_USE_MSI\"],\n        \"arm_subscription_id\": os.environ['ARM_SUBSCRIPTION_ID'],\n        \"arm_client_id\": os.environ[\"ARM_CLIENT_ID\"],\n        \"arm_tenant_id\": os.environ[\"ARM_TENANT_ID\"]\n    }\n\n    env_vars[\"arm_client_secret\"] = os.environ[\"ARM_CLIENT_SECRET\"] if env_vars[\"arm_use_msi\"] == \"false\" else \"\"\n\n    return env_vars\n\n\nif __name__ == \"__main__\":\n    try:\n        env_vars = read_env_vars()\n    except KeyError as e:\n        logger_adapter.error(f\"Environment variable {e} is not set correctly...Exiting\")\n        sys.exit(1)\n    logger_adapter.info(\"Started processor\")\n    asyncio.run(runner(env_vars))\n", "path": "resource_processor/vmss_porter/runner.py"}], "after_files": [{"content": "import os\nimport sys\nimport json\nimport socket\nimport asyncio\nimport logging\n\nfrom shared.logging import disable_unwanted_loggers, initialize_logging, get_message_id_logger  # pylint: disable=import-error # noqa\nfrom resources import strings  # pylint: disable=import-error # noqa\nfrom contextlib import asynccontextmanager\nfrom azure.servicebus import ServiceBusMessage\nfrom azure.servicebus.aio import ServiceBusClient, AutoLockRenewer\nfrom azure.identity.aio import DefaultAzureCredential\n\nlogger_adapter = initialize_logging(logging.INFO, socket.gethostname())\ndisable_unwanted_loggers()\n\n\n@asynccontextmanager\nasync def default_credentials(msi_id):\n    \"\"\"\n    Context manager which yields the default credentials.\n    \"\"\"\n    credential = DefaultAzureCredential(managed_identity_client_id=msi_id) if msi_id else DefaultAzureCredential()\n    yield credential\n    await credential.close()\n\n\nasync def receive_message(env_vars, service_bus_client):\n    \"\"\"\n    This method is an async generator which receives messages from service bus\n    and yields those messages. If the yielded function return True the message is\n    marked complete.\n    \"\"\"\n    async with service_bus_client:\n        q_name = env_vars[\"resource_request_queue\"]\n        renewer = AutoLockRenewer(max_lock_renewal_duration=1800)\n        receiver = service_bus_client.get_queue_receiver(queue_name=q_name, auto_lock_renewer=renewer)\n\n        async with receiver:\n            received_msgs = await receiver.receive_messages(max_message_count=10, max_wait_time=5)\n\n            for msg in received_msgs:\n                result = True\n                message = \"\"\n\n                try:\n                    message = json.loads(str(msg))\n                    result = (yield message)\n                except (json.JSONDecodeError) as e:\n                    logging.error(f\"Received bad service bus resource request message: {e}\")\n                if result:\n                    logging.info(f\"Resource request for {message} is complete\")\n                else:\n                    logging.error('Message processing failed!')\n                logger_adapter.info(f\"Message with id = {message['id']} processed as {result} and marked complete.\")\n                await receiver.complete_message(msg)\n\n\ndef azure_login_command(env_vars):\n    local_login = f\"az login --service-principal --username {env_vars['arm_client_id']} --password {env_vars['arm_client_secret']} --tenant {env_vars['arm_tenant_id']}\"\n    vmss_login = f\"az login --identity -u {env_vars['vmss_msi_id']}\"\n    command = vmss_login if env_vars['vmss_msi_id'] else local_login\n    return command\n\n\nasync def filter_parameters_not_needed_by_porter(msg_body, env_vars):\n    parameters = msg_body[\"parameters\"]\n    command = [f\"{azure_login_command(env_vars)} >/dev/null && \\\n        az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} >/dev/null && \\\n        porter explain --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']} -ojson\"]\n    proc = await asyncio.create_subprocess_shell(\n        ''.join(command),\n        stdout=asyncio.subprocess.PIPE,\n        stderr=asyncio.subprocess.PIPE,\n        env=porter_envs(env_vars))\n\n    stdout, stderr = await proc.communicate()\n    logging.info(f'[{command!r} exited with {proc.returncode}]')\n    result_stdout = None\n    result_stderr = None\n    if stdout:\n        result_stdout = stdout.decode()\n        porter_explain_parameters = json.loads(result_stdout)[\"parameters\"]\n        items = [item[\"name\"] for item in porter_explain_parameters]\n        porter_keys = set(items).intersection(set(parameters.keys()))\n        return porter_keys\n    if stderr:\n        result_stderr = stderr.decode()\n        logger_adapter.info('[stderr]')\n        for string in result_stderr.split('\\n'):\n            logger_adapter.info(str(string))\n\n    return parameters.keys()\n\n\nasync def build_porter_command(msg_body, env_vars):\n    porter_parameters = \"\"\n\n    porter_keys = await filter_parameters_not_needed_by_porter(msg_body, env_vars)\n    for parameter in porter_keys:\n        porter_parameters = porter_parameters + f\" --param {parameter}=\\\"{msg_body['parameters'][parameter]}\\\"\"\n\n    installation_id = msg_body['parameters']['tre_id'] + \"-\" + msg_body['parameters']['workspace_id']\n\n    porter_parameters = porter_parameters + f\" --param tfstate_container_name={env_vars['tfstate_container_name']}\"\n    porter_parameters = porter_parameters + f\" --param tfstate_resource_group_name={env_vars['tfstate_resource_group_name']}\"\n    porter_parameters = porter_parameters + f\" --param tfstate_storage_account_name={env_vars['tfstate_storage_account_name']}\"\n    porter_parameters = porter_parameters + f\" --param arm_use_msi={env_vars['arm_use_msi']}\"\n\n    command_line = [f\"{azure_login_command(env_vars)} && az acr login --name {env_vars['registry_server'].replace('.azurecr.io','')} && porter \"\n                    f\"{msg_body['action']} \\\"{installation_id}\\\" \"\n                    f\" --reference {env_vars['registry_server']}/{msg_body['name']}:v{msg_body['version']}\"\n                    f\" {porter_parameters} --cred ./vmss_porter/azure.json --allow-docker-host-access\"\n                    f\" && porter show {installation_id}\"]\n    return command_line\n\n\ndef porter_envs(env_var):\n    porter_env_vars = {}\n    porter_env_vars[\"HOME\"] = os.environ['HOME']\n    porter_env_vars[\"PATH\"] = os.environ['PATH']\n    porter_env_vars[\"ARM_CLIENT_ID\"] = env_var[\"arm_client_id\"]\n    porter_env_vars[\"ARM_CLIENT_SECRET\"] = env_var[\"arm_client_secret\"]\n    porter_env_vars[\"ARM_SUBSCRIPTION_ID\"] = env_var[\"arm_subscription_id\"]\n    porter_env_vars[\"ARM_TENANT_ID\"] = env_var[\"arm_tenant_id\"]\n\n    return porter_env_vars\n\n\nasync def run_porter(command, env_vars):\n    proc = await asyncio.create_subprocess_shell(\n        ''.join(command),\n        stdout=asyncio.subprocess.PIPE,\n        stderr=asyncio.subprocess.PIPE,\n        env=porter_envs(env_vars))\n\n    stdout, stderr = await proc.communicate()\n    logging.info(f'[{command!r} exited with {proc.returncode}]')\n    result_stdout = None\n    result_stderr = None\n    if stdout:\n        result_stdout = stdout.decode()\n        logger_adapter.info('[stdout]')\n        for string in result_stdout.split('\\n'):\n            if len(string) != 0:\n                logger_adapter.info(str(string))\n    if stderr:\n        result_stderr = stderr.decode()\n        logger_adapter.info('[stderr]')\n        for string in result_stderr.split('\\n'):\n            if len(string) != 0:\n                logger_adapter.info(str(string))\n\n    return (proc.returncode, result_stdout, result_stderr)\n\n\ndef service_bus_message_generator(sb_message, status, deployment_message):\n    installation_id = sb_message['parameters']['tre_id'] + \"-\" + sb_message['parameters']['workspace_id']\n    resource_request_message = json.dumps({\n        \"id\": sb_message[\"id\"],\n        \"status\": status,\n        \"message\": f\"{installation_id}: {deployment_message}\"\n    })\n    return resource_request_message\n\n\nasync def deploy_porter_bundle(msg_body, sb_client, env_vars, message_logger_adapter):\n    installation_id = msg_body['parameters']['tre_id'] + \"-\" + msg_body['parameters']['workspace_id']\n    message_logger_adapter.info(f\"{installation_id}: Deployment job configuration starting\")\n    sb_sender = sb_client.get_queue_sender(queue_name=env_vars[\"deployment_status_queue\"])\n    resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYING, \"Deployment job starting\")\n    await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n    porter_command = await build_porter_command(msg_body, env_vars)\n    returncode, _, err = await run_porter(porter_command, env_vars)\n    if returncode != 0:\n        error_message = \"Error context message = \" + \" \".join(err.split('\\n')) + \" ; Command executed: \".join(porter_command)\n        resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_FAILED, error_message)\n        await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n        message_logger_adapter.info(f\"{installation_id}: Deployment job configuration failed error = {error_message}\")\n        return False\n    else:\n        success_message = \"Workspace was deployed successfully...\"\n        resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYED, success_message)\n        await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n        message_logger_adapter.info(f\"{installation_id}: {success_message}\")\n        return True\n\n\nasync def runner(env_vars):\n    msi_id = env_vars[\"vmss_msi_id\"]\n    service_bus_namespace = env_vars[\"service_bus_namespace\"]\n    async with default_credentials(msi_id) as credential:\n        service_bus_client = ServiceBusClient(service_bus_namespace, credential)\n        logger_adapter.info(\"Starting message receiving loop...\")\n        while True:\n            logger_adapter.info(\"Checking for new messages...\")\n            receive_message_gen = receive_message(env_vars, service_bus_client)\n            try:\n                async for message in receive_message_gen:\n                    logger_adapter.info(f\"Message received for id={message['id']}\")\n                    message_logger_adapter = get_message_id_logger(message['id'])  # logger includes message id in every entry.\n                    result = await deploy_porter_bundle(message, service_bus_client, env_vars, message_logger_adapter)\n                    await receive_message_gen.asend(result)\n            except StopAsyncIteration:  # the async generator when finished signals end with this exception.\n                pass\n            logger_adapter.info(\"All messages done sleeping...\")\n            await asyncio.sleep(60)\n\n\ndef read_env_vars():\n    env_vars = {\n        # Needed for local dev\n        \"app_id\": os.environ.get(\"AZURE_CLIENT_ID\", None),\n        \"app_password\": os.environ.get(\"AZURE_CLIENT_SECRET\", None),\n\n        \"registry_server\": os.environ[\"REGISTRY_SERVER\"],\n        \"tfstate_container_name\": os.environ['TERRAFORM_STATE_CONTAINER_NAME'],\n        \"tfstate_resource_group_name\": os.environ['MGMT_RESOURCE_GROUP_NAME'],\n        \"tfstate_storage_account_name\": os.environ['MGMT_STORAGE_ACCOUNT_NAME'],\n        \"deployment_status_queue\": os.environ['SERVICE_BUS_DEPLOYMENT_STATUS_UPDATE_QUEUE'],\n        \"resource_request_queue\": os.environ['SERVICE_BUS_RESOURCE_REQUEST_QUEUE'],\n        \"service_bus_namespace\": os.environ['SERVICE_BUS_FULLY_QUALIFIED_NAMESPACE'],\n        \"vmss_msi_id\": os.environ.get('VMSS_MSI_ID', None),\n\n        # Needed for running porter\n        \"arm_use_msi\": os.environ[\"ARM_USE_MSI\"],\n        \"arm_subscription_id\": os.environ['ARM_SUBSCRIPTION_ID'],\n        \"arm_client_id\": os.environ[\"ARM_CLIENT_ID\"],\n        \"arm_tenant_id\": os.environ[\"ARM_TENANT_ID\"]\n    }\n\n    env_vars[\"arm_client_secret\"] = os.environ[\"ARM_CLIENT_SECRET\"] if env_vars[\"arm_use_msi\"] == \"false\" else \"\"\n\n    return env_vars\n\n\nif __name__ == \"__main__\":\n    try:\n        env_vars = read_env_vars()\n    except KeyError as e:\n        logger_adapter.error(f\"Environment variable {e} is not set correctly...Exiting\")\n        sys.exit(1)\n    logger_adapter.info(\"Started processor\")\n    asyncio.run(runner(env_vars))\n", "path": "resource_processor/vmss_porter/runner.py"}]}