We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
All logs aren't making it through
It seems that the QueueHandler for our logs doesn't block correctly, so some logs don't make it to our logger service.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/utilities/logging.py`
Content:
```
1 # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula
2 import logging
3 import os
4 import queue
5 from logging.handlers import QueueHandler, QueueListener
6 
7 import prefect
8 from prefect.configuration import config
9 
10 
11 class RemoteHandler(logging.StreamHandler):
12     def __init__(self) -> None:
13         super().__init__()
14         self.logger_server = config.cloud.log
15         self.client = None
16 
17     def emit(self, record):
18         if self.client is None:
19             from prefect.client import Client
20 
21             self.client = Client()
22         r = self.client.post(path="", server=self.logger_server, **record.__dict__)
23 
24 
25 old_factory = logging.getLogRecordFactory()
26 
27 
28 def cloud_record_factory(*args, **kwargs):
29     record = old_factory(*args, **kwargs)
30     record.flowrunid = prefect.context.get("flow_run_id", "")
31     record.taskrunid = prefect.context.get("task_run_id", "")
32     return record
33 
34 
35 def configure_logging() -> logging.Logger:
36     """
37     Creates a "prefect" root logger with a `StreamHandler` that has level and formatting
38     set from `prefect.config`.
39 
40     Returns:
41         logging.Logger
42     """
43     logger = logging.getLogger("prefect")
44     handler = logging.StreamHandler()
45     formatter = logging.Formatter(config.logging.format)
46     handler.setFormatter(formatter)
47     logger.addHandler(handler)
48     logger.setLevel(config.logging.level)
49 
50     # send logs to server
51     if config.logging.log_to_cloud:
52         logging.setLogRecordFactory(cloud_record_factory)
53         log_queue = queue.Queue(-1)  # unlimited size queue
54         queue_handler = QueueHandler(log_queue)
55         remote_handler = RemoteHandler()
56         remote_listener = QueueListener(log_queue, remote_handler)
57         logger.addHandler(queue_handler)
58         remote_listener.start()
59 
60     return logger
61 
62 
63 prefect_logger = configure_logging()
64 
65 
66 def get_logger(name: str = None) -> logging.Logger:
67     """
68     Returns a "prefect" logger.
69 
70     Args:
71         - name (str): if `None`, the root Prefect logger is returned. If provided, a child
72             logger of the name `"prefect.{name}"` is returned. The child logger inherits
73             the root logger's settings.
74 
75     Returns:
76         logging.Logger
77     """
78     if name is None:
79         return prefect_logger
80     else:
81         return prefect_logger.getChild(name)
82
```
--- 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/src/prefect/utilities/logging.py b/src/prefect/utilities/logging.py
--- a/src/prefect/utilities/logging.py
+++ b/src/prefect/utilities/logging.py
@@ -1,4 +1,5 @@
 # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula
+import atexit
 import logging
 import os
 import queue
@@ -56,6 +57,8 @@
         remote_listener = QueueListener(log_queue, remote_handler)
         logger.addHandler(queue_handler)
         remote_listener.start()
+        stopper = lambda listener: listener.stop()
+        atexit.register(stopper, remote_listener)
 
     return logger

{"golden_diff": "diff --git a/src/prefect/utilities/logging.py b/src/prefect/utilities/logging.py\n--- a/src/prefect/utilities/logging.py\n+++ b/src/prefect/utilities/logging.py\n@@ -1,4 +1,5 @@\n # Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\n+import atexit\n import logging\n import os\n import queue\n@@ -56,6 +57,8 @@\n         remote_listener = QueueListener(log_queue, remote_handler)\n         logger.addHandler(queue_handler)\n         remote_listener.start()\n+        stopper = lambda listener: listener.stop()\n+        atexit.register(stopper, remote_listener)\n \n     return logger\n", "issue": "All logs aren't making it through\nIt seems that the QueueHandler for our logs doesn't block correctly, so some logs don't make it to our logger service.\n", "before_files": [{"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\nimport logging\nimport os\nimport queue\nfrom logging.handlers import QueueHandler, QueueListener\n\nimport prefect\nfrom prefect.configuration import config\n\n\nclass RemoteHandler(logging.StreamHandler):\n    def __init__(self) -> None:\n        super().__init__()\n        self.logger_server = config.cloud.log\n        self.client = None\n\n    def emit(self, record):\n        if self.client is None:\n            from prefect.client import Client\n\n            self.client = Client()\n        r = self.client.post(path=\"\", server=self.logger_server, **record.__dict__)\n\n\nold_factory = logging.getLogRecordFactory()\n\n\ndef cloud_record_factory(*args, **kwargs):\n    record = old_factory(*args, **kwargs)\n    record.flowrunid = prefect.context.get(\"flow_run_id\", \"\")\n    record.taskrunid = prefect.context.get(\"task_run_id\", \"\")\n    return record\n\n\ndef configure_logging() -> logging.Logger:\n    \"\"\"\n    Creates a \"prefect\" root logger with a `StreamHandler` that has level and formatting\n    set from `prefect.config`.\n\n    Returns:\n        logging.Logger\n    \"\"\"\n    logger = logging.getLogger(\"prefect\")\n    handler = logging.StreamHandler()\n    formatter = logging.Formatter(config.logging.format)\n    handler.setFormatter(formatter)\n    logger.addHandler(handler)\n    logger.setLevel(config.logging.level)\n\n    # send logs to server\n    if config.logging.log_to_cloud:\n        logging.setLogRecordFactory(cloud_record_factory)\n        log_queue = queue.Queue(-1)  # unlimited size queue\n        queue_handler = QueueHandler(log_queue)\n        remote_handler = RemoteHandler()\n        remote_listener = QueueListener(log_queue, remote_handler)\n        logger.addHandler(queue_handler)\n        remote_listener.start()\n\n    return logger\n\n\nprefect_logger = configure_logging()\n\n\ndef get_logger(name: str = None) -> logging.Logger:\n    \"\"\"\n    Returns a \"prefect\" logger.\n\n    Args:\n        - name (str): if `None`, the root Prefect logger is returned. If provided, a child\n            logger of the name `\"prefect.{name}\"` is returned. The child logger inherits\n            the root logger's settings.\n\n    Returns:\n        logging.Logger\n    \"\"\"\n    if name is None:\n        return prefect_logger\n    else:\n        return prefect_logger.getChild(name)\n", "path": "src/prefect/utilities/logging.py"}], "after_files": [{"content": "# Licensed under LICENSE.md; also available at https://www.prefect.io/licenses/alpha-eula\nimport atexit\nimport logging\nimport os\nimport queue\nfrom logging.handlers import QueueHandler, QueueListener\n\nimport prefect\nfrom prefect.configuration import config\n\n\nclass RemoteHandler(logging.StreamHandler):\n    def __init__(self) -> None:\n        super().__init__()\n        self.logger_server = config.cloud.log\n        self.client = None\n\n    def emit(self, record):\n        if self.client is None:\n            from prefect.client import Client\n\n            self.client = Client()\n        r = self.client.post(path=\"\", server=self.logger_server, **record.__dict__)\n\n\nold_factory = logging.getLogRecordFactory()\n\n\ndef cloud_record_factory(*args, **kwargs):\n    record = old_factory(*args, **kwargs)\n    record.flowrunid = prefect.context.get(\"flow_run_id\", \"\")\n    record.taskrunid = prefect.context.get(\"task_run_id\", \"\")\n    return record\n\n\ndef configure_logging() -> logging.Logger:\n    \"\"\"\n    Creates a \"prefect\" root logger with a `StreamHandler` that has level and formatting\n    set from `prefect.config`.\n\n    Returns:\n        logging.Logger\n    \"\"\"\n    logger = logging.getLogger(\"prefect\")\n    handler = logging.StreamHandler()\n    formatter = logging.Formatter(config.logging.format)\n    handler.setFormatter(formatter)\n    logger.addHandler(handler)\n    logger.setLevel(config.logging.level)\n\n    # send logs to server\n    if config.logging.log_to_cloud:\n        logging.setLogRecordFactory(cloud_record_factory)\n        log_queue = queue.Queue(-1)  # unlimited size queue\n        queue_handler = QueueHandler(log_queue)\n        remote_handler = RemoteHandler()\n        remote_listener = QueueListener(log_queue, remote_handler)\n        logger.addHandler(queue_handler)\n        remote_listener.start()\n        stopper = lambda listener: listener.stop()\n        atexit.register(stopper, remote_listener)\n\n    return logger\n\n\nprefect_logger = configure_logging()\n\n\ndef get_logger(name: str = None) -> logging.Logger:\n    \"\"\"\n    Returns a \"prefect\" logger.\n\n    Args:\n        - name (str): if `None`, the root Prefect logger is returned. If provided, a child\n            logger of the name `\"prefect.{name}\"` is returned. The child logger inherits\n            the root logger's settings.\n\n    Returns:\n        logging.Logger\n    \"\"\"\n    if name is None:\n        return prefect_logger\n    else:\n        return prefect_logger.getChild(name)\n", "path": "src/prefect/utilities/logging.py"}]}