We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Attempt to log possible log writing errors
## Current behavior
Currently if the logger encounters an error it isn't immediately surfaced to Cloud users what the reason is for the logging error.



## Proposed behavior
Something along the lines of (but not limited to):
```python
try:
   write_some_logs()
except Exception as exc:
    sleep(1)
    write_this_error_log(exc)
except Exception as exc2:
    raise exc    # raise exception as it normally would
```

Where if the writing of the logs fails then the exception is passed as an attempted log. Otherwise if the exception logging fails then raise as it normally would. This would give some nice insight into "missing" logs.

## Example
Possible issues such as non serializeable bytes, large payloads, etc. could be surfaced

```
self.logger.info(b"bad bytes string")
CRITICAL - CloudHandler | Failed to write log with error: Object of type bytes is not JSON serializable
```

```
self.logger.info("LARGE STRING...........")
CRITICAL - CloudHandler | Failed to write log with error: 413 Client Error: Request Entity Too Large for url: https://api.prefect.io/graphql/alpha
```

--- 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 """
2 Utility functions for interacting with and configuring logging.  The main entrypoint for retrieving loggers for
3 customization is the `get_logger` utility.
4 
5 Note that Prefect Tasks come equipped with their own loggers.  These can be accessed via:
6     - `self.logger` if implementing a Task class
7     - `prefect.context.get("logger")` if using the `task` decorator
8 
9 When running locally, log levels and message formatting are set via your Prefect configuration file.
10 """
11 import atexit
12 import json
13 import logging
14 import sys
15 import threading
16 import time
17 from queue import Queue, Empty
18 from typing import Any
19 
20 import pendulum
21 
22 import prefect
23 from prefect.utilities.context import context
24 
25 
26 class CloudHandler(logging.StreamHandler):
27     def __init__(self) -> None:
28         super().__init__(sys.stdout)
29         self.client = None
30         self.logger = logging.getLogger("CloudHandler")
31         handler = logging.StreamHandler(sys.stdout)
32         formatter = logging.Formatter(context.config.logging.format)
33         formatter.converter = time.gmtime  # type: ignore
34         handler.setFormatter(formatter)
35         self.logger.addHandler(handler)
36         self.logger.setLevel(context.config.logging.level)
37 
38     @property
39     def queue(self) -> Queue:
40         if not hasattr(self, "_queue"):
41             self._queue = Queue()  # type: Queue
42             self._flush = False
43             self.start()
44         return self._queue
45 
46     def flush(self) -> None:
47         self._flush = True
48         if self.client is not None:
49             self.batch_upload()
50             self._thread.join()
51 
52     def batch_upload(self) -> None:
53         logs = []
54         try:
55             while True:
56                 log = self.queue.get(False)
57                 logs.append(log)
58         except Empty:
59             pass
60 
61         if logs:
62             try:
63                 assert self.client is not None
64                 self.client.write_run_logs(logs)
65             except Exception as exc:
66                 self.logger.critical(
67                     "Failed to write log with error: {}".format(str(exc))
68                 )
69 
70     def _monitor(self) -> None:
71         while not self._flush:
72             self.batch_upload()
73             time.sleep(self.heartbeat)
74 
75     def __del__(self) -> None:
76         if hasattr(self, "_thread"):
77             self.flush()
78             atexit.unregister(self.flush)
79 
80     def start(self) -> None:
81         if not hasattr(self, "_thread"):
82             self.heartbeat = context.config.cloud.logging_heartbeat
83             self._thread = t = threading.Thread(
84                 target=self._monitor, name="PrefectCloudLoggingThread"
85             )
86             t.daemon = True
87             t.start()
88             atexit.register(self.flush)
89 
90     def put(self, log: dict) -> None:
91         try:
92             json.dumps(log)  # make sure the payload is serializable
93             self.queue.put(log)
94         except TypeError as exc:
95             self.logger.critical("Failed to write log with error: {}".format(str(exc)))
96 
97     def emit(self, record) -> None:  # type: ignore
98         # if we shouldn't log to cloud, don't emit
99         if not prefect.context.config.logging.log_to_cloud:
100             return
101 
102         try:
103             from prefect.client import Client
104 
105             if self.client is None:
106                 self.client = Client()  # type: ignore
107 
108             assert isinstance(self.client, Client)  # mypy assert
109 
110             record_dict = record.__dict__.copy()
111             log = dict()
112             log["flowRunId"] = prefect.context.get("flow_run_id", None)
113             log["taskRunId"] = prefect.context.get("task_run_id", None)
114             log["timestamp"] = pendulum.from_timestamp(
115                 record_dict.pop("created", time.time())
116             ).isoformat()
117             log["name"] = record_dict.pop("name", None)
118             log["message"] = record_dict.pop("message", None)
119             log["level"] = record_dict.pop("levelname", None)
120 
121             if record_dict.get("exc_text") is not None:
122                 log["message"] += "\n" + record_dict.pop("exc_text", "")
123                 record_dict.pop("exc_info", None)
124 
125             log["info"] = record_dict
126             self.put(log)
127         except Exception as exc:
128             self.logger.critical("Failed to write log with error: {}".format(str(exc)))
129 
130 
131 def configure_logging(testing: bool = False) -> logging.Logger:
132     """
133     Creates a "prefect" root logger with a `StreamHandler` that has level and formatting
134     set from `prefect.config`.
135 
136     Args:
137         - testing (bool, optional): a boolean specifying whether this configuration
138             is for testing purposes only; this helps us isolate any global state during testing
139             by configuring a "prefect-test-logger" instead of the standard "prefect" logger
140 
141     Returns:
142         - logging.Logger: a configured logging object
143     """
144     name = "prefect-test-logger" if testing else "prefect"
145     logger = logging.getLogger(name)
146     handler = logging.StreamHandler(sys.stdout)
147     formatter = logging.Formatter(context.config.logging.format)
148     formatter.converter = time.gmtime  # type: ignore
149     handler.setFormatter(formatter)
150     logger.addHandler(handler)
151     logger.setLevel(context.config.logging.level)
152 
153     cloud_handler = CloudHandler()
154     cloud_handler.setLevel("DEBUG")
155     logger.addHandler(cloud_handler)
156     return logger
157 
158 
159 prefect_logger = configure_logging()
160 
161 
162 def get_logger(name: str = None) -> logging.Logger:
163     """
164     Returns a "prefect" logger.
165 
166     Args:
167         - name (str): if `None`, the root Prefect logger is returned. If provided, a child
168             logger of the name `"prefect.{name}"` is returned. The child logger inherits
169             the root logger's settings.
170 
171     Returns:
172         - logging.Logger: a configured logging object with the appropriate name
173     """
174     if name is None:
175         return prefect_logger
176     else:
177         return prefect_logger.getChild(name)
178
```
--- 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
@@ -63,9 +63,15 @@
                 assert self.client is not None
                 self.client.write_run_logs(logs)
             except Exception as exc:
-                self.logger.critical(
-                    "Failed to write log with error: {}".format(str(exc))
-                )
+                message = "Failed to write log with error: {}".format(str(exc))
+                self.logger.critical(message)
+
+                # Attempt to write batch error log otherwise log invalid cloud communication
+                try:
+                    assert self.client is not None
+                    self.client.write_run_logs([self._make_error_log(message)])
+                except Exception as exc:
+                    self.logger.critical("Unable to write logs to Prefect Cloud")
 
     def _monitor(self) -> None:
         while not self._flush:
@@ -92,7 +98,10 @@
             json.dumps(log)  # make sure the payload is serializable
             self.queue.put(log)
         except TypeError as exc:
-            self.logger.critical("Failed to write log with error: {}".format(str(exc)))
+            message = "Failed to write log with error: {}".format(str(exc))
+            self.logger.critical(message)
+
+            self.queue.put(self._make_error_log(message))
 
     def emit(self, record) -> None:  # type: ignore
         # if we shouldn't log to cloud, don't emit
@@ -125,7 +134,21 @@
             log["info"] = record_dict
             self.put(log)
         except Exception as exc:
-            self.logger.critical("Failed to write log with error: {}".format(str(exc)))
+            message = "Failed to write log with error: {}".format(str(exc))
+            self.logger.critical(message)
+
+            self.put(self._make_error_log(message))
+
+    def _make_error_log(self, message: str) -> dict:
+        log = dict()
+        log["flowRunId"] = prefect.context.get("flow_run_id", None)
+        log["timestamp"] = pendulum.from_timestamp(time.time()).isoformat()
+        log["name"] = self.logger.name
+        log["message"] = message
+        log["level"] = "CRITICAL"
+        log["info"] = {}
+
+        return log
 
 
 def configure_logging(testing: bool = False) -> logging.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@@ -63,9 +63,15 @@\n                 assert self.client is not None\n                 self.client.write_run_logs(logs)\n             except Exception as exc:\n-                self.logger.critical(\n-                    \"Failed to write log with error: {}\".format(str(exc))\n-                )\n+                message = \"Failed to write log with error: {}\".format(str(exc))\n+                self.logger.critical(message)\n+\n+                # Attempt to write batch error log otherwise log invalid cloud communication\n+                try:\n+                    assert self.client is not None\n+                    self.client.write_run_logs([self._make_error_log(message)])\n+                except Exception as exc:\n+                    self.logger.critical(\"Unable to write logs to Prefect Cloud\")\n \n     def _monitor(self) -> None:\n         while not self._flush:\n@@ -92,7 +98,10 @@\n             json.dumps(log)  # make sure the payload is serializable\n             self.queue.put(log)\n         except TypeError as exc:\n-            self.logger.critical(\"Failed to write log with error: {}\".format(str(exc)))\n+            message = \"Failed to write log with error: {}\".format(str(exc))\n+            self.logger.critical(message)\n+\n+            self.queue.put(self._make_error_log(message))\n \n     def emit(self, record) -> None:  # type: ignore\n         # if we shouldn't log to cloud, don't emit\n@@ -125,7 +134,21 @@\n             log[\"info\"] = record_dict\n             self.put(log)\n         except Exception as exc:\n-            self.logger.critical(\"Failed to write log with error: {}\".format(str(exc)))\n+            message = \"Failed to write log with error: {}\".format(str(exc))\n+            self.logger.critical(message)\n+\n+            self.put(self._make_error_log(message))\n+\n+    def _make_error_log(self, message: str) -> dict:\n+        log = dict()\n+        log[\"flowRunId\"] = prefect.context.get(\"flow_run_id\", None)\n+        log[\"timestamp\"] = pendulum.from_timestamp(time.time()).isoformat()\n+        log[\"name\"] = self.logger.name\n+        log[\"message\"] = message\n+        log[\"level\"] = \"CRITICAL\"\n+        log[\"info\"] = {}\n+\n+        return log\n \n \n def configure_logging(testing: bool = False) -> logging.Logger:\n", "issue": "Attempt to log possible log writing errors\n## Current behavior\r\nCurrently if the logger encounters an error it isn't immediately surfaced to Cloud users what the reason is for the logging error.\r\n\r\n\r\n\r\n## Proposed behavior\r\nSomething along the lines of (but not limited to):\r\n```python\r\ntry:\r\n   write_some_logs()\r\nexcept Exception as exc:\r\n    sleep(1)\r\n    write_this_error_log(exc)\r\nexcept Exception as exc2:\r\n    raise exc    # raise exception as it normally would\r\n```\r\n\r\nWhere if the writing of the logs fails then the exception is passed as an attempted log. Otherwise if the exception logging fails then raise as it normally would. This would give some nice insight into \"missing\" logs.\r\n\r\n## Example\r\nPossible issues such as non serializeable bytes, large payloads, etc. could be surfaced\r\n\r\n```\r\nself.logger.info(b\"bad bytes string\")\r\nCRITICAL - CloudHandler | Failed to write log with error: Object of type bytes is not JSON serializable\r\n```\r\n\r\n```\r\nself.logger.info(\"LARGE STRING...........\")\r\nCRITICAL - CloudHandler | Failed to write log with error: 413 Client Error: Request Entity Too Large for url: https://api.prefect.io/graphql/alpha\r\n```\n", "before_files": [{"content": "\"\"\"\nUtility functions for interacting with and configuring logging.  The main entrypoint for retrieving loggers for\ncustomization is the `get_logger` utility.\n\nNote that Prefect Tasks come equipped with their own loggers.  These can be accessed via:\n    - `self.logger` if implementing a Task class\n    - `prefect.context.get(\"logger\")` if using the `task` decorator\n\nWhen running locally, log levels and message formatting are set via your Prefect configuration file.\n\"\"\"\nimport atexit\nimport json\nimport logging\nimport sys\nimport threading\nimport time\nfrom queue import Queue, Empty\nfrom typing import Any\n\nimport pendulum\n\nimport prefect\nfrom prefect.utilities.context import context\n\n\nclass CloudHandler(logging.StreamHandler):\n    def __init__(self) -> None:\n        super().__init__(sys.stdout)\n        self.client = None\n        self.logger = logging.getLogger(\"CloudHandler\")\n        handler = logging.StreamHandler(sys.stdout)\n        formatter = logging.Formatter(context.config.logging.format)\n        formatter.converter = time.gmtime  # type: ignore\n        handler.setFormatter(formatter)\n        self.logger.addHandler(handler)\n        self.logger.setLevel(context.config.logging.level)\n\n    @property\n    def queue(self) -> Queue:\n        if not hasattr(self, \"_queue\"):\n            self._queue = Queue()  # type: Queue\n            self._flush = False\n            self.start()\n        return self._queue\n\n    def flush(self) -> None:\n        self._flush = True\n        if self.client is not None:\n            self.batch_upload()\n            self._thread.join()\n\n    def batch_upload(self) -> None:\n        logs = []\n        try:\n            while True:\n                log = self.queue.get(False)\n                logs.append(log)\n        except Empty:\n            pass\n\n        if logs:\n            try:\n                assert self.client is not None\n                self.client.write_run_logs(logs)\n            except Exception as exc:\n                self.logger.critical(\n                    \"Failed to write log with error: {}\".format(str(exc))\n                )\n\n    def _monitor(self) -> None:\n        while not self._flush:\n            self.batch_upload()\n            time.sleep(self.heartbeat)\n\n    def __del__(self) -> None:\n        if hasattr(self, \"_thread\"):\n            self.flush()\n            atexit.unregister(self.flush)\n\n    def start(self) -> None:\n        if not hasattr(self, \"_thread\"):\n            self.heartbeat = context.config.cloud.logging_heartbeat\n            self._thread = t = threading.Thread(\n                target=self._monitor, name=\"PrefectCloudLoggingThread\"\n            )\n            t.daemon = True\n            t.start()\n            atexit.register(self.flush)\n\n    def put(self, log: dict) -> None:\n        try:\n            json.dumps(log)  # make sure the payload is serializable\n            self.queue.put(log)\n        except TypeError as exc:\n            self.logger.critical(\"Failed to write log with error: {}\".format(str(exc)))\n\n    def emit(self, record) -> None:  # type: ignore\n        # if we shouldn't log to cloud, don't emit\n        if not prefect.context.config.logging.log_to_cloud:\n            return\n\n        try:\n            from prefect.client import Client\n\n            if self.client is None:\n                self.client = Client()  # type: ignore\n\n            assert isinstance(self.client, Client)  # mypy assert\n\n            record_dict = record.__dict__.copy()\n            log = dict()\n            log[\"flowRunId\"] = prefect.context.get(\"flow_run_id\", None)\n            log[\"taskRunId\"] = prefect.context.get(\"task_run_id\", None)\n            log[\"timestamp\"] = pendulum.from_timestamp(\n                record_dict.pop(\"created\", time.time())\n            ).isoformat()\n            log[\"name\"] = record_dict.pop(\"name\", None)\n            log[\"message\"] = record_dict.pop(\"message\", None)\n            log[\"level\"] = record_dict.pop(\"levelname\", None)\n\n            if record_dict.get(\"exc_text\") is not None:\n                log[\"message\"] += \"\\n\" + record_dict.pop(\"exc_text\", \"\")\n                record_dict.pop(\"exc_info\", None)\n\n            log[\"info\"] = record_dict\n            self.put(log)\n        except Exception as exc:\n            self.logger.critical(\"Failed to write log with error: {}\".format(str(exc)))\n\n\ndef configure_logging(testing: bool = False) -> logging.Logger:\n    \"\"\"\n    Creates a \"prefect\" root logger with a `StreamHandler` that has level and formatting\n    set from `prefect.config`.\n\n    Args:\n        - testing (bool, optional): a boolean specifying whether this configuration\n            is for testing purposes only; this helps us isolate any global state during testing\n            by configuring a \"prefect-test-logger\" instead of the standard \"prefect\" logger\n\n    Returns:\n        - logging.Logger: a configured logging object\n    \"\"\"\n    name = \"prefect-test-logger\" if testing else \"prefect\"\n    logger = logging.getLogger(name)\n    handler = logging.StreamHandler(sys.stdout)\n    formatter = logging.Formatter(context.config.logging.format)\n    formatter.converter = time.gmtime  # type: ignore\n    handler.setFormatter(formatter)\n    logger.addHandler(handler)\n    logger.setLevel(context.config.logging.level)\n\n    cloud_handler = CloudHandler()\n    cloud_handler.setLevel(\"DEBUG\")\n    logger.addHandler(cloud_handler)\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: a configured logging object with the appropriate name\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": "\"\"\"\nUtility functions for interacting with and configuring logging.  The main entrypoint for retrieving loggers for\ncustomization is the `get_logger` utility.\n\nNote that Prefect Tasks come equipped with their own loggers.  These can be accessed via:\n    - `self.logger` if implementing a Task class\n    - `prefect.context.get(\"logger\")` if using the `task` decorator\n\nWhen running locally, log levels and message formatting are set via your Prefect configuration file.\n\"\"\"\nimport atexit\nimport json\nimport logging\nimport sys\nimport threading\nimport time\nfrom queue import Queue, Empty\nfrom typing import Any\n\nimport pendulum\n\nimport prefect\nfrom prefect.utilities.context import context\n\n\nclass CloudHandler(logging.StreamHandler):\n    def __init__(self) -> None:\n        super().__init__(sys.stdout)\n        self.client = None\n        self.logger = logging.getLogger(\"CloudHandler\")\n        handler = logging.StreamHandler(sys.stdout)\n        formatter = logging.Formatter(context.config.logging.format)\n        formatter.converter = time.gmtime  # type: ignore\n        handler.setFormatter(formatter)\n        self.logger.addHandler(handler)\n        self.logger.setLevel(context.config.logging.level)\n\n    @property\n    def queue(self) -> Queue:\n        if not hasattr(self, \"_queue\"):\n            self._queue = Queue()  # type: Queue\n            self._flush = False\n            self.start()\n        return self._queue\n\n    def flush(self) -> None:\n        self._flush = True\n        if self.client is not None:\n            self.batch_upload()\n            self._thread.join()\n\n    def batch_upload(self) -> None:\n        logs = []\n        try:\n            while True:\n                log = self.queue.get(False)\n                logs.append(log)\n        except Empty:\n            pass\n\n        if logs:\n            try:\n                assert self.client is not None\n                self.client.write_run_logs(logs)\n            except Exception as exc:\n                message = \"Failed to write log with error: {}\".format(str(exc))\n                self.logger.critical(message)\n\n                # Attempt to write batch error log otherwise log invalid cloud communication\n                try:\n                    assert self.client is not None\n                    self.client.write_run_logs([self._make_error_log(message)])\n                except Exception as exc:\n                    self.logger.critical(\"Unable to write logs to Prefect Cloud\")\n\n    def _monitor(self) -> None:\n        while not self._flush:\n            self.batch_upload()\n            time.sleep(self.heartbeat)\n\n    def __del__(self) -> None:\n        if hasattr(self, \"_thread\"):\n            self.flush()\n            atexit.unregister(self.flush)\n\n    def start(self) -> None:\n        if not hasattr(self, \"_thread\"):\n            self.heartbeat = context.config.cloud.logging_heartbeat\n            self._thread = t = threading.Thread(\n                target=self._monitor, name=\"PrefectCloudLoggingThread\"\n            )\n            t.daemon = True\n            t.start()\n            atexit.register(self.flush)\n\n    def put(self, log: dict) -> None:\n        try:\n            json.dumps(log)  # make sure the payload is serializable\n            self.queue.put(log)\n        except TypeError as exc:\n            message = \"Failed to write log with error: {}\".format(str(exc))\n            self.logger.critical(message)\n\n            self.queue.put(self._make_error_log(message))\n\n    def emit(self, record) -> None:  # type: ignore\n        # if we shouldn't log to cloud, don't emit\n        if not prefect.context.config.logging.log_to_cloud:\n            return\n\n        try:\n            from prefect.client import Client\n\n            if self.client is None:\n                self.client = Client()  # type: ignore\n\n            assert isinstance(self.client, Client)  # mypy assert\n\n            record_dict = record.__dict__.copy()\n            log = dict()\n            log[\"flowRunId\"] = prefect.context.get(\"flow_run_id\", None)\n            log[\"taskRunId\"] = prefect.context.get(\"task_run_id\", None)\n            log[\"timestamp\"] = pendulum.from_timestamp(\n                record_dict.pop(\"created\", time.time())\n            ).isoformat()\n            log[\"name\"] = record_dict.pop(\"name\", None)\n            log[\"message\"] = record_dict.pop(\"message\", None)\n            log[\"level\"] = record_dict.pop(\"levelname\", None)\n\n            if record_dict.get(\"exc_text\") is not None:\n                log[\"message\"] += \"\\n\" + record_dict.pop(\"exc_text\", \"\")\n                record_dict.pop(\"exc_info\", None)\n\n            log[\"info\"] = record_dict\n            self.put(log)\n        except Exception as exc:\n            message = \"Failed to write log with error: {}\".format(str(exc))\n            self.logger.critical(message)\n\n            self.put(self._make_error_log(message))\n\n    def _make_error_log(self, message: str) -> dict:\n        log = dict()\n        log[\"flowRunId\"] = prefect.context.get(\"flow_run_id\", None)\n        log[\"timestamp\"] = pendulum.from_timestamp(time.time()).isoformat()\n        log[\"name\"] = self.logger.name\n        log[\"message\"] = message\n        log[\"level\"] = \"CRITICAL\"\n        log[\"info\"] = {}\n\n        return log\n\n\ndef configure_logging(testing: bool = False) -> logging.Logger:\n    \"\"\"\n    Creates a \"prefect\" root logger with a `StreamHandler` that has level and formatting\n    set from `prefect.config`.\n\n    Args:\n        - testing (bool, optional): a boolean specifying whether this configuration\n            is for testing purposes only; this helps us isolate any global state during testing\n            by configuring a \"prefect-test-logger\" instead of the standard \"prefect\" logger\n\n    Returns:\n        - logging.Logger: a configured logging object\n    \"\"\"\n    name = \"prefect-test-logger\" if testing else \"prefect\"\n    logger = logging.getLogger(name)\n    handler = logging.StreamHandler(sys.stdout)\n    formatter = logging.Formatter(context.config.logging.format)\n    formatter.converter = time.gmtime  # type: ignore\n    handler.setFormatter(formatter)\n    logger.addHandler(handler)\n    logger.setLevel(context.config.logging.level)\n\n    cloud_handler = CloudHandler()\n    cloud_handler.setLevel(\"DEBUG\")\n    logger.addHandler(cloud_handler)\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: a configured logging object with the appropriate name\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"}]}