We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add Flow and Task identifying information to LogRecords
## Current behavior
- All log entries have the default [LogRecord attributes](https://docs.python.org/3/library/logging.html#logrecord-objects)... 
- There is no universally identifiable flow slug

## Proposed behavior
...It would be useful to attach Core-specific information to log entries:
- `task_name` & `task_slug`: available only on entries emitted from Tasks and TaskRunners
- `flow_name` & `flow_slug`: available only on entries emitted from Flows and FlowRunners

The addition of the `flow_slug` makes is possible to uniquely identify flows other than by name.

## Example
One can now take a single stream of log entries during a local flow run and separate it by entity without parsing the `LogRecord.msg` field (which is unreliable). This makes is possible to build richer feedback interfaces for filtering, organizing, and otherwise selectively manipulating log entries.

Here is an example CLI interface that would be made possible by theses features:

![local-progress](https://user-images.githubusercontent.com/590471/72913913-dbd00880-3d0b-11ea-8c85-a0e74e1c9551.gif)

(Given the following flow):
```python
from utils import seed, random_data
import prefect

@prefect.task
def extract():
    return seed.get()

@prefect.task
def transform(seed):
    logger = prefect.context.get("logger")
    data = []
    for x in range(10):
        c_data = random_data(seed)
        logger.info("debug info: {}".format(c_data))
        data.append(c_data)
    return data

@prefect.task
def load(data):
    # do something with the data....
    pass

with prefect.Flow("ETL") as flow:
    e = extract()
    t = transform(e)
    load(t)
```


--- 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                 message = "Failed to write log with error: {}".format(str(exc))
67                 self.logger.critical(message)
68 
69                 # Attempt to write batch error log otherwise log invalid cloud communication
70                 try:
71                     assert self.client is not None
72                     self.client.write_run_logs([self._make_error_log(message)])
73                 except Exception as exc:
74                     self.logger.critical("Unable to write logs to Prefect Cloud")
75 
76     def _monitor(self) -> None:
77         while not self._flush:
78             self.batch_upload()
79             time.sleep(self.heartbeat)
80 
81     def __del__(self) -> None:
82         if hasattr(self, "_thread"):
83             self.flush()
84             atexit.unregister(self.flush)
85 
86     def start(self) -> None:
87         if not hasattr(self, "_thread"):
88             self.heartbeat = context.config.cloud.logging_heartbeat
89             self._thread = t = threading.Thread(
90                 target=self._monitor, name="PrefectCloudLoggingThread"
91             )
92             t.daemon = True
93             t.start()
94             atexit.register(self.flush)
95 
96     def put(self, log: dict) -> None:
97         try:
98             json.dumps(log)  # make sure the payload is serializable
99             self.queue.put(log)
100         except TypeError as exc:
101             message = "Failed to write log with error: {}".format(str(exc))
102             self.logger.critical(message)
103 
104             self.queue.put(self._make_error_log(message))
105 
106     def emit(self, record) -> None:  # type: ignore
107         # if we shouldn't log to cloud, don't emit
108         if not prefect.context.config.logging.log_to_cloud:
109             return
110 
111         try:
112             from prefect.client import Client
113 
114             if self.client is None:
115                 self.client = Client()  # type: ignore
116 
117             assert isinstance(self.client, Client)  # mypy assert
118 
119             record_dict = record.__dict__.copy()
120             log = dict()
121             log["flowRunId"] = prefect.context.get("flow_run_id", None)
122             log["taskRunId"] = prefect.context.get("task_run_id", None)
123             log["timestamp"] = pendulum.from_timestamp(
124                 record_dict.pop("created", time.time())
125             ).isoformat()
126             log["name"] = record_dict.pop("name", None)
127             log["message"] = record_dict.pop("message", None)
128             log["level"] = record_dict.pop("levelname", None)
129 
130             if record_dict.get("exc_text") is not None:
131                 log["message"] += "\n" + record_dict.pop("exc_text", "")
132                 record_dict.pop("exc_info", None)
133 
134             log["info"] = record_dict
135             self.put(log)
136         except Exception as exc:
137             message = "Failed to write log with error: {}".format(str(exc))
138             self.logger.critical(message)
139 
140             self.put(self._make_error_log(message))
141 
142     def _make_error_log(self, message: str) -> dict:
143         log = dict()
144         log["flowRunId"] = prefect.context.get("flow_run_id", None)
145         log["timestamp"] = pendulum.from_timestamp(time.time()).isoformat()
146         log["name"] = self.logger.name
147         log["message"] = message
148         log["level"] = "CRITICAL"
149         log["info"] = {}
150 
151         return log
152 
153 
154 def configure_logging(testing: bool = False) -> logging.Logger:
155     """
156     Creates a "prefect" root logger with a `StreamHandler` that has level and formatting
157     set from `prefect.config`.
158 
159     Args:
160         - testing (bool, optional): a boolean specifying whether this configuration
161             is for testing purposes only; this helps us isolate any global state during testing
162             by configuring a "prefect-test-logger" instead of the standard "prefect" logger
163 
164     Returns:
165         - logging.Logger: a configured logging object
166     """
167     name = "prefect-test-logger" if testing else "prefect"
168     logger = logging.getLogger(name)
169     handler = logging.StreamHandler(sys.stdout)
170     formatter = logging.Formatter(context.config.logging.format)
171     formatter.converter = time.gmtime  # type: ignore
172     handler.setFormatter(formatter)
173     logger.addHandler(handler)
174     logger.setLevel(context.config.logging.level)
175 
176     cloud_handler = CloudHandler()
177     cloud_handler.setLevel("DEBUG")
178     logger.addHandler(cloud_handler)
179     return logger
180 
181 
182 prefect_logger = configure_logging()
183 
184 
185 def get_logger(name: str = None) -> logging.Logger:
186     """
187     Returns a "prefect" logger.
188 
189     Args:
190         - name (str): if `None`, the root Prefect logger is returned. If provided, a child
191             logger of the name `"prefect.{name}"` is returned. The child logger inherits
192             the root logger's settings.
193 
194     Returns:
195         - logging.Logger: a configured logging object with the appropriate name
196     """
197     if name is None:
198         return prefect_logger
199     else:
200         return prefect_logger.getChild(name)
201
```
--- 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
@@ -23,6 +23,17 @@
 from prefect.utilities.context import context
 
 
+_original_log_record_factory = logging.getLogRecordFactory()
+
+PREFECT_LOG_RECORD_ATTRIBUTES = (
+    "flow_name",
+    "flow_run_id",
+    "task_name",
+    "task_slug",
+    "task_run_id",
+)
+
+
 class CloudHandler(logging.StreamHandler):
     def __init__(self) -> None:
         super().__init__(sys.stdout)
@@ -151,6 +162,27 @@
         return log
 
 
+def _log_record_context_injector(*args: Any, **kwargs: Any) -> logging.LogRecord:
+    """
+    A custom logger LogRecord Factory that injects selected context parameters into newly created logs.
+
+    Args:
+        - *args: arguments to pass to the original LogRecord Factory
+        - **kwargs: keyword arguments to pass to the original LogRecord Factory
+
+    Returns:
+        - logging.LogRecord: the newly created LogRecord
+    """
+    record = _original_log_record_factory(*args, **kwargs)
+
+    for attr in PREFECT_LOG_RECORD_ATTRIBUTES:
+        value = prefect.context.get(attr, None)
+        if value:
+            setattr(record, attr, value)
+
+    return record
+
+
 def configure_logging(testing: bool = False) -> logging.Logger:
     """
     Creates a "prefect" root logger with a `StreamHandler` that has level and formatting
@@ -164,6 +196,8 @@
     Returns:
         - logging.Logger: a configured logging object
     """
+    logging.setLogRecordFactory(_log_record_context_injector)
+
     name = "prefect-test-logger" if testing else "prefect"
     logger = logging.getLogger(name)
     handler = logging.StreamHandler(sys.stdout)
@@ -194,6 +228,7 @@
     Returns:
         - logging.Logger: a configured logging object with the appropriate name
     """
+
     if name is None:
         return prefect_logger
     else:

{"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@@ -23,6 +23,17 @@\n from prefect.utilities.context import context\n \n \n+_original_log_record_factory = logging.getLogRecordFactory()\n+\n+PREFECT_LOG_RECORD_ATTRIBUTES = (\n+    \"flow_name\",\n+    \"flow_run_id\",\n+    \"task_name\",\n+    \"task_slug\",\n+    \"task_run_id\",\n+)\n+\n+\n class CloudHandler(logging.StreamHandler):\n     def __init__(self) -> None:\n         super().__init__(sys.stdout)\n@@ -151,6 +162,27 @@\n         return log\n \n \n+def _log_record_context_injector(*args: Any, **kwargs: Any) -> logging.LogRecord:\n+    \"\"\"\n+    A custom logger LogRecord Factory that injects selected context parameters into newly created logs.\n+\n+    Args:\n+        - *args: arguments to pass to the original LogRecord Factory\n+        - **kwargs: keyword arguments to pass to the original LogRecord Factory\n+\n+    Returns:\n+        - logging.LogRecord: the newly created LogRecord\n+    \"\"\"\n+    record = _original_log_record_factory(*args, **kwargs)\n+\n+    for attr in PREFECT_LOG_RECORD_ATTRIBUTES:\n+        value = prefect.context.get(attr, None)\n+        if value:\n+            setattr(record, attr, value)\n+\n+    return record\n+\n+\n def configure_logging(testing: bool = False) -> logging.Logger:\n     \"\"\"\n     Creates a \"prefect\" root logger with a `StreamHandler` that has level and formatting\n@@ -164,6 +196,8 @@\n     Returns:\n         - logging.Logger: a configured logging object\n     \"\"\"\n+    logging.setLogRecordFactory(_log_record_context_injector)\n+\n     name = \"prefect-test-logger\" if testing else \"prefect\"\n     logger = logging.getLogger(name)\n     handler = logging.StreamHandler(sys.stdout)\n@@ -194,6 +228,7 @@\n     Returns:\n         - logging.Logger: a configured logging object with the appropriate name\n     \"\"\"\n+\n     if name is None:\n         return prefect_logger\n     else:\n", "issue": "Add Flow and Task identifying information to LogRecords\n## Current behavior\r\n- All log entries have the default [LogRecord attributes](https://docs.python.org/3/library/logging.html#logrecord-objects)... \r\n- There is no universally identifiable flow slug\r\n\r\n## Proposed behavior\r\n...It would be useful to attach Core-specific information to log entries:\r\n- `task_name` & `task_slug`: available only on entries emitted from Tasks and TaskRunners\r\n- `flow_name` & `flow_slug`: available only on entries emitted from Flows and FlowRunners\r\n\r\nThe addition of the `flow_slug` makes is possible to uniquely identify flows other than by name.\r\n\r\n## Example\r\nOne can now take a single stream of log entries during a local flow run and separate it by entity without parsing the `LogRecord.msg` field (which is unreliable). This makes is possible to build richer feedback interfaces for filtering, organizing, and otherwise selectively manipulating log entries.\r\n\r\nHere is an example CLI interface that would be made possible by theses features:\r\n\r\n![local-progress](https://user-images.githubusercontent.com/590471/72913913-dbd00880-3d0b-11ea-8c85-a0e74e1c9551.gif)\r\n\r\n(Given the following flow):\r\n```python\r\nfrom utils import seed, random_data\r\nimport prefect\r\n\r\n@prefect.task\r\ndef extract():\r\n    return seed.get()\r\n\r\n@prefect.task\r\ndef transform(seed):\r\n    logger = prefect.context.get(\"logger\")\r\n    data = []\r\n    for x in range(10):\r\n        c_data = random_data(seed)\r\n        logger.info(\"debug info: {}\".format(c_data))\r\n        data.append(c_data)\r\n    return data\r\n\r\n@prefect.task\r\ndef load(data):\r\n    # do something with the data....\r\n    pass\r\n\r\nwith prefect.Flow(\"ETL\") as flow:\r\n    e = extract()\r\n    t = transform(e)\r\n    load(t)\r\n```\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                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"}], "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\n_original_log_record_factory = logging.getLogRecordFactory()\n\nPREFECT_LOG_RECORD_ATTRIBUTES = (\n    \"flow_name\",\n    \"flow_run_id\",\n    \"task_name\",\n    \"task_slug\",\n    \"task_run_id\",\n)\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 _log_record_context_injector(*args: Any, **kwargs: Any) -> logging.LogRecord:\n    \"\"\"\n    A custom logger LogRecord Factory that injects selected context parameters into newly created logs.\n\n    Args:\n        - *args: arguments to pass to the original LogRecord Factory\n        - **kwargs: keyword arguments to pass to the original LogRecord Factory\n\n    Returns:\n        - logging.LogRecord: the newly created LogRecord\n    \"\"\"\n    record = _original_log_record_factory(*args, **kwargs)\n\n    for attr in PREFECT_LOG_RECORD_ATTRIBUTES:\n        value = prefect.context.get(attr, None)\n        if value:\n            setattr(record, attr, value)\n\n    return record\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    logging.setLogRecordFactory(_log_record_context_injector)\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\n    if name is None:\n        return prefect_logger\n    else:\n        return prefect_logger.getChild(name)\n", "path": "src/prefect/utilities/logging.py"}]}