Hi I m seeing a very weird error in my workflow see thread P Flyte #flyte-support

Hi, I'm seeing a very weird error in my workflow (...

salmon-stone-89503

04/20/2023, 12:31 PM

Hi, I'm seeing a very weird error in my workflow (see thread). Propeller is complaining that there is no input to a node in the workflow. When I look into the gcs bucket I see this in the

<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/>

path:

Copy code

data
|  0/
|  | start-node/
|  |    |  0/
|  |    |  | outputs.pb
|  
|  inputs.pb

i.e. there's no

n0/

directory under

data/0/

. When I look into

data/0/start-node/0/outputs.pb

it looks to be the correct input to the entire workflow. I'm running

flytekit 1.5.0

and deployed from helm

flyte-core:1.5.0

. I tried reverting to

1.4.0

in the core helm chart but it didn't help. Any idea why this is happening?

salmon-stone-89503

04/20/2023, 2:42 PM

Copy code

[
  {
    "json": {
      "exec_id": "mye-exec-id",
      "node": "n1/n0",
      "ns": "myns",
      "res_ver": "445663335",
      "routine": "worker-7",
      "wf": "myproject:myns:mywf.mytask"
    },
    "level": "error",
    "msg": "Failed to read from the raw store [<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>] Error: not found",
    "ts": "2023-04-20T12:13:39Z"
  },
  {
    "json": {
      "exec_id": "myexecid",
      "node": "n1/n0",
      "ns": "myns",
      "res_ver": "445663335",
      "routine": "worker-7",
      "wf": "myproject:myns:mywf.mytask"
    },
    "level": "error",
    "msg": "Catalog Failure: release reservation failed. err: failed to read inputs when trying to query catalog: [READ_FAILED] failed to read data from dataDir [<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>]., caused by: path:<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>: not found",
    "ts": "2023-04-20T12:13:39Z"
  },
  {
    "json": {
      "exec_id": "myexecid",
      "node": "n1/n0",
      "ns": "myns",
      "res_ver": "445663335",
      "routine": "worker-7",
      "wf": "myproject:myns:mywf.mytask"
    },
    "level": "error",
    "msg": "Failed to finalize Dynamic Nodes Parent.",
    "ts": "2023-04-20T12:13:39Z"
  },
  {
    "json": {
      "exec_id": "myexecid",
      "node": "n1",
      "ns": "myns",
      "res_ver": "445663335",
      "routine": "worker-7",
      "wf": "myproject:myns:mywf.mytask"
    },
    "level": "error",
    "msg": "failed Execute for node. Error: 0: failed at Node[n0]. CatalogCallFailed: failed to release reservation, caused by: failed to read inputs when trying to query catalog: [READ_FAILED] failed to read data from dataDir [<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>]., caused by: path:<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>: not found",
    "ts": "2023-04-20T12:13:39Z"
  },
  {
    "json": {
      "exec_id": "myexecid",
      "ns": "myns",
      "res_ver": "445663335",
      "routine": "worker-7",
      "wf": "myproject:myns:mywf.mytask"
    },
    "level": "warning",
    "msg": "Error in handling running workflow [0: failed at Node[n0]. CatalogCallFailed: failed to release reservation, caused by: failed to read inputs when trying to query catalog: [READ_FAILED] failed to read data from dataDir [<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>]., caused by: path:<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>: not found]",
    "ts": "2023-04-20T12:13:39Z"
  },
  {
    "json": {
      "exec_id": "myexecid",
      "ns": "myns",
      "res_ver": "445663335",
      "routine": "worker-7",
      "wf": "myproject:myns:mywf.mytask"
    },
    "level": "error",
    "msg": "Error when trying to reconcile workflow. Error [0: failed at Node[n0]. CatalogCallFailed: failed to release reservation, caused by: failed to read inputs when trying to query catalog: [READ_FAILED] failed to read data from dataDir [<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>]., caused by: path:<gs://mybucket/metadata/propeller/myproject-myns-myexecid/n1/data/0/n0/inputs.pb>: not found]. Error Type[errors.ErrorCollection]",
    "ts": "2023-04-20T12:13:39Z"
  }
]

melodic-magician-71351

04/20/2023, 2:49 PM

We've done some more debugging, including reverting the sql database to a previous state, and reverting flyte to 1.4.x. We were then seeing other errors in the propeller and datacatalog which suggested that the offending workflows were still present. To my knowledge the only other place there is state is in k8s etc for the state of the workflow CRD resource. We're now in the process of taking down all the workflows left after removing the flyte deployment (which is difficult because we need to edit the resource definition to remove the

flyte-finalizer

before deletion), and then restoring flyte to see if that fixes it. Has anyone seen these migration issues before? Hunch is maybe there were some workflows running when we did the 1.4.0 -> 1.5.0 upgrade and they got propeller in a corrupted state, but I'm not sure why that should cause new workflows to break in the way ena described above.

hallowed-mouse-14616

04/20/2023, 3:54 PM

@salmon-stone-89503 is the failure ^^ reproducible? I think there is a bug here. It looks like we call ReleaseCatalogReservation on Finalize, which is called when aborting a Node. It looks like it's possible to abort a Node in the NotYetStarted phase. So the

n0/inputs.pb

is not yet created, and Flyte attempts to abort the Node, then it will fail. Could you file an issue on this? I will plan on submitting a fix quickly!

hallowed-mouse-14616

04/20/2023, 3:55 PM

However, this doesn't explain some of the larger issues it seems your seeing. @melodic-magician-71351 can you elaborate on this? During the upgrade from 1.4 -> 1.5 did you have two flytepropeller instances running simultaneously? This might explain the state issues you mentioned.

salmon-stone-89503

04/20/2023, 4:48 PM

Yeah I'm not entirely sure what happened either. Let me unroll the sequence of events properly now that we've looked more into it. 1. I deployed 1.5.0 from flyte-core on Monday 2. I reverted to 1.4.0 yesterday 3. Workflows started failing with the errors I sent in the original message 4. We saw errors in the propeller saying that a dataset can't be created because it already exists when writing outputs of a workflow. If I ran a workflow with cache override and different inputs I still saw the error 5. We brought down flyte and rolled back the db to a backup 6. There were dangling executions which we had to delete in a hacky way (see Eli's message above) 7. We redeployed flyte 1.5.0

salmon-stone-89503

04/21/2023, 10:09 AM

Looks like the same thing is happening again. This is the workflow that created the issue. I'm seeing the same logs as the ones I saw yesterday. I'm really lost as to why that would happen. I'll try to reproduce it with a mock workflow within a workflow. Not sure what else to do.

salmon-stone-89503

04/21/2023, 11:56 AM

I managed to reproduce with this:

Copy code

import time
import flytekit

from flytekit import workflow, task
from flytekit.types.file import FlyteFile
from pathlib import Path
from typing import Optional


@task(cache=True, cache_serialize=True, cache_version='0.0.0')
def t1() -> FlyteFile:
    out = Path(flytekit.current_context().working_directory) / 'out.txt'
    out.write_text('Hi from t1')
    return FlyteFile(path=str(out))


@task(cache=True, cache_serialize=True, cache_version='0.0.0')
def t2(t1_: FlyteFile, optional: Optional[int] = None) -> FlyteFile:
    out = Path(flytekit.current_context().working_directory) / 'out.txt'
    with open(t1_, 'r') as f:
        out.write_text(f.read() + f'\nHi from t2, optional={optional}')
    return FlyteFile(path=str(out))


@workflow
def inner_wf(t1_: FlyteFile, optional: Optional[int] = None) -> FlyteFile:
    return t2(t1_=t2(t1_=t1_, optional=optional), optional=optional)


@workflow
def wf() -> FlyteFile:
    return inner_wf(t1_=t1())

So the issue seems to be with a workflow calling a subworkflow that has an optional input.

salmon-stone-89503

04/21/2023, 11:57 AM

Here's the graph. Without cache I get this error:

failed at Node[n0]. BindingResolutionError: Error binding Var [wf].[optional], caused by: failed at Node[start-node]. OutputsNotFoundError: Failed to find [start-node].[optional]

salmon-stone-89503

04/21/2023, 12:08 PM

When I add caching the error is the same as the original one (complaining about missing

../inputs.pb

)

hallowed-mouse-14616

04/21/2023, 12:19 PM

Very sorry about these issues. It looks like there may be multiple things happening here, almost certainly bugs. The issue with caching (that you mentioned yesterday) is a high priority bug - basically Flyte is attempting to release the cache reservation when it shouldn't be. However, this new issue is probably the root cause - we will need to do some internal investigation as to why the input / output binding values are not being set correctly.

hallowed-mouse-14616

04/21/2023, 12:22 PM

Now that I'm looking a little closer, the error is saying that the call the

inner_wf

can not find the

optional

parameter. @thankful-minister-83577 can you take a brief look at the workflow ^^^?

salmon-stone-89503

04/21/2023, 12:38 PM

Yes that's the error if I don't request caching but my original error is that an

inputs.pb

file is missing, that one I only see if I add caching @hallowed-mouse-14616

salmon-stone-89503

04/21/2023, 12:45 PM

here's the full error log and the workflow definition that caused it

main.py error.json

hallowed-mouse-14616

04/21/2023, 12:56 PM

Yeah, that cache issue is caused because Flyte is attempted to abort the workflow and that process is failing. The abort is attempted because of the issue in resolving inputs.

hallowed-mouse-14616

04/21/2023, 12:57 PM

I just tested locally and removing the

None

from the

inner_wf

definition fixes this in both cases (caching and non-caching). So in your repro example:

Copy code

import time
import flytekit

from flytekit import workflow, task
from flytekit.types.file import FlyteFile
from pathlib import Path
from typing import Optional


@task(cache=True, cache_serialize=True, cache_version='0.0.0')
def t1() -> FlyteFile:
    out = Path(flytekit.current_context().working_directory) / 'out.txt'
    out.write_text('Hi from t1')
    return FlyteFile(path=str(out))


@task(cache=True, cache_serialize=True, cache_version='0.0.0')
def t2(t1_: FlyteFile, optional: Optional[int] = None) -> FlyteFile:
    out = Path(flytekit.current_context().working_directory) / 'out.txt'
    with open(t1_, 'r') as f:
        out.write_text(f.read() + f'\nHi from t2, optional={optional}')
    return FlyteFile(path=str(out))


@workflow
def inner_wf(t1_: FlyteFile, optional: Optional[int]) -> FlyteFile:
    return t2(t1_=t2(t1_=t1_, optional=optional), optional=optional)


@workflow
def wf() -> FlyteFile:
    return inner_wf(t1_=t1(), optional=None)

hallowed-mouse-14616

04/21/2023, 12:58 PM

I'm not an expert on the flytekit side (we'll get others to look into that), but I'm wondering if there is a bug in supporting

Optional[int] = None

on workflows.

👍 1

salmon-stone-89503

04/21/2023, 1:29 PM

yeah, seems like a reasonable conclusion. thanks

hallowed-mouse-14616

04/21/2023, 2:06 PM

I submitted an issue for the catalog reservation release failure. Like I said, I believe that the optional param on the subworkflow is the root cause, but this is still an issue that we need to resolve.

🙏 1

449 Views

Open in Slack

Previous Next