We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Prefect logger doesn't use local Python logger timezone configuration
## Opened from the [Prefect Public Slack Community](https://join.slack.com/t/prefect-public/shared_invite/enQtNzE5OTU3OTQwNzc1LTQ5M2FkZmQzZjI0ODg1ZTBmOTc0ZjVjYWFjMWExZDAyYzBmYjVmMTE1NTQ1Y2IxZTllOTc4MmI3NzYxMDlhYWU)

**dcshah88**: Hello Everyone, 

Have started looking into prefect core library recently and exploring it to use for simple use cases.

One thing I am struggling with right now is to provide a different timezone to prefect instead of default UTC.

By default, prefect is generating UTC timestamps in logs but wanted to change it to my local timezone. 

Could you please help ?

**znicholasbrown**: Hi <@U01CWF05D6Y> - Prefect uses Python's built-in logging module; please look at the <https://docs.python.org/3/library/logging.html|Python logging documentation> for info on how to configure your local Python logger.

**dcshah88**: Thanks <@UN6FTLFAS> : what I am confused about is, if I use python logging without prefect, it prints logs using local timezone by default but when used with prefect it prints timestamps in UTC.

looks like prefect is overriding default python logging behavior.

So was wondering if prefect has any config option to provide which timezone to use while printing logs ?

**znicholasbrown**: Ah, that's a good flag - it looks like we're explicitly defaulting to UTC instead of local in the logger, which is probably unintentional. Let me see if there's a reason for that and if not we can PR it :slightly_smiling_face:

**znicholasbrown**: I think this is something we can open a ticket for <@U01CWF05D6Y> - <https://github.com/PrefectHQ/prefect/blob/master/src/prefect/utilities/logging.py#L220|this line> is where the timezone is set in the logger; I'll use <@ULVA73B9P> to open the ticket from this thread :slightly_smiling_face:

**znicholasbrown**: <@ULVA73B9P> open "Prefect logger doesn't use local Python logger timezone configuration"

Original thread can be found [here](https://prefect-community.slack.com/archives/CL09KU1K7/p1603234095357600?thread_ts=1603234095.357600&cid=CL09KU1K7).



--- 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
3 retrieving loggers for 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 Empty, Queue
18 from typing import Any
19 
20 import pendulum
21 
22 import prefect
23 from prefect.utilities.context import context
24 
25 _original_log_record_factory = logging.getLogRecordFactory()
26 
27 PREFECT_LOG_RECORD_ATTRIBUTES = (
28     "flow_name",
29     "flow_run_id",
30     "task_name",
31     "task_slug",
32     "task_run_id",
33 )
34 
35 
36 class CloudHandler(logging.StreamHandler):
37     def __init__(self) -> None:
38         super().__init__(sys.stdout)
39         self.client = None
40         self.logger = logging.getLogger("CloudHandler")
41         handler = logging.StreamHandler(sys.stdout)
42         formatter = logging.Formatter(
43             context.config.logging.format, context.config.logging.datefmt
44         )
45         formatter.converter = time.gmtime  # type: ignore
46         handler.setFormatter(formatter)
47         self.logger.addHandler(handler)
48         self.logger.setLevel(context.config.logging.level)
49 
50     @property
51     def queue(self) -> Queue:
52         if not hasattr(self, "_queue"):
53             self._queue = Queue()  # type: Queue
54             self._flush = False
55             self.start()
56         return self._queue
57 
58     def flush(self) -> None:
59         self._flush = True
60         if self.client is not None:
61             self.batch_upload()
62             self._thread.join()
63 
64     def batch_upload(self) -> None:
65         logs = []
66         try:
67             while True:
68                 log = self.queue.get(False)
69                 logs.append(log)
70         except Empty:
71             pass
72 
73         if logs:
74             try:
75                 assert self.client is not None
76                 self.client.write_run_logs(logs)
77             except Exception as exc:
78                 message = "Failed to write log with error: {}".format(str(exc))
79                 self.logger.critical(message)
80 
81                 # Attempt to write batch error log otherwise log invalid cloud communication
82                 try:
83                     assert self.client is not None
84                     self.client.write_run_logs([self._make_error_log(message)])
85                 except Exception:
86                     self.logger.critical("Unable to write logs to Prefect Cloud")
87 
88     def _monitor(self) -> None:
89         while not self._flush:
90             self.batch_upload()
91             time.sleep(self.heartbeat)
92 
93     def __del__(self) -> None:
94         if hasattr(self, "_thread"):
95             self.flush()
96             atexit.unregister(self.flush)
97 
98     def start(self) -> None:
99         if not hasattr(self, "_thread"):
100             self.heartbeat = context.config.cloud.logging_heartbeat
101             self._thread = t = threading.Thread(
102                 target=self._monitor, name="PrefectCloudLoggingThread"
103             )
104             t.daemon = True
105             t.start()
106             atexit.register(self.flush)
107 
108     def put(self, log: dict) -> None:
109         try:
110             json.dumps(log)  # make sure the payload is serializable
111             self.queue.put(log)
112         except TypeError as exc:
113             message = "Failed to write log with error: {}".format(str(exc))
114             self.logger.critical(message)
115 
116             self.queue.put(self._make_error_log(message))
117 
118     def emit(self, record) -> None:  # type: ignore
119         # if we shouldn't log to cloud, don't emit
120         if not prefect.context.config.logging.log_to_cloud:
121             return
122 
123         try:
124             from prefect.client import Client
125 
126             if self.client is None:
127                 self.client = Client()  # type: ignore
128 
129             assert isinstance(self.client, Client)  # mypy assert
130 
131             record_dict = record.__dict__.copy()
132 
133             # ensures emitted logs respect configured logging level
134             config_level = getattr(
135                 logging, prefect.context.config.logging.level, logging.INFO
136             )
137 
138             if record_dict["levelno"] < config_level:
139                 return
140 
141             # remove potentially non-json serializable formatting args
142             record_dict.pop("args", None)
143 
144             log = dict()
145             log["flow_run_id"] = prefect.context.get("flow_run_id", None)
146             log["task_run_id"] = prefect.context.get("task_run_id", None)
147             log["timestamp"] = pendulum.from_timestamp(
148                 record_dict.pop("created", time.time())
149             ).isoformat()
150             log["name"] = record_dict.pop("name", None)
151             log["message"] = record_dict.pop("message", None)
152             log["level"] = record_dict.pop("levelname", None)
153 
154             if record_dict.get("exc_text") is not None:
155                 log["message"] += "\n" + record_dict.pop("exc_text", "")
156                 record_dict.pop("exc_info", None)
157 
158             log["info"] = record_dict
159             self.put(log)
160         except Exception as exc:
161             message = "Failed to write log with error: {}".format(str(exc))
162             self.logger.critical(message)
163 
164             self.put(self._make_error_log(message))
165 
166     def _make_error_log(self, message: str) -> dict:
167         log = dict()
168         log["flow_run_id"] = prefect.context.get("flow_run_id", None)
169         log["timestamp"] = pendulum.from_timestamp(time.time()).isoformat()
170         log["name"] = self.logger.name
171         log["message"] = message
172         log["level"] = "CRITICAL"
173         log["info"] = {}
174 
175         return log
176 
177 
178 def _log_record_context_injector(*args: Any, **kwargs: Any) -> logging.LogRecord:
179     """
180     A custom logger LogRecord Factory that injects selected context parameters into newly
181     created logs.
182 
183     Args:
184         - *args: arguments to pass to the original LogRecord Factory
185         - **kwargs: keyword arguments to pass to the original LogRecord Factory
186 
187     Returns:
188         - logging.LogRecord: the newly created LogRecord
189     """
190     record = _original_log_record_factory(*args, **kwargs)
191 
192     additional_attrs = context.config.logging.get("log_attributes", [])
193 
194     for attr in PREFECT_LOG_RECORD_ATTRIBUTES + tuple(additional_attrs):
195         value = prefect.context.get(attr, None)
196         if value or attr in additional_attrs:
197             setattr(record, attr, value)
198 
199     return record
200 
201 
202 def _create_logger(name: str) -> logging.Logger:
203     """
204     Creates a logger with a `StreamHandler` that has level and formatting
205     set from `prefect.config`.
206 
207     Args:
208         - name (str): Name to use for logger.
209 
210     Returns:
211         - logging.Logger: a configured logging object
212     """
213     logging.setLogRecordFactory(_log_record_context_injector)
214 
215     logger = logging.getLogger(name)
216     handler = logging.StreamHandler(sys.stdout)
217     formatter = logging.Formatter(
218         context.config.logging.format, context.config.logging.datefmt
219     )
220     formatter.converter = time.gmtime  # type: ignore
221     handler.setFormatter(formatter)
222     logger.addHandler(handler)
223     logger.setLevel(context.config.logging.level)
224 
225     # we set the cloud handler to DEBUG level
226     # but the handler itself will dynamically respond
227     # to the configured level in the emit() method call
228     cloud_handler = CloudHandler()
229     cloud_handler.setLevel("DEBUG")
230     logger.addHandler(cloud_handler)
231     return logger
232 
233 
234 def configure_logging(testing: bool = False) -> logging.Logger:
235     """
236     Creates a "prefect" root logger with a `StreamHandler` that has level and formatting
237     set from `prefect.config`.
238 
239     Args:
240         - testing (bool, optional): a boolean specifying whether this configuration
241             is for testing purposes only; this helps us isolate any global state during testing
242             by configuring a "prefect-test-logger" instead of the standard "prefect" logger
243 
244     Returns:
245         - logging.Logger: a configured logging object
246     """
247     name = "prefect-test-logger" if testing else "prefect"
248     return _create_logger(name)
249 
250 
251 context.logger = prefect_logger = configure_logging()
252 
253 
254 def configure_extra_loggers() -> None:
255     """
256     Creates a "Prefect" configured logger for all strings in extra_loggers config list.
257     The logging.extra_loggers config defaults to an empty list.
258     """
259     loggers = context.config.logging.get("extra_loggers", [])
260     for l in loggers:
261         _create_logger(l)
262 
263 
264 configure_extra_loggers()
265 
266 
267 def create_diagnostic_logger(name: str) -> logging.Logger:
268     """
269     Create a logger that does not use the `CloudHandler` but preserves all other
270     Prefect logging configuration.  For diagnostic / debugging / internal use only.
271     """
272     logger = _create_logger(name)
273     logger.handlers = [h for h in logger.handlers if not isinstance(h, CloudHandler)]
274     return logger
275 
276 
277 def get_logger(name: str = None) -> logging.Logger:
278     """
279     Returns a "prefect" logger.
280 
281     Args:
282         - name (str): if `None`, the root Prefect logger is returned. If provided, a child
283             logger of the name `"prefect.{name}"` is returned. The child logger inherits
284             the root logger's settings.
285 
286     Returns:
287         - logging.Logger: a configured logging object with the appropriate name
288     """
289 
290     if name is None:
291         return prefect_logger
292     else:
293         return prefect_logger.getChild(name)
294 
295 
296 class RedirectToLog:
297     """
298     Custom redirect of stdout messages to logs
299 
300     Args:
301         - logger (logging.Logger, optional): an optional logger to redirect stdout. If
302             not provided a logger names `stdout` will be created.
303     """
304 
305     def __init__(self, logger: logging.Logger = None) -> None:
306         self.stdout_logger = logger or get_logger("stdout")
307 
308     def write(self, s: str) -> None:
309         """
310         Write message from stdout to a prefect logger.
311         Note: blank newlines will not be logged.
312 
313         Args:
314             s (str): the message from stdout to be logged
315         """
316         if not isinstance(s, str):
317             # stdout is expecting str
318             raise TypeError(f"string argument expected, got {type(s)}")
319 
320         if s.strip():
321             self.stdout_logger.info(s)
322 
323     def flush(self) -> None:
324         """
325         Implemented flush operation for logger handler
326         """
327         for handler in self.stdout_logger.handlers:
328             handler.flush()
329
```
--- 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
@@ -217,7 +217,6 @@
     formatter = logging.Formatter(
         context.config.logging.format, context.config.logging.datefmt
     )
-    formatter.converter = time.gmtime  # type: ignore
     handler.setFormatter(formatter)
     logger.addHandler(handler)
     logger.setLevel(context.config.logging.level)

{"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@@ -217,7 +217,6 @@\n     formatter = logging.Formatter(\n         context.config.logging.format, context.config.logging.datefmt\n     )\n-    formatter.converter = time.gmtime  # type: ignore\n     handler.setFormatter(formatter)\n     logger.addHandler(handler)\n     logger.setLevel(context.config.logging.level)\n", "issue": "Prefect logger doesn't use local Python logger timezone configuration\n## Opened from the [Prefect Public Slack Community](https://join.slack.com/t/prefect-public/shared_invite/enQtNzE5OTU3OTQwNzc1LTQ5M2FkZmQzZjI0ODg1ZTBmOTc0ZjVjYWFjMWExZDAyYzBmYjVmMTE1NTQ1Y2IxZTllOTc4MmI3NzYxMDlhYWU)\n\n**dcshah88**: Hello Everyone, \n\nHave started looking into prefect core library recently and exploring it to use for simple use cases.\n\nOne thing I am struggling with right now is to provide a different timezone to prefect instead of default UTC.\n\nBy default, prefect is generating UTC timestamps in logs but wanted to change it to my local timezone. \n\nCould you please help ?\n\n**znicholasbrown**: Hi <@U01CWF05D6Y> - Prefect uses Python's built-in logging module; please look at the <https://docs.python.org/3/library/logging.html|Python logging documentation> for info on how to configure your local Python logger.\n\n**dcshah88**: Thanks <@UN6FTLFAS> : what I am confused about is, if I use python logging without prefect, it prints logs using local timezone by default but when used with prefect it prints timestamps in UTC.\n\nlooks like prefect is overriding default python logging behavior.\n\nSo was wondering if prefect has any config option to provide which timezone to use while printing logs ?\n\n**znicholasbrown**: Ah, that's a good flag - it looks like we're explicitly defaulting to UTC instead of local in the logger, which is probably unintentional. Let me see if there's a reason for that and if not we can PR it :slightly_smiling_face:\n\n**znicholasbrown**: I think this is something we can open a ticket for <@U01CWF05D6Y> - <https://github.com/PrefectHQ/prefect/blob/master/src/prefect/utilities/logging.py#L220|this line> is where the timezone is set in the logger; I'll use <@ULVA73B9P> to open the ticket from this thread :slightly_smiling_face:\n\n**znicholasbrown**: <@ULVA73B9P> open \"Prefect logger doesn't use local Python logger timezone configuration\"\n\nOriginal thread can be found [here](https://prefect-community.slack.com/archives/CL09KU1K7/p1603234095357600?thread_ts=1603234095.357600&cid=CL09KU1K7).\n\n\n", "before_files": [{"content": "\"\"\"\nUtility functions for interacting with and configuring logging.  The main entrypoint for\nretrieving loggers for customization 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 Empty, Queue\nfrom typing import Any\n\nimport pendulum\n\nimport prefect\nfrom prefect.utilities.context import context\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(\n            context.config.logging.format, context.config.logging.datefmt\n        )\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:\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\n            # ensures emitted logs respect configured logging level\n            config_level = getattr(\n                logging, prefect.context.config.logging.level, logging.INFO\n            )\n\n            if record_dict[\"levelno\"] < config_level:\n                return\n\n            # remove potentially non-json serializable formatting args\n            record_dict.pop(\"args\", None)\n\n            log = dict()\n            log[\"flow_run_id\"] = prefect.context.get(\"flow_run_id\", None)\n            log[\"task_run_id\"] = 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[\"flow_run_id\"] = 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\n    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    additional_attrs = context.config.logging.get(\"log_attributes\", [])\n\n    for attr in PREFECT_LOG_RECORD_ATTRIBUTES + tuple(additional_attrs):\n        value = prefect.context.get(attr, None)\n        if value or attr in additional_attrs:\n            setattr(record, attr, value)\n\n    return record\n\n\ndef _create_logger(name: str) -> logging.Logger:\n    \"\"\"\n    Creates a logger with a `StreamHandler` that has level and formatting\n    set from `prefect.config`.\n\n    Args:\n        - name (str): Name to use for logger.\n\n    Returns:\n        - logging.Logger: a configured logging object\n    \"\"\"\n    logging.setLogRecordFactory(_log_record_context_injector)\n\n    logger = logging.getLogger(name)\n    handler = logging.StreamHandler(sys.stdout)\n    formatter = logging.Formatter(\n        context.config.logging.format, context.config.logging.datefmt\n    )\n    formatter.converter = time.gmtime  # type: ignore\n    handler.setFormatter(formatter)\n    logger.addHandler(handler)\n    logger.setLevel(context.config.logging.level)\n\n    # we set the cloud handler to DEBUG level\n    # but the handler itself will dynamically respond\n    # to the configured level in the emit() method call\n    cloud_handler = CloudHandler()\n    cloud_handler.setLevel(\"DEBUG\")\n    logger.addHandler(cloud_handler)\n    return logger\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    return _create_logger(name)\n\n\ncontext.logger = prefect_logger = configure_logging()\n\n\ndef configure_extra_loggers() -> None:\n    \"\"\"\n    Creates a \"Prefect\" configured logger for all strings in extra_loggers config list.\n    The logging.extra_loggers config defaults to an empty list.\n    \"\"\"\n    loggers = context.config.logging.get(\"extra_loggers\", [])\n    for l in loggers:\n        _create_logger(l)\n\n\nconfigure_extra_loggers()\n\n\ndef create_diagnostic_logger(name: str) -> logging.Logger:\n    \"\"\"\n    Create a logger that does not use the `CloudHandler` but preserves all other\n    Prefect logging configuration.  For diagnostic / debugging / internal use only.\n    \"\"\"\n    logger = _create_logger(name)\n    logger.handlers = [h for h in logger.handlers if not isinstance(h, CloudHandler)]\n    return logger\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\n\nclass RedirectToLog:\n    \"\"\"\n    Custom redirect of stdout messages to logs\n\n    Args:\n        - logger (logging.Logger, optional): an optional logger to redirect stdout. If\n            not provided a logger names `stdout` will be created.\n    \"\"\"\n\n    def __init__(self, logger: logging.Logger = None) -> None:\n        self.stdout_logger = logger or get_logger(\"stdout\")\n\n    def write(self, s: str) -> None:\n        \"\"\"\n        Write message from stdout to a prefect logger.\n        Note: blank newlines will not be logged.\n\n        Args:\n            s (str): the message from stdout to be logged\n        \"\"\"\n        if not isinstance(s, str):\n            # stdout is expecting str\n            raise TypeError(f\"string argument expected, got {type(s)}\")\n\n        if s.strip():\n            self.stdout_logger.info(s)\n\n    def flush(self) -> None:\n        \"\"\"\n        Implemented flush operation for logger handler\n        \"\"\"\n        for handler in self.stdout_logger.handlers:\n            handler.flush()\n", "path": "src/prefect/utilities/logging.py"}], "after_files": [{"content": "\"\"\"\nUtility functions for interacting with and configuring logging.  The main entrypoint for\nretrieving loggers for customization 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 Empty, Queue\nfrom typing import Any\n\nimport pendulum\n\nimport prefect\nfrom prefect.utilities.context import context\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(\n            context.config.logging.format, context.config.logging.datefmt\n        )\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:\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\n            # ensures emitted logs respect configured logging level\n            config_level = getattr(\n                logging, prefect.context.config.logging.level, logging.INFO\n            )\n\n            if record_dict[\"levelno\"] < config_level:\n                return\n\n            # remove potentially non-json serializable formatting args\n            record_dict.pop(\"args\", None)\n\n            log = dict()\n            log[\"flow_run_id\"] = prefect.context.get(\"flow_run_id\", None)\n            log[\"task_run_id\"] = 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[\"flow_run_id\"] = 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\n    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    additional_attrs = context.config.logging.get(\"log_attributes\", [])\n\n    for attr in PREFECT_LOG_RECORD_ATTRIBUTES + tuple(additional_attrs):\n        value = prefect.context.get(attr, None)\n        if value or attr in additional_attrs:\n            setattr(record, attr, value)\n\n    return record\n\n\ndef _create_logger(name: str) -> logging.Logger:\n    \"\"\"\n    Creates a logger with a `StreamHandler` that has level and formatting\n    set from `prefect.config`.\n\n    Args:\n        - name (str): Name to use for logger.\n\n    Returns:\n        - logging.Logger: a configured logging object\n    \"\"\"\n    logging.setLogRecordFactory(_log_record_context_injector)\n\n    logger = logging.getLogger(name)\n    handler = logging.StreamHandler(sys.stdout)\n    formatter = logging.Formatter(\n        context.config.logging.format, context.config.logging.datefmt\n    )\n    handler.setFormatter(formatter)\n    logger.addHandler(handler)\n    logger.setLevel(context.config.logging.level)\n\n    # we set the cloud handler to DEBUG level\n    # but the handler itself will dynamically respond\n    # to the configured level in the emit() method call\n    cloud_handler = CloudHandler()\n    cloud_handler.setLevel(\"DEBUG\")\n    logger.addHandler(cloud_handler)\n    return logger\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    return _create_logger(name)\n\n\ncontext.logger = prefect_logger = configure_logging()\n\n\ndef configure_extra_loggers() -> None:\n    \"\"\"\n    Creates a \"Prefect\" configured logger for all strings in extra_loggers config list.\n    The logging.extra_loggers config defaults to an empty list.\n    \"\"\"\n    loggers = context.config.logging.get(\"extra_loggers\", [])\n    for l in loggers:\n        _create_logger(l)\n\n\nconfigure_extra_loggers()\n\n\ndef create_diagnostic_logger(name: str) -> logging.Logger:\n    \"\"\"\n    Create a logger that does not use the `CloudHandler` but preserves all other\n    Prefect logging configuration.  For diagnostic / debugging / internal use only.\n    \"\"\"\n    logger = _create_logger(name)\n    logger.handlers = [h for h in logger.handlers if not isinstance(h, CloudHandler)]\n    return logger\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\n\nclass RedirectToLog:\n    \"\"\"\n    Custom redirect of stdout messages to logs\n\n    Args:\n        - logger (logging.Logger, optional): an optional logger to redirect stdout. If\n            not provided a logger names `stdout` will be created.\n    \"\"\"\n\n    def __init__(self, logger: logging.Logger = None) -> None:\n        self.stdout_logger = logger or get_logger(\"stdout\")\n\n    def write(self, s: str) -> None:\n        \"\"\"\n        Write message from stdout to a prefect logger.\n        Note: blank newlines will not be logged.\n\n        Args:\n            s (str): the message from stdout to be logged\n        \"\"\"\n        if not isinstance(s, str):\n            # stdout is expecting str\n            raise TypeError(f\"string argument expected, got {type(s)}\")\n\n        if s.strip():\n            self.stdout_logger.info(s)\n\n    def flush(self) -> None:\n        \"\"\"\n        Implemented flush operation for logger handler\n        \"\"\"\n        for handler in self.stdout_logger.handlers:\n            handler.flush()\n", "path": "src/prefect/utilities/logging.py"}]}