Allow disabling sending flow run logs on a per flow run basis via PREFECT__CLOUD__SEND_FLOW_RUN_LOGS environment variable

[marvin-robot]

Opened from the Prefect Public Slack Community

david.beck: Hi Prefect Team! I'm looking to disable cloud logging since our implementation will be routing our logs via k8s directly to splunk. I tried setting an env value of PREFECT__CLOUD__SEND_FLOW_RUN_LOGS=False in our global container config which I see being recognized by Prefect Cloud, however there are logs still populating for the run (see attached). Do I perhaps have the wrong value defined or could I be missing something?

kevin701: That looks right so confused why it’s behaving like that. I can ask the team but you can also try the --no-cloud-logs on the agent when you start per https://docs.prefect.io/core/idioms/logging.html#logging-with-a-backend|this

david.beck: <@U01QEJ9PP53> okay I thought I set that up correctly. Please let me know what the team says. I have to look into our CI/CD process to see if/how we can include that run flag. We are submitting scheduled flow runs, so I'd have to be sure that they are receiving the --no-cloud-logs flag

michael054: Hey David, the agent overrides this environment variable value with its own setting.

michael054: I’d accept a pull request to change this behavior to allow override as you’ve done. This isn’t intended behavior.

anna: <@ULVA73B9P> open "Allow disabling sending flow run logs on a per flow run basis via PREFECT__CLOUD__SEND_FLOW_RUN_LOGS environment variable"

Original thread can be found here.

[zanieb]

The agents override this variable e.g.

        # 1. Logging level from config
        # Default to the config logging level, allowing it to be overriden
        # by later config soruces
        env.update({"PREFECT__LOGGING__LEVEL": config.logging.level})

        # 2. Values set on the agent via `--env`
        env.update(self.env_vars)

        # 3. Values set on a DockerRun RunConfig (if present)
        if run_config is not None and run_config.env is not None:
            env.update(run_config.env)

        # 4. Non-overrideable required env vars
        env.update(
            {
                "PREFECT__BACKEND": config.backend,
                "PREFECT__CLOUD__API_KEY": self.flow_run_api_key or "",
                "PREFECT__CLOUD__TENANT_ID": (
                    # A tenant id is only required when authenticating
                    self.client.tenant_id
                    if self.flow_run_api_key
                    else ""
                ),
                "PREFECT__CLOUD__AGENT__LABELS": str(self.labels),
                "PREFECT__CLOUD__SEND_FLOW_RUN_LOGS": str(self.log_to_cloud).lower(),
                "PREFECT__CONTEXT__FLOW_RUN_ID": flow_run.id,  # type: ignore
                "PREFECT__CONTEXT__FLOW_ID": flow_run.flow.id,  # type: ignore
                "PREFECT__CONTEXT__IMAGE": image,
                "PREFECT__CLOUD__USE_LOCAL_SECRETS": "false",
                "PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudFlowRunner",
                "PREFECT__ENGINE__TASK_RUNNER__DEFAULT_CLASS": "prefect.engine.cloud.CloudTaskRunner",
                # Backwards compatibility variable for containers on Prefect <0.15.0
                "PREFECT__LOGGING__LOG_TO_CLOUD": str(self.log_to_cloud).lower(),
                # Backwards compatibility variable for containers on Prefect <1.0.0
                "PREFECT__CLOUD__AUTH_TOKEN": self.flow_run_api_key or "",
            }
        )

But we should allow the run config to override it to False even if it's True on the agent. I'm hesitant to always allow the run config to have precedence since an agent with --no-cloud-logs should probably force it to be disabled for all flow runs, but I'm open to an argument to allow it.

[anna-geller]

Currently it's only possible to disable sending flow run logs to Cloud by setting that on the agent using the --no-cloud-logs flag:

prefect agent local start --no-cloud-logs --label xxx

It should be possible to overwrite that setting on a per-flow-run basis using:

from prefect import task, Flow
from prefect.storage.local import Local
from prefect.run_configs import UniversalRun


@task(log_stdout=True)
def hello_world():
    print("hello world")


with Flow(
    "local_storage_universal_run_explicit",
    storage=Local(),
    run_config=UniversalRun(
        labels=["xxx"], env={"PREFECT__CLOUD__SEND_FLOW_RUN_LOGS": False}
    ),
) as flow:
    hw = hello_world()

Currently, the above still sends flow run logs as long as the agent is not started with the --no-cloud-logs flag

[david-beck-rel]

I will be watching this issue. Please let me know when this will be added to the roadmap @anna-geller @madkinsz .