https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have an issue getting logs to appea...
# ask-marvin
o
@Marvin I have an issue getting logs to appear in the prefect UI (i self host on docker) when providing the runtime logger for prefect and using logger.info(...)
m
✅ thought for 143.1 seconds
Got it — let’s get your logs showing up in the UI. In Prefect 3.x, only logs that go through Prefect’s API log handler are displayed in the UI. That happens when: - You use 
get_run_logger()
inside a flow/task run, or - You tell Prefect to capture additional Python loggers via 
PREFECT_LOGGING_EXTRA_LOGGERS
If you call 
logging.getLogger(...).info(...)
directly, those logs won’t go to the UI unless you register that logger as an “extra logger.” Also, 
get_run_logger()
must be called within an active flow/task run; calling it too early (e.g., module import time) won’t associate logs to a run and they’ll be dropped. Two reliable ways to fix it 1) Prefer 
get_run_logger()
inside your flow/task
Copy code
from prefect import flow, task
from prefect.logging import get_run_logger

@task
def my_task():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello from task")

@flow
def my_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Hello from flow")
    my_task()

if __name__ == "__main__":
    my_flow()
2) If you want to keep using your own logger name, register it - Set this in the environment of your worker/flow container(s):
Copy code
PREFECT_LOGGING_EXTRA_LOGGERS=myapp,uvicorn,any.other.logger.names
- Keep using 
logger = logging.getLogger("myapp"); <http://logger.info|logger.info>("...")
— those will go to the UI. Key settings to check in your Docker/self-hosted setup - Ensure your workers/flow containers can reach your API: - 
PREFECT_API_URL
must point to your server (e.g., 
<http://prefect-server:4200/api>
depending on your network) - Ensure logs are being sent to the API: - 
PREFECT_LOGGING_TO_API_ENABLED=true
(default is true, but best to be explicit) - Make sure your log level allows INFO: - 
PREFECT_LOGGING_ROOT_LEVEL=INFO
(or 
DEBUG
for troubleshooting) - Give it a moment: - Logs are batched (default ~2s). They should appear shortly after being emitted. Troubleshooting checklist - Are you calling 
get_run_logger()
inside the flow/task (not at import time)? - If using standard Python loggers, did you add their names to 
PREFECT_LOGGING_EXTRA_LOGGERS
? - Do logs appear in the worker/container stdout, but not the UI? If so, check API reachability and 
PREFECT_API_URL
. - Turn on debug logging and look for API log handler messages:
Copy code
PREFECT_LOGGING_LEVEL=DEBUG
- If you use multiprocessing/threads, each process/thread that logs needs its own 
get_run_logger()
call within the run context. If you can share: - A minimal snippet of your flow/task showing how you obtain/use the logger - Which work pool/worker type you’re using (Docker, Process, Kubernetes) - The relevant env vars from your worker container (
PREFECT_API_URL
, 
PREFECT_LOGGING_*
) - Whether you see the logs in the container’s stdout I can pinpoint the exact issue quickly. Docs for reference: - Add logging to flows and tasks - Customize logging and capture extra loggers - Settings reference (PREFECT_LOGGING_*)
o
@Marvin if i do @flow() def my_flow(): return non_prefect_task(logger=get_run_logger()) should that work?
m
✅ thought for 34.0 seconds
Short answer: yes — passing 
get_run_logger()
into a non-Prefect function will work, as long as that function uses the provided logger to emit logs during the active flow run. Example:
Copy code
from prefect import flow
from prefect.logging import get_run_logger

def non_prefect_task(logger):
    <http://logger.info|logger.info>("Hello from non-Prefect code")

@flow
def my_flow():
    non_prefect_task(logger=get_run_logger())
Those logs will appear in the UI under the flow run (not under a task run). Important caveats: - The function must actually use the passed 
logger
(not re-create its own via 
logging.getLogger(...)
). - This works reliably in the same process during the flow run. If the function creates threads/subprocesses or defers work until after the flow completes, the logs may not have run context and could be dropped. - If you want task-level association, wrap the call in a 
@task
and get the logger inside the task. - If the function uses 
print()
, enable capture via 
@flow(log_prints=True)
or 
PREFECT_LOGGING_LOG_PRINTS=true
. If you’re not seeing logs with this pattern, let me know: - Does 
non_prefect_task
spawn threads or subprocesses? - Is it using the passed 
logger
for 
info
calls? - What worker type are you using and what are the relevant 
PREFECT_*
env vars? Docs: - Add logging to flows and tasks - Customize/capture extra loggers
o
@Marvin So my setup is The deployed flow: @flow(name='nn_get_intraday_stock_prices') def nn_get_intraday_stock_prices_flow( instrument_ids: list[int] | None = None, n_random: Optional[int] = None, iterations: int = 3, db_type: Literal['sqlite', 'duckdb'] = 'duckdb' ) -> Dict[str, object]: """Main flow that orchestrates chunked ETL with retry logic for failed instruments.""" stocks = fetch_instruments(instrument_ids=instrument_ids, n_random=n_random, db_type=db_type) return price_etl_flow(stocks=stocks, db_type=db_type, intraday=True, iterations=iterations) @flow def price_etl_flow(stocks: pl.DataFrame, intraday: bool, iterations: int=3, **kwargs) -> Dict[str, int | list[int]]: return price_etl( stocks=stocks, intraday=intraday, iterations=iterations, _create_extract_kwarg_list=create_extract_kwarg_list, _extract_data=extract_data, _transform_data=transform_data, _fetch_latest_db_for_load=fetch_latest_db_for_load, _apply_delta_load=apply_delta_load, _load_data=load_data, logger=get_run_logger(), **kwargs ) def price_etl(stocks: pl.DataFrame, intraday: bool = False, iterations: int=3, **kwargs) -> Dict[str, int | list[int]]: return generalized_etl( stocks=stocks, extract_method_name='get_instrument_prices', static_kwargs={'resolution': 'MIN_1', 'period': 'WEEK_1'} if intraday else {}, apply_delta_load=True, iterations=iterations, **kwargs, ) def generalized_etl( stocks: pl.DataFrame, extract_method_name: Literal['get_instrument_prices', 'get_instrument_ownership'] = 'get_instrument_prices', static_kwargs: dict = {}, apply_delta_load: bool = True, raw_base_path: str = None, transformed_base_path: str = None, iterations: int = 3, retry_sleep_seconds: int = 60*3, db_type: Literal['sqlite', 'duckdb'] = 'duckdb', logger: Optional = None, _create_extract_kwarg_list: Callable = _create_extract_kwarg_list, _extract_data: Callable = _extract_data, _transform_data: Callable = _transform_data, _fetch_latest_db_for_load: Callable = _fetch_latest_db_for_load, _apply_delta_load: Callable = _apply_delta_load, _load_data: Callable = _load_data, ) -> Dict[str, int | list[int]]: ''' Generalized ETL wrapper method for Nordnet data collection Parameters: ----------- stocks : DataFrame containing instruments to process method_name : Literal['get_instrument_prices', 'get_instrument_ownership']. Data collector method name static_kwargs : dict. Static parameters for the data collection method. Give {'resolution': 'MIN_1', 'period': 'WEEK_1'} for intraday data for get_instrument_prices apply_delta_load : bool. Whether to apply delta loading logic raw_base_path : Base path for raw data files transformed_base_path : Base path for transformed data files iterations : int. Number of iterations to split the work ''' transform_methods, load_methods, transform_kwargs = _select_transform_and_load(extract_method_name=extract_method_name, static_kwargs=static_kwargs) failed_first_load = [] n = stocks.shape[0] chunk_size = (n + iterations - 1) // iterations for idx in range(iterations): start = idx * chunk_size stop = min((idx + 1) * chunk_size, n) current_stocks = stocks.slice(start, stop - start) print('Iteration', idx + 1, f'of {iterations}') failed_first_load += _generalized_etl( idx=idx, current_stocks=current_stocks, transform_methods=transform_methods, load_methods=load_methods, transform_kwargs=transform_kwargs, raw_base_path=raw_base_path, transformed_base_path=transformed_base_path, extract_method_name=extract_method_name, static_kwargs=static_kwargs, apply_delta_load=apply_delta_load, db_type=db_type, logger=logger, _create_extract_kwarg_list=_create_extract_kwarg_list, _extract_data=_extract_data, _transform_data=_transform_data, _apply_delta_load=_apply_delta_load, _load_data=_load_data, ) summary = {'attempted': n, 'failed_instruments': failed_first_load} missing_pre_transform_col = 'value' if extract_method_name == 'get_instrument_ownership' else 'last' if failed_first_load: statement = f'Retrying failed first load for {len(failed_first_load)} instruments after sleeping {retry_sleep_seconds} seconds...' if logger is not None: logger.info(statement) else: print(statement) i know from the ui and setup that failed_first_load will be a non-empty list, so it should do logger.info(statement). There are also logging inside of _generalized_etl() that should be triggered my worker pool is of docker type, so each deployment run should be in a container. however no threading or subprocesses being started except for that afaik. Not sure entirely what environment variables you are asking me to give
m
✅ thought for 454.0 seconds
Thanks for the details — this helps. The key thing I’m noticing: 
price_etl_flow
is decorated with 
@flow
and is called from another 
@flow
(
nn_get_intraday_stock_prices_flow
). That means 
price_etl_flow
runs as a child flow. Any 
get_run_logger()
created inside 
price_etl_flow
(and passed down to your helpers) will emit logs to the child flow run, not the parent. By default, those logs will not show up in the parent run’s log view unless you enable “show child logs”. A few quick checks/actions: - In the UI, open the parent run (nn_get_intraday_stock_prices_flow) and toggle “Show logs from child runs” (or open the Related runs > the child run for 
price_etl_flow
) and check logs there. - Add a quick sanity check in both flows to confirm where logs land:
Copy code
from prefect import get_run_logger, get_run_context

  @flow(name="nn_get_intraday_stock_prices")
  def nn_get_intraday_stock_prices_flow(...):
      get_run_logger().info(f"Parent flow run id: {get_run_context().flow_run.id}")
      ...
      return price_etl_flow(...)

  @flow
  def price_etl_flow(...):
      get_run_logger().info(f"Child flow run id: {get_run_context().flow_run.id}")
      ...
- In 
generalized_etl
, make sure you’re using the Prefect run logger (which you are passing). If there’s any standard 
print
usage you care about, prefer 
get_run_logger().info(...)
or explicitly patch prints:
Copy code
from prefect.logging.loggers import patch_print
  patch_print()  # optional; routes print to logs
Common reasons logs don’t appear where expected: - Viewing only the parent flow’s logs: child flow logs live on the child run unless you enable “show child logs”. - 
get_run_logger()
called outside a flow/task context: it won’t be associated with a run and won’t go to the UI. In your code you’re calling it inside 
price_etl_flow
, which is good. - Logging level or API logging disabled. In your Docker worker environment, verify you’re not disabling API logging: - Ensure 
PREFECT_LOGGING_TO_API
is not set to 
false
. - (Optional) Set 
PREFECT_LOGGING_LEVEL=INFO
to be explicit. If you want all of these logs to appear on the parent run instead of a separate child run, remove 
@flow
from 
price_etl_flow
so it becomes a plain function called by the parent, or keep it a flow but rely on the UI toggle to include child logs. To help me narrow further, could you confirm: - Are you checking the logs on the child run for 
price_etl_flow
(or with “Show child logs” enabled on the parent)? - Are any logs from inside 
_generalized_etl()
using 
print
instead of a Prefect logger? - Do you have any env vars like 
PREFECT_LOGGING_TO_API
, 
PREFECT_LOGGING_LEVEL
, or custom logging config set on your Docker work pool infrastructure or deployment? If you’re not sure, sharing the env section from your Docker infrastructure block/deployment would help.
37 Views
Open in Slack
PreviousNext
Powered by