Hi again, I have a question about extra logging. We have own Python module 'bear' whereas logging is defined as:

logger = logging.getLogger(__ _name ___)

and I want log messages from that module to appear in Prefect. I registered extra logger, and log messages defined in Task (AWSPOC) appear, but not the log messages defined inside 'bear'. My code for "prefect_flow.py" :

import logging

format_string = "%(asctime)s %(name)s [%(levelname)s] %(message)s"
logger = logging.getLogger('magic_logger')
logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler()
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter(format_string)
handler.setFormatter(formatter)
logger.addHandler(handler)

class AWSPOC(Task):

    def __init__(self, name: str, config_file: str):
        self.config_file = config_file
        super().__init__(name=name)

    def run(self):
        configuration = load_file(self.config_file)
        loader = Loader(configuration=configuration)
        <http://logger.info|logger.info>("This message appears in Prefect log output")
        loader.run()

executor = LocalDaskExecutor()
host_config = {...}
storage = S3(...)
env = {
    "PREFECT__LOGGING__EXTRA_LOGGERS": "['magic_logger']"
}
docker_run_config = DockerRun(image=..., host_config=host_config, env=env)

with Flow(name="AWS POC", executor=executor, storage=storage, run_config=docker_run_config) as flow:
    task = AWSPOC(name='...', config_file='...')
    task()

flow.register(project_name='AWS POC')

Where I'm wrong? The only idea I have is that 'bear' should have it's own named logger.

This looks fine on first glance. Will look more later

also have no ideas

Oh sorry slipped my mind to return. Looking now

magic_logger is not defined in custom module. It's just above in the script. Customer module has just "logger = logging.getLogger(name)" and that' it.

Ahh I see. Sorry I misunderstood the first question. I think S3 storage is still pickle based by default. Do you have

stored_as_script=True

?

yeap

There is no stream in your StreamHandler, maybe you need

logging.StreamHandler(sys.stdout)

? Also, just confirming you have debug level logs set on the flow as well?

I've added sys.stdout. Really missed that, but nothing changed

This is a full working example of using a custom module

Is there any extra support available, personal, paid or any? I cannot explain what is wrong with it. If I run library out of Prefect logs appear as usual. Adding Prefect - logs disappear and even if flow fails inside, in Prefect flow still completes successfully.

# poc.py
from custom_module import Loader

logger = logging.getLogger('custom_logger')

class POC(Task):

    def __init__(self, name: str, config_file: str):
        self.config_file = config_file
        super().__init__(name=name)

    def run(self):
        configuration = load_file(self.config_file)
        loader = Loader(configuration=configuration)
        <http://logger.info|logger.info>(f"Starting POC Loader. This message appears. Logger is {logger.name}")
        loader.start()
        <http://logger.info|logger.info>("Completed POC Loader. This message appears")

env['PREFECT__LOGGING__EXTRA_LOGGERS'] = "['custom_logger']"

storage = S3(stored_as_script=True, local_script_path=f'poc.py', bucket='...',
             client_options={'aws_access_key_id': ...),
                             'aws_secret_access_key': ...)}
             )

docker_run_config = DockerRun(image='...', env=env)

with Flow(name="POC", executor=LocalDaskExecutor(), storage=storage, run_config=docker_run_config) as flow:
    load_data = POC(name="POC", config_file='/root/prefect/flows/poc/poc.json')
    load_data()

flow.register(project_name='AWS')

custom_module:

######
custom_module.py

def init_logger(name: str = 'custom_logger', level=logging.DEBUG, stream=sys.stdout,
                format_string: str = "%(asctime)s %(name)s %(levelname)s %(message)s"):
    logger = logging.getLogger(name)
    logger.setLevel(level)
    handler = logging.StreamHandler(stream)
    handler.setLevel(level)
    formatter = logging.Formatter(format_string)
    handler.setFormatter(formatter)
    logger.addHandler(handler)
	
init_logger(level=<http://logging.INFO|logging.INFO>)
logger = logging.getLogger('custom_logger')


class Loader:
	def __init__(self):
		<http://logger.info|logger.info>("This message doesn't appear in Prefect")
		
	def start():
		<http://logger.info|logger.info>("This message doesn't appear in Prefect")

Looking into this now

doing

flow.register()

didnt work for you?

need few more minutes to pull prefect locally 🙂

Ah ok lol

local run also doesn't show any messages from custom module. Only those that are in POC(Task) are visible. Also you've mentioned that you modified custom_module, while I don't see any difference.

So I see them. I think I may have just added

self

to the

start

method for that

Yes. Ok it started working, but I cannot explain why. Need to compare step by step. I also had to add working_dir to run_config:

run_config = LocalRun(env={"PREFECT__LOGGING__EXTRA_LOGGERS": "['custom_logger']"},
                      working_dir='.....')

Ah ok that’s fine. I can’t explain fully either myself for now, but at least we have a working example to start with 😅

@Kevin Kho, ok so the root cause is a combination of factors. 1. missing definition of prefect.context.logger 2. https://prefect-community.slack.com/archives/CL09KU1K7/p1648115064317039 - this thread. 3. https://prefect-community.slack.com/archives/CL09KU1K7/p1648114337459679 - this thread. Overall the messages inside the custom_module were generated from threads. Since multithreading in Prefect works a bit indifferent, that is what happened: task was submitted form Prefect, threads in custom_module created and were running, but Prefect immediately succeeded the task and all threads output was missed. Thus: task completed, no logs. I think the proper way will be to wrap each separate function on custom_module into Prefect Task, which allows to run them in parallel.

Ahhh that clears things for me as well. Yes the LocalDaskExecutor will propagate logs better than multiprocessing