Structured Logging with Datadog #5142
Replies: 1 comment 3 replies
-
|
Thanks for sharing! I'll keep this in mind while I work on logging in Orion. I've already got a stub for a JSON logger that you could turn on with an environment variable. |
Beta Was this translation helpful? Give feedback.
-
|
No problem! Great to hear that, I think it would be super helpful to add as built-in functionality. |
Beta Was this translation helpful? Give feedback.
-
|
@speedyturkey Just curious if you ever refined this. I'm looking into accomplishing the same thing (Prefect running on Kubernetes sending data to DataDog). We are using the DataDog daemon in Kubernetes to ship our logs from AKS in Azure to DataDog. This first thing I noticed was everything seemed to come in as errors. Not sure how good of an approach this is but I was able to use the following to get better level handling: class PrefectJsonFormatter(jsonlogger.JsonFormatter):
def add_fields(self, log_record, record, message_dict):
super().add_fields(log_record, record, message_dict)
if log_record.get('level'):
log_record['level'] = log_record['level'].upper()
else:
log_record['level'] = record.levelnameSecondly, with the addition of a standard handler to the logger, DataDog is ingesting the duplicate messages from Kubernetes which doubles the cost of shipping the logs to DataDog. I'm having a hard time trying to cleanly keep the logs in a format for Prefect Server and get the structured logs to DataDog without the duplication. Another caveat to this approach is the standard Prefect logs for tasks and flows don't get logged to DataDog. I'm glad to hear this may be more native in Orion. @madkinsz - I wanted to provide an example use case in the case it might help with future plans. We use DataDog for a centralized location to build alerts along with dashboards. I really want to have clean information from Prefect about when tasks and flows fail (including differentiating between final failures and failures that lead to retries that may succeed). This way I can design alerts on things like deviations in flow/task performance, failure rates, etc. |
Beta Was this translation helpful? Give feedback.
-
|
I don't use Orion Server/Cloud at the moment, so this will be missing a few pieces, but I actually just re-wrote our logging set up and it handles this kind of logic. It logs to stdout and a file (our original setup), as well as a database via a SqlAlchemy logger I set up. The biggest annoyance I had was duplicate logs to stdout because of Prefect's standard logger, so I popped that one out so that it would only log once. Not sure if there is a better way (logging is not my forte) and/or what this might mess up with logs to Cloud. At this point I have structured logs using structlog that go to the database, and use the standard Prefect formatting for stdout and file. import logging
import os
import platform
import sys
from datetime import datetime
from typing import Optional, Union
import prefect
import structlog
from prefect.utilities.logging import prefect_logger
from structlog.contextvars import merge_contextvars
from structlog.processors import CallsiteParameter
class LogConfig:
def __init__(
self,
db_name: Optional[str] = None,
job_id: Optional[Union[int, str]] = None,
):
self.root_logger = logging.getLogger()
self.root_logger.setLevel(logging.INFO)
self.db_formatter: logging.Formatter = None
self.prefect_formatter: logging.Formatter = logging.Formatter(
prefect.context.config.logging.format,
datefmt=prefect.context.config.logging.datefmt,
)
self.log_path = None
if db_name and job_id:
self.log_path = self.get_log_path(db_name, job_id)
self.configured_handlers: set = set()
self.setup_structlog()
self.setup_logging()
def setup_structlog(self):
shared_processors = [
structlog.stdlib.add_log_level,
structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S"),
structlog.processors.CallsiteParameterAdder(
parameters=[CallsiteParameter.MODULE, CallsiteParameter.FUNC_NAME]
),
]
self.db_formatter = structlog.stdlib.ProcessorFormatter(
# These run ONLY on `logging` entries that do NOT originate within
# structlog.
foreign_pre_chain=shared_processors, # type: ignore
# These run on ALL entries after the pre_chain is done.
processors=[
# Remove _record & _from_structlog.
merge_contextvars,
structlog.stdlib.add_logger_name,
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.JSONRenderer(sort_keys=False),
],
)
## wrap_for_formatter can't be added to shared_processors list - causes errors for some reason
structlog.configure(
processors=shared_processors # type: ignore
+ [
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
logger_factory=structlog.stdlib.LoggerFactory(),
cache_logger_on_first_use=True,
)
def get_log_path(self, db_name: str, job_id: Union[str, int]):
if self.log_path:
return self.log_path
self.log_path = os.path.join(
"//LogServer/PrefectLogs",
platform.node(),
db_name,
str(job_id),
datetime.strftime(datetime.now(), "%Y%m%d") + ".log",
)
if not os.path.exists(os.path.dirname(self.log_path)):
os.makedirs(os.path.dirname(self.log_path))
return self.log_path
def setup_logging(self):
self.remove_prefect_stdout()
self.add_handler(logging.StreamHandler, self.prefect_formatter, sys.stdout)
self.add_handler(SQLAlchemyHandler, self.db_formatter)
if self.log_path:
self.add_handler(
logging.FileHandler, self.prefect_formatter, filename=self.log_path
)
def add_handler(
self,
handler_cls: logging.Handler,
formatter: logging.Formatter,
*handler_args,
**handler_kwargs
):
name = handler_cls.__name__ # type: ignore
if name in self.configured_handlers:
return
handler: logging.Handler = handler_cls(*handler_args, **handler_kwargs) # type: ignore
handler.setFormatter(formatter)
self.root_logger.addHandler(handler)
self.configured_handlers.add(name)
def remove_prefect_stdout(self):
for i, lgger in enumerate(prefect_logger.handlers):
if isinstance(lgger, logging.StreamHandler):
prefect_logger.handlers.pop(i) |
Beta Was this translation helpful? Give feedback.
-
I wanted to share some basic information about how I am sending structured logs from prefect flows to Datadog, prompted by @kvnkho's request in the Prefect Slack community. I am happy to share how my organization does this, but I am not an expert on either Kubernetes or Datadog and I'm sure there are other good ways to accomplish this. I am still developing this solution, so it is likely that there are improvements to be made.
Background
I am running Prefect on Kubernetes (EKS). When a running pod logs to stdout, the logs are processed by fluentbit, which is running as a daemonset in each of our clusters. We use labels to tell fluentbit which Cloudwatch log group to send logs to:
cloudwatch/log_group_name. We then use a Kinesis firehose to copy the logs from Cloudwatch to Datadog. I believe this is accomplished using an account-wide subscription filter (I'm not clear on the details).Prefect
Given the above infrastructure, this part is easy. The goal is to send structured logs as json, because Datadog knows how to ingest these and will allow you to create facets based on flow name, task name, etc for easy search and filtering.
We have a shared Python library which is installed in the base image used for all Prefect flows, which contains a number of common utility functions. The Prefect logging file looks like this:
Sample usage is as follows. Note that because we are adding the json handler, the output does mix standard output with json output - I like this because I can choose to view logs in Prefect Cloud directly, or in Datadog. Note that Prefect's default logger adds
flow_name,task_name, etc automatically which is very useful.I'm sure I will come to refine this setup over time but hopefully this is helpful to someone.
Beta Was this translation helpful? Give feedback.
All reactions