I successfully ran a flow locally however when I deploy the

Question

I successfully ran a flow locally however when I deploy the flow to prefect cloud I got this warning message at the `healthcheck` step ``` opt prefect healthcheck py 149 UserWarning Task lt Task fetch runnable models gt has retry settings but some upstream dependencies do not have result types See <https docs prefect io core concepts results html> for more details result check flows ``` please see the thread for more details

Hui Zheng · Answer

`fetch runnable models` is the second task in the flow The flow up to this task is defined as below ```with Flow Data Platform Flow as flow secrets inside prefect context secrets come from locally ~ prefect flow config toml remotely Prefect Cloud Secret Mgmt dash dbt cloud tkn = prefect context secrets DBT CLOUD API PERSONAL ACCESS TOKEN dbt cloud api = API dbt cloud tkn current stamp = get fixed stamp now fs models = fetch runnable models current stamp ```

Hui Zheng · Answer

the first task is defined as below ``` task log stdout=True def get fixed stamp now gt datetime print Flow w Target Release in env format getenv SCHEDULE TYPE getenv TARGET RELEASE VERSION getenv ENVIRONMENT default= stage noqa round down to the minute stamp = pytz utc localize datetime utcnow replace second=0 microsecond=0 return stamp```

Hui Zheng · Answer

at flow global level I set ``` PREFECT TASKS DEFAULTS MAX RETRIES 2 PREFECT TASKS DEFAULTS RETRY DELAY 10 PREFECT TASKS DEFAULTS CHECKPOINT True ```

Chris White · Answer

Hi < Hui Zheng> this isn t necessarily a problem depending on what scenarios you hope to cover with this Flow Assuming you are using `Docker` storage this Flow is not configured to store your task outputs anywhere This means that if `fetch runnable models` were to fail and the process job were to unexpectedly die before the retry could proceed you would not be able to successfully retry the task because it would not be able to recreate its upstream data dependency In normal operating scenarios though the retry will occur within the same process and you should be fine it s the edge case when the process job unexpectedly dies that this warning is highlighting

Hui Zheng · Answer

Thank you Chris yes we are using `Docker` storage Because this flow is high critical I do hope to recover from the worst edge cases such as job were to unexpectedly die How do I store your task outputs outside the docker container I already set the flow result with the following settings to build out the container ```flow result = GCSResult bucket= ``` it seems not cover everything given to the warning message

Chris White · Answer

Hmmm are you setting that result prior to registering the flow building its storage

Hui Zheng · Answer

yes I set the result prior to registering the flow building its storage step It has been done like that and I never see those warning message before There is one change recently though around the time these warning message appears I declared some variables in my `flow py` file at the global level outside the scope of tasks and flow and then use those variables in some tasks Are those variables the cause of the warning ``` schedule type = getenv SCHEDULE TYPE target release u = getenv TARGET RELEASE VERSION replace environment = getenv ENVIRONMENT default= stage fs collection name = join environment target release u schedule type ```

Chris White · Answer

No that wouldn t do it the check is incredibly simple <https github com PrefectHQ prefect blob master src prefect environments storage healthcheck py L67> You can see that if the flow has a result the check is skipped so for some reason your Flow object within your Docker image does not have a result attribute

Hui Zheng · Answer

Could that related to our another conversation <https prefect community slack com archives CL09KU1K7 p1603838678211000> my t`ask B` in this case is the `SkipNoneDbtShellTask` task below You suggested that the `task B` could check if the provided value is `None` and `SKIP` if so However actually when I implement it I did it a bit differently I say ``task B` that is `SkipNoneDbtShellTask` return None if the provided value is `None` ```class SkipNoneDbtShellTask DbtShellTask defaults from attrs command env dbt kwargs def run self command str = None env dict = None dbt kwargs dict = None kwargs Any gt str if command is None return None else return super SkipNoneDbtShellTask self run command=command kwargs ```