We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Service Bus Sessions never terminate
After a processor receives a message on a session, it hangs onto that session indefinitely, blocking the thread and meaning other messages cannot be processed. 

We need to terminate the session after each message has been processed / errored out. 

--- 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 threading
2 from multiprocessing import Process
3 import json
4 import socket
5 import asyncio
6 import logging
7 import sys
8 from resources.commands import build_porter_command, build_porter_command_for_outputs
9 from shared.config import get_config
10 from resources.helpers import get_installation_id
11 from resources.httpserver import start_server
12 
13 from shared.logging import disable_unwanted_loggers, initialize_logging, get_message_id_logger, shell_output_logger  # pylint: disable=import-error # noqa
14 from resources import strings, statuses  # pylint: disable=import-error # noqa
15 from contextlib import asynccontextmanager
16 from azure.servicebus import ServiceBusMessage, NEXT_AVAILABLE_SESSION
17 from azure.servicebus.exceptions import OperationTimeoutError, ServiceBusConnectionError
18 from azure.servicebus.aio import ServiceBusClient, AutoLockRenewer
19 from azure.identity.aio import DefaultAzureCredential
20 
21 
22 # Initialise logging
23 logger_adapter = initialize_logging(logging.INFO, socket.gethostname())
24 disable_unwanted_loggers()
25 
26 # Initialise config
27 try:
28     config = get_config(logger_adapter)
29 except KeyError as e:
30     logger_adapter.error(f"Environment variable {e} is not set correctly...Exiting")
31     sys.exit(1)
32 
33 
34 @asynccontextmanager
35 async def default_credentials(msi_id):
36     """
37     Context manager which yields the default credentials.
38     """
39     credential = DefaultAzureCredential(managed_identity_client_id=msi_id) if msi_id else DefaultAzureCredential()
40     yield credential
41     await credential.close()
42 
43 
44 async def receive_message(service_bus_client):
45     """
46     This method is run per process. Each process will connect to service bus and try to establish a session.
47     If messages are there, the process will continue to receive all the messages associated with that session.
48     If no messages are there, the session connection will time out, sleep, and retry.
49     """
50     q_name = config["resource_request_queue"]
51 
52     while True:
53         try:
54             logger_adapter.info("Looking for new session...")
55             async with service_bus_client.get_queue_receiver(queue_name=q_name, session_id=NEXT_AVAILABLE_SESSION) as receiver:
56                 logger_adapter.info("Got a session containing messages")
57                 async with AutoLockRenewer() as renewer:
58                     # allow a message to be auto lock renewed for up to an hour
59                     renewer.register(receiver, receiver.session, max_lock_renewal_duration=3600)
60 
61                     async for msg in receiver:
62                         result = True
63                         message = ""
64 
65                         try:
66                             message = json.loads(str(msg))
67                             logger_adapter.info(f"Message received for resource_id={message['id']}, operation_id={message['operationId']}")
68                             message_logger_adapter = get_message_id_logger(message['operationId'])  # correlate messages per operation
69                             result = await invoke_porter_action(message, service_bus_client, message_logger_adapter)
70                         except (json.JSONDecodeError) as e:
71                             logging.error(f"Received bad service bus resource request message: {e}")
72 
73                         if result:
74                             logging.info(f"Resource request for {message} is complete")
75                         else:
76                             logging.error('Message processing failed!')
77 
78                         logger_adapter.info(f"Message with id = {message['id']} processed as {result} and marked complete.")
79                         await receiver.complete_message(msg)
80 
81         except OperationTimeoutError:
82             # Timeout occurred whilst connecting to a session - this is expected and indicates no non-empty sessions are available
83             logger_adapter.info("No sessions for this process. Sleeping 30s then will look again...")
84 
85         except ServiceBusConnectionError:
86             # Occasionally there will be a transient / network-level error in connecting to SB.
87             logger_adapter.info("Unknown Service Bus connection error. Sleeping and will retry...")
88 
89         except Exception:
90             # Catch all other exceptions, log them via .exception to get the stack trace, sleep, and reconnect
91             logger_adapter.exception("Unknown exception. Sleeping and will retry...")
92 
93         finally:
94             await asyncio.sleep(30)
95 
96 
97 async def run_porter(command):
98     """
99     Run a Porter command
100     """
101     proc = await asyncio.create_subprocess_shell(
102         ''.join(command),
103         stdout=asyncio.subprocess.PIPE,
104         stderr=asyncio.subprocess.PIPE,
105         env=config["porter_env"])
106 
107     stdout, stderr = await proc.communicate()
108     logging.info(f'run porter exited with {proc.returncode}')
109     result_stdout = None
110     result_stderr = None
111 
112     if stdout:
113         result_stdout = stdout.decode()
114         shell_output_logger(result_stdout, '[stdout]', logger_adapter, logging.INFO)
115 
116     if stderr:
117         result_stderr = stderr.decode()
118         shell_output_logger(result_stderr, '[stderr]', logger_adapter, logging.WARN)
119 
120     return (proc.returncode, result_stdout, result_stderr)
121 
122 
123 def service_bus_message_generator(sb_message, status, deployment_message, outputs=None):
124     """
125     Generate a resource request message
126     """
127     installation_id = get_installation_id(sb_message)
128     message_dict = {
129         "operationId": sb_message["operationId"],
130         "id": sb_message["id"],
131         "status": status,
132         "message": f"{installation_id}: {deployment_message}"}
133 
134     if outputs is not None:
135         message_dict["outputs"] = outputs
136 
137     resource_request_message = json.dumps(message_dict)
138     return resource_request_message
139 
140 
141 async def invoke_porter_action(msg_body, sb_client, message_logger_adapter) -> bool:
142     """
143     Handle resource message by invoking specified porter action (i.e. install, uninstall)
144     """
145     installation_id = get_installation_id(msg_body)
146     action = msg_body["action"]
147     message_logger_adapter.info(f"{installation_id}: {action} action starting...")
148     sb_sender = sb_client.get_queue_sender(queue_name=config["deployment_status_queue"])
149 
150     # If the action is install/upgrade, post message on sb queue to start a deployment job
151     if action == "install" or action == "upgrade":
152         resource_request_message = service_bus_message_generator(msg_body, strings.RESOURCE_STATUS_DEPLOYING, "Deployment job starting")
153         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
154 
155     # Build and run porter command (flagging if its a built-in action or custom so we can adapt porter command appropriately)
156     is_custom_action = action not in ["install", "upgrade", "uninstall"]
157     porter_command = await build_porter_command(config, message_logger_adapter, msg_body, is_custom_action)
158     returncode, _, err = await run_porter(porter_command)
159 
160     # Handle command output
161     if returncode != 0:
162         error_message = "Error context message = " + " ".join(err.split('\n')) + " ; Command executed: ".join(porter_command)
163         resource_request_message = service_bus_message_generator(msg_body, statuses.failed_status_string_for[action], error_message)
164 
165         # Post message on sb queue to notify receivers of action failure
166         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
167         message_logger_adapter.info(f"{installation_id}: Porter action failed with error = {error_message}")
168         return False
169 
170     else:
171         # Get the outputs
172         # TODO: decide if this should "fail" the deployment
173         _, outputs = await get_porter_outputs(msg_body, message_logger_adapter)
174 
175         success_message = f"{action} action completed successfully."
176         resource_request_message = service_bus_message_generator(msg_body, statuses.pass_status_string_for[action], success_message, outputs)
177 
178         await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body["id"]))
179         message_logger_adapter.info(f"{installation_id}: {success_message}")
180         return True
181 
182 
183 async def get_porter_outputs(msg_body, message_logger_adapter):
184     """
185     Get outputs JSON from a Porter command
186     """
187     porter_command = await build_porter_command_for_outputs(msg_body)
188     returncode, stdout, err = await run_porter(porter_command)
189 
190     if returncode != 0:
191         error_message = "Error context message = " + " ".join(err.split('\n'))
192         message_logger_adapter.info(f"{get_installation_id(msg_body)}: Failed to get outputs with error = {error_message}")
193         return False, ""
194     else:
195         outputs_json = {}
196         try:
197             outputs_json = json.loads(stdout)
198             message_logger_adapter.info(f"Got outputs as json: {outputs_json}")
199         except ValueError:
200             message_logger_adapter.error(f"Got outputs invalid json: {stdout}")
201 
202         return True, outputs_json
203 
204 
205 async def runner():
206     async with default_credentials(config["vmss_msi_id"]) as credential:
207         service_bus_client = ServiceBusClient(config["service_bus_namespace"], credential)
208         await receive_message(service_bus_client)
209 
210 
211 def start_runner_process():
212     asyncio.ensure_future(runner())
213     event_loop = asyncio.get_event_loop()
214     event_loop.run_forever()
215     logger_adapter.info("Started resource processor")
216 
217 
218 if __name__ == "__main__":
219     httpserver_thread = threading.Thread(target=start_server)
220     httpserver_thread.start()
221     logger_adapter.info("Started http server")
222 
223     logger_adapter.info(f'Starting {str(config["number_processes_int"])} processes...')
224     for i in range(config["number_processes_int"]):
225         logger_adapter.info(f'Starting process {str(i)}')
226         process = Process(target=start_runner_process)
227         process.start()
228
```
--- 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
@@ -52,10 +52,11 @@
     while True:
         try:
             logger_adapter.info("Looking for new session...")
-            async with service_bus_client.get_queue_receiver(queue_name=q_name, session_id=NEXT_AVAILABLE_SESSION) as receiver:
+            # max_wait_time=1 -> don't hold the session open after processing of the message has finished
+            async with service_bus_client.get_queue_receiver(queue_name=q_name, max_wait_time=1, session_id=NEXT_AVAILABLE_SESSION) as receiver:
                 logger_adapter.info("Got a session containing messages")
                 async with AutoLockRenewer() as renewer:
-                    # allow a message to be auto lock renewed for up to an hour
+                    # allow a session to be auto lock renewed for up to an hour - if it's processing a message
                     renewer.register(receiver, receiver.session, max_lock_renewal_duration=3600)
 
                     async for msg in receiver:
@@ -75,23 +76,23 @@
                         else:
                             logging.error('Message processing failed!')
 
-                        logger_adapter.info(f"Message with id = {message['id']} processed as {result} and marked complete.")
+                        logger_adapter.info(f"Message for resource_id={message['id']}, operation_id={message['operationId']} processed as {result} and marked complete.")
                         await receiver.complete_message(msg)
 
+                    logger_adapter.info("Closing session")
+                    await renewer.close()
+
         except OperationTimeoutError:
             # Timeout occurred whilst connecting to a session - this is expected and indicates no non-empty sessions are available
-            logger_adapter.info("No sessions for this process. Sleeping 30s then will look again...")
+            logger_adapter.info("No sessions for this process. Will look again...")
 
         except ServiceBusConnectionError:
             # Occasionally there will be a transient / network-level error in connecting to SB.
-            logger_adapter.info("Unknown Service Bus connection error. Sleeping and will retry...")
+            logger_adapter.info("Unknown Service Bus connection error. Will retry...")
 
         except Exception:
             # Catch all other exceptions, log them via .exception to get the stack trace, sleep, and reconnect
-            logger_adapter.exception("Unknown exception. Sleeping and will retry...")
-
-        finally:
-            await asyncio.sleep(30)
+            logger_adapter.exception("Unknown exception. Will retry...")
 
 
 async def run_porter(command):

{"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@@ -52,10 +52,11 @@\n     while True:\n         try:\n             logger_adapter.info(\"Looking for new session...\")\n-            async with service_bus_client.get_queue_receiver(queue_name=q_name, session_id=NEXT_AVAILABLE_SESSION) as receiver:\n+            # max_wait_time=1 -> don't hold the session open after processing of the message has finished\n+            async with service_bus_client.get_queue_receiver(queue_name=q_name, max_wait_time=1, session_id=NEXT_AVAILABLE_SESSION) as receiver:\n                 logger_adapter.info(\"Got a session containing messages\")\n                 async with AutoLockRenewer() as renewer:\n-                    # allow a message to be auto lock renewed for up to an hour\n+                    # allow a session to be auto lock renewed for up to an hour - if it's processing a message\n                     renewer.register(receiver, receiver.session, max_lock_renewal_duration=3600)\n \n                     async for msg in receiver:\n@@ -75,23 +76,23 @@\n                         else:\n                             logging.error('Message processing failed!')\n \n-                        logger_adapter.info(f\"Message with id = {message['id']} processed as {result} and marked complete.\")\n+                        logger_adapter.info(f\"Message for resource_id={message['id']}, operation_id={message['operationId']} processed as {result} and marked complete.\")\n                         await receiver.complete_message(msg)\n \n+                    logger_adapter.info(\"Closing session\")\n+                    await renewer.close()\n+\n         except OperationTimeoutError:\n             # Timeout occurred whilst connecting to a session - this is expected and indicates no non-empty sessions are available\n-            logger_adapter.info(\"No sessions for this process. Sleeping 30s then will look again...\")\n+            logger_adapter.info(\"No sessions for this process. Will look again...\")\n \n         except ServiceBusConnectionError:\n             # Occasionally there will be a transient / network-level error in connecting to SB.\n-            logger_adapter.info(\"Unknown Service Bus connection error. Sleeping and will retry...\")\n+            logger_adapter.info(\"Unknown Service Bus connection error. Will retry...\")\n \n         except Exception:\n             # Catch all other exceptions, log them via .exception to get the stack trace, sleep, and reconnect\n-            logger_adapter.exception(\"Unknown exception. Sleeping and will retry...\")\n-\n-        finally:\n-            await asyncio.sleep(30)\n+            logger_adapter.exception(\"Unknown exception. Will retry...\")\n \n \n async def run_porter(command):\n", "issue": "Service Bus Sessions never terminate\nAfter a processor receives a message on a session, it hangs onto that session indefinitely, blocking the thread and meaning other messages cannot be processed. \r\n\r\nWe need to terminate the session after each message has been processed / errored out. \n", "before_files": [{"content": "import threading\nfrom multiprocessing import Process\nimport json\nimport socket\nimport asyncio\nimport logging\nimport sys\nfrom resources.commands import build_porter_command, build_porter_command_for_outputs\nfrom shared.config import get_config\nfrom resources.helpers import get_installation_id\nfrom resources.httpserver import start_server\n\nfrom shared.logging import disable_unwanted_loggers, initialize_logging, get_message_id_logger, shell_output_logger  # pylint: disable=import-error # noqa\nfrom resources import strings, statuses  # pylint: disable=import-error # noqa\nfrom contextlib import asynccontextmanager\nfrom azure.servicebus import ServiceBusMessage, NEXT_AVAILABLE_SESSION\nfrom azure.servicebus.exceptions import OperationTimeoutError, ServiceBusConnectionError\nfrom azure.servicebus.aio import ServiceBusClient, AutoLockRenewer\nfrom azure.identity.aio import DefaultAzureCredential\n\n\n# Initialise logging\nlogger_adapter = initialize_logging(logging.INFO, socket.gethostname())\ndisable_unwanted_loggers()\n\n# Initialise config\ntry:\n    config = get_config(logger_adapter)\nexcept KeyError as e:\n    logger_adapter.error(f\"Environment variable {e} is not set correctly...Exiting\")\n    sys.exit(1)\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(service_bus_client):\n    \"\"\"\n    This method is run per process. Each process will connect to service bus and try to establish a session.\n    If messages are there, the process will continue to receive all the messages associated with that session.\n    If no messages are there, the session connection will time out, sleep, and retry.\n    \"\"\"\n    q_name = config[\"resource_request_queue\"]\n\n    while True:\n        try:\n            logger_adapter.info(\"Looking for new session...\")\n            async with service_bus_client.get_queue_receiver(queue_name=q_name, session_id=NEXT_AVAILABLE_SESSION) as receiver:\n                logger_adapter.info(\"Got a session containing messages\")\n                async with AutoLockRenewer() as renewer:\n                    # allow a message to be auto lock renewed for up to an hour\n                    renewer.register(receiver, receiver.session, max_lock_renewal_duration=3600)\n\n                    async for msg in receiver:\n                        result = True\n                        message = \"\"\n\n                        try:\n                            message = json.loads(str(msg))\n                            logger_adapter.info(f\"Message received for resource_id={message['id']}, operation_id={message['operationId']}\")\n                            message_logger_adapter = get_message_id_logger(message['operationId'])  # correlate messages per operation\n                            result = await invoke_porter_action(message, service_bus_client, message_logger_adapter)\n                        except (json.JSONDecodeError) as e:\n                            logging.error(f\"Received bad service bus resource request message: {e}\")\n\n                        if result:\n                            logging.info(f\"Resource request for {message} is complete\")\n                        else:\n                            logging.error('Message processing failed!')\n\n                        logger_adapter.info(f\"Message with id = {message['id']} processed as {result} and marked complete.\")\n                        await receiver.complete_message(msg)\n\n        except OperationTimeoutError:\n            # Timeout occurred whilst connecting to a session - this is expected and indicates no non-empty sessions are available\n            logger_adapter.info(\"No sessions for this process. Sleeping 30s then will look again...\")\n\n        except ServiceBusConnectionError:\n            # Occasionally there will be a transient / network-level error in connecting to SB.\n            logger_adapter.info(\"Unknown Service Bus connection error. Sleeping and will retry...\")\n\n        except Exception:\n            # Catch all other exceptions, log them via .exception to get the stack trace, sleep, and reconnect\n            logger_adapter.exception(\"Unknown exception. Sleeping and will retry...\")\n\n        finally:\n            await asyncio.sleep(30)\n\n\nasync def run_porter(command):\n    \"\"\"\n    Run a Porter command\n    \"\"\"\n    proc = await asyncio.create_subprocess_shell(\n        ''.join(command),\n        stdout=asyncio.subprocess.PIPE,\n        stderr=asyncio.subprocess.PIPE,\n        env=config[\"porter_env\"])\n\n    stdout, stderr = await proc.communicate()\n    logging.info(f'run porter exited with {proc.returncode}')\n    result_stdout = None\n    result_stderr = None\n\n    if stdout:\n        result_stdout = stdout.decode()\n        shell_output_logger(result_stdout, '[stdout]', logger_adapter, logging.INFO)\n\n    if stderr:\n        result_stderr = stderr.decode()\n        shell_output_logger(result_stderr, '[stderr]', logger_adapter, logging.WARN)\n\n    return (proc.returncode, result_stdout, result_stderr)\n\n\ndef service_bus_message_generator(sb_message, status, deployment_message, outputs=None):\n    \"\"\"\n    Generate a resource request message\n    \"\"\"\n    installation_id = get_installation_id(sb_message)\n    message_dict = {\n        \"operationId\": sb_message[\"operationId\"],\n        \"id\": sb_message[\"id\"],\n        \"status\": status,\n        \"message\": f\"{installation_id}: {deployment_message}\"}\n\n    if outputs is not None:\n        message_dict[\"outputs\"] = outputs\n\n    resource_request_message = json.dumps(message_dict)\n    return resource_request_message\n\n\nasync def invoke_porter_action(msg_body, sb_client, message_logger_adapter) -> bool:\n    \"\"\"\n    Handle resource message by invoking specified porter action (i.e. install, uninstall)\n    \"\"\"\n    installation_id = get_installation_id(msg_body)\n    action = msg_body[\"action\"]\n    message_logger_adapter.info(f\"{installation_id}: {action} action starting...\")\n    sb_sender = sb_client.get_queue_sender(queue_name=config[\"deployment_status_queue\"])\n\n    # If the action is install/upgrade, post message on sb queue to start a deployment job\n    if action == \"install\" or action == \"upgrade\":\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\n    # Build and run porter command (flagging if its a built-in action or custom so we can adapt porter command appropriately)\n    is_custom_action = action not in [\"install\", \"upgrade\", \"uninstall\"]\n    porter_command = await build_porter_command(config, message_logger_adapter, msg_body, is_custom_action)\n    returncode, _, err = await run_porter(porter_command)\n\n    # Handle command output\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, statuses.failed_status_string_for[action], error_message)\n\n        # Post message on sb queue to notify receivers of action failure\n        await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n        message_logger_adapter.info(f\"{installation_id}: Porter action failed with error = {error_message}\")\n        return False\n\n    else:\n        # Get the outputs\n        # TODO: decide if this should \"fail\" the deployment\n        _, outputs = await get_porter_outputs(msg_body, message_logger_adapter)\n\n        success_message = f\"{action} action completed successfully.\"\n        resource_request_message = service_bus_message_generator(msg_body, statuses.pass_status_string_for[action], success_message, outputs)\n\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 get_porter_outputs(msg_body, message_logger_adapter):\n    \"\"\"\n    Get outputs JSON from a Porter command\n    \"\"\"\n    porter_command = await build_porter_command_for_outputs(msg_body)\n    returncode, stdout, err = await run_porter(porter_command)\n\n    if returncode != 0:\n        error_message = \"Error context message = \" + \" \".join(err.split('\\n'))\n        message_logger_adapter.info(f\"{get_installation_id(msg_body)}: Failed to get outputs with error = {error_message}\")\n        return False, \"\"\n    else:\n        outputs_json = {}\n        try:\n            outputs_json = json.loads(stdout)\n            message_logger_adapter.info(f\"Got outputs as json: {outputs_json}\")\n        except ValueError:\n            message_logger_adapter.error(f\"Got outputs invalid json: {stdout}\")\n\n        return True, outputs_json\n\n\nasync def runner():\n    async with default_credentials(config[\"vmss_msi_id\"]) as credential:\n        service_bus_client = ServiceBusClient(config[\"service_bus_namespace\"], credential)\n        await receive_message(service_bus_client)\n\n\ndef start_runner_process():\n    asyncio.ensure_future(runner())\n    event_loop = asyncio.get_event_loop()\n    event_loop.run_forever()\n    logger_adapter.info(\"Started resource processor\")\n\n\nif __name__ == \"__main__\":\n    httpserver_thread = threading.Thread(target=start_server)\n    httpserver_thread.start()\n    logger_adapter.info(\"Started http server\")\n\n    logger_adapter.info(f'Starting {str(config[\"number_processes_int\"])} processes...')\n    for i in range(config[\"number_processes_int\"]):\n        logger_adapter.info(f'Starting process {str(i)}')\n        process = Process(target=start_runner_process)\n        process.start()\n", "path": "resource_processor/vmss_porter/runner.py"}], "after_files": [{"content": "import threading\nfrom multiprocessing import Process\nimport json\nimport socket\nimport asyncio\nimport logging\nimport sys\nfrom resources.commands import build_porter_command, build_porter_command_for_outputs\nfrom shared.config import get_config\nfrom resources.helpers import get_installation_id\nfrom resources.httpserver import start_server\n\nfrom shared.logging import disable_unwanted_loggers, initialize_logging, get_message_id_logger, shell_output_logger  # pylint: disable=import-error # noqa\nfrom resources import strings, statuses  # pylint: disable=import-error # noqa\nfrom contextlib import asynccontextmanager\nfrom azure.servicebus import ServiceBusMessage, NEXT_AVAILABLE_SESSION\nfrom azure.servicebus.exceptions import OperationTimeoutError, ServiceBusConnectionError\nfrom azure.servicebus.aio import ServiceBusClient, AutoLockRenewer\nfrom azure.identity.aio import DefaultAzureCredential\n\n\n# Initialise logging\nlogger_adapter = initialize_logging(logging.INFO, socket.gethostname())\ndisable_unwanted_loggers()\n\n# Initialise config\ntry:\n    config = get_config(logger_adapter)\nexcept KeyError as e:\n    logger_adapter.error(f\"Environment variable {e} is not set correctly...Exiting\")\n    sys.exit(1)\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(service_bus_client):\n    \"\"\"\n    This method is run per process. Each process will connect to service bus and try to establish a session.\n    If messages are there, the process will continue to receive all the messages associated with that session.\n    If no messages are there, the session connection will time out, sleep, and retry.\n    \"\"\"\n    q_name = config[\"resource_request_queue\"]\n\n    while True:\n        try:\n            logger_adapter.info(\"Looking for new session...\")\n            # max_wait_time=1 -> don't hold the session open after processing of the message has finished\n            async with service_bus_client.get_queue_receiver(queue_name=q_name, max_wait_time=1, session_id=NEXT_AVAILABLE_SESSION) as receiver:\n                logger_adapter.info(\"Got a session containing messages\")\n                async with AutoLockRenewer() as renewer:\n                    # allow a session to be auto lock renewed for up to an hour - if it's processing a message\n                    renewer.register(receiver, receiver.session, max_lock_renewal_duration=3600)\n\n                    async for msg in receiver:\n                        result = True\n                        message = \"\"\n\n                        try:\n                            message = json.loads(str(msg))\n                            logger_adapter.info(f\"Message received for resource_id={message['id']}, operation_id={message['operationId']}\")\n                            message_logger_adapter = get_message_id_logger(message['operationId'])  # correlate messages per operation\n                            result = await invoke_porter_action(message, service_bus_client, message_logger_adapter)\n                        except (json.JSONDecodeError) as e:\n                            logging.error(f\"Received bad service bus resource request message: {e}\")\n\n                        if result:\n                            logging.info(f\"Resource request for {message} is complete\")\n                        else:\n                            logging.error('Message processing failed!')\n\n                        logger_adapter.info(f\"Message for resource_id={message['id']}, operation_id={message['operationId']} processed as {result} and marked complete.\")\n                        await receiver.complete_message(msg)\n\n                    logger_adapter.info(\"Closing session\")\n                    await renewer.close()\n\n        except OperationTimeoutError:\n            # Timeout occurred whilst connecting to a session - this is expected and indicates no non-empty sessions are available\n            logger_adapter.info(\"No sessions for this process. Will look again...\")\n\n        except ServiceBusConnectionError:\n            # Occasionally there will be a transient / network-level error in connecting to SB.\n            logger_adapter.info(\"Unknown Service Bus connection error. Will retry...\")\n\n        except Exception:\n            # Catch all other exceptions, log them via .exception to get the stack trace, sleep, and reconnect\n            logger_adapter.exception(\"Unknown exception. Will retry...\")\n\n\nasync def run_porter(command):\n    \"\"\"\n    Run a Porter command\n    \"\"\"\n    proc = await asyncio.create_subprocess_shell(\n        ''.join(command),\n        stdout=asyncio.subprocess.PIPE,\n        stderr=asyncio.subprocess.PIPE,\n        env=config[\"porter_env\"])\n\n    stdout, stderr = await proc.communicate()\n    logging.info(f'run porter exited with {proc.returncode}')\n    result_stdout = None\n    result_stderr = None\n\n    if stdout:\n        result_stdout = stdout.decode()\n        shell_output_logger(result_stdout, '[stdout]', logger_adapter, logging.INFO)\n\n    if stderr:\n        result_stderr = stderr.decode()\n        shell_output_logger(result_stderr, '[stderr]', logger_adapter, logging.WARN)\n\n    return (proc.returncode, result_stdout, result_stderr)\n\n\ndef service_bus_message_generator(sb_message, status, deployment_message, outputs=None):\n    \"\"\"\n    Generate a resource request message\n    \"\"\"\n    installation_id = get_installation_id(sb_message)\n    message_dict = {\n        \"operationId\": sb_message[\"operationId\"],\n        \"id\": sb_message[\"id\"],\n        \"status\": status,\n        \"message\": f\"{installation_id}: {deployment_message}\"}\n\n    if outputs is not None:\n        message_dict[\"outputs\"] = outputs\n\n    resource_request_message = json.dumps(message_dict)\n    return resource_request_message\n\n\nasync def invoke_porter_action(msg_body, sb_client, message_logger_adapter) -> bool:\n    \"\"\"\n    Handle resource message by invoking specified porter action (i.e. install, uninstall)\n    \"\"\"\n    installation_id = get_installation_id(msg_body)\n    action = msg_body[\"action\"]\n    message_logger_adapter.info(f\"{installation_id}: {action} action starting...\")\n    sb_sender = sb_client.get_queue_sender(queue_name=config[\"deployment_status_queue\"])\n\n    # If the action is install/upgrade, post message on sb queue to start a deployment job\n    if action == \"install\" or action == \"upgrade\":\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\n    # Build and run porter command (flagging if its a built-in action or custom so we can adapt porter command appropriately)\n    is_custom_action = action not in [\"install\", \"upgrade\", \"uninstall\"]\n    porter_command = await build_porter_command(config, message_logger_adapter, msg_body, is_custom_action)\n    returncode, _, err = await run_porter(porter_command)\n\n    # Handle command output\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, statuses.failed_status_string_for[action], error_message)\n\n        # Post message on sb queue to notify receivers of action failure\n        await sb_sender.send_messages(ServiceBusMessage(body=resource_request_message, correlation_id=msg_body[\"id\"]))\n        message_logger_adapter.info(f\"{installation_id}: Porter action failed with error = {error_message}\")\n        return False\n\n    else:\n        # Get the outputs\n        # TODO: decide if this should \"fail\" the deployment\n        _, outputs = await get_porter_outputs(msg_body, message_logger_adapter)\n\n        success_message = f\"{action} action completed successfully.\"\n        resource_request_message = service_bus_message_generator(msg_body, statuses.pass_status_string_for[action], success_message, outputs)\n\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 get_porter_outputs(msg_body, message_logger_adapter):\n    \"\"\"\n    Get outputs JSON from a Porter command\n    \"\"\"\n    porter_command = await build_porter_command_for_outputs(msg_body)\n    returncode, stdout, err = await run_porter(porter_command)\n\n    if returncode != 0:\n        error_message = \"Error context message = \" + \" \".join(err.split('\\n'))\n        message_logger_adapter.info(f\"{get_installation_id(msg_body)}: Failed to get outputs with error = {error_message}\")\n        return False, \"\"\n    else:\n        outputs_json = {}\n        try:\n            outputs_json = json.loads(stdout)\n            message_logger_adapter.info(f\"Got outputs as json: {outputs_json}\")\n        except ValueError:\n            message_logger_adapter.error(f\"Got outputs invalid json: {stdout}\")\n\n        return True, outputs_json\n\n\nasync def runner():\n    async with default_credentials(config[\"vmss_msi_id\"]) as credential:\n        service_bus_client = ServiceBusClient(config[\"service_bus_namespace\"], credential)\n        await receive_message(service_bus_client)\n\n\ndef start_runner_process():\n    asyncio.ensure_future(runner())\n    event_loop = asyncio.get_event_loop()\n    event_loop.run_forever()\n    logger_adapter.info(\"Started resource processor\")\n\n\nif __name__ == \"__main__\":\n    httpserver_thread = threading.Thread(target=start_server)\n    httpserver_thread.start()\n    logger_adapter.info(\"Started http server\")\n\n    logger_adapter.info(f'Starting {str(config[\"number_processes_int\"])} processes...')\n    for i in range(config[\"number_processes_int\"]):\n        logger_adapter.info(f'Starting process {str(i)}')\n        process = Process(target=start_runner_process)\n        process.start()\n", "path": "resource_processor/vmss_porter/runner.py"}]}