Logs disappearing for deferred CloudRunExecuteJobOperator while state is deferred

[MarcusCramer91]

Apache Airflow version

Other Airflow 2/3 version (please specify below)

If "Other Airflow 2/3 version" selected, which one?

3.1.1

What happened?

I am using a custom class that extends CloudRunExecuteJobOperator and overrides the pre_execute and execute functions. Among other things, these functions produce logs.

When running the DAG with that class and deferrable=True those logs briefly appear in the UI. Then they disappear, only to re-appear when the status is no longer deferred.

Sounds similar to this one #28647, but that issue is pretty old...

What you think should happen instead?

I should be seeing all logs all the time, regardless of the tasks status.

Since I am pretty sure this is related to deferred maybe there is also another proper way to see the logs, or a workaround for this?

How to reproduce

Use this in a DAG:

class CustomCloudRunJobOperator(CloudRunExecuteJobOperator):
    def pre_execute(self, context: Context) -> None:
        self.log.info("Some log here")
        super().pre_execute(context)

Operating System

Debian GNU/Linux 12 (bookworm)

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Lee-W]

Hey @MarcusCramer91 , I tried a few ways but have not yet reproduced it. Is it consistently happening, and do you have a way to produce it? Thanks!

[MarcusCramer91]

@Lee-W Thanks for investigating. Happens 100% for me. Logs briefly appear (for like 10s), then they disappear while the task is deferred, and they re-appear once the task has finished. I suspect it has to do with deferred tasks being passed over to the triggerer.

Any particular information regarding my setup that would help debugging?

[Lee-W]

If we could have a minimum Dag (ideally with a minimum custom deferrable operator that does almost nothing but logging), that would help us investigate it faster. But I'll try to create one operator that stays in deferred state longer and see whether I can reproduce it on my end as well. Thanks!

[Lee-W]

@MarcusCramer91, is there any airflow config you feel is suspicious? Or maybe let's check whether the error happens in core or google-provider. Could you please test with a Dag like the following one and see whether the log is still disappearing?

from __future__ import annotations

from datetime import datetime, timedelta

from airflow.providers.standard.triggers.temporal import TimeDeltaTrigger
from airflow.sdk import DAG, BaseOperator, Context


class SimpleDeferrableOperator(BaseOperator):
    def __init__(self, wait_seconds: int = 5, **kwargs):
        super().__init__(**kwargs)
        self.wait_seconds = wait_seconds

    def pre_execute(self, context: Context):
        self.log.info(">>> pre_execute called by self.log")
        super().pre_execute(context)

    def execute(self, context: Context):
        print(">>> Deferring task...")
        self.defer(
            method_name="execute_complete",
            trigger=TimeDeltaTrigger(timedelta(seconds=self.wait_seconds)),
        )

    def execute_complete(self, context: Context, event=None):
        print(">>> execute_complete called, event:", event)
        return "completed"


with DAG(
    dag_id="minimal_deferrable_dag",
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
):
    SimpleDeferrableOperator(
        task_id="wait_then_complete",
        wait_seconds=10,
    )

If so, then we can almost confirm this is an issue from the core. And the next step could be turn on and turn off your airflow config one by one and see what makes this happened