https://prefect.io logo
Powered by
Title
r

Raúl Mansilla

05/25/2021, 4:38 PM
Hello, I´m testing with codecommit storage and I have the next error: 
Failed to retrieve task state with error: ClientError([{'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 5}], 'path': ['get_or_create_task_run_info'], 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: '}}}])
Traceback (most recent call last):
File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/engine/cloud/task_runner.py", line 154, in initialize_run
task_run_info = self.client.get_task_run_info(
File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/client/client.py", line 1399, in get_task_run_info
result = self.graphql(mutation)  # type: Any
File "/home/ubuntu/.local/lib/python3.8/site-packages/prefect/client/client.py", line 319, in graphql
raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'message': 'Expected type UUID!, found ""; Could not parse UUID: ', 'locations': [{'line': 2, 'column': 5}], 'path': ['get_or_create_task_run_info'], 'extensions': {'code': 'INTERNAL_SERVER_ERROR', 'exception': {'message': 'Expected type UUID!, found ""; Could not parse UUID: '}}}
This only happen when I try to run the flow: 
from prefect import task, Flow, Parameter
from prefect.executors import LocalExecutor
from prefect.run_configs import LocalRun
from prefect.storage import CodeCommit
@task()
def say_hello(name):
print("Hello, {}!".format(name))
with Flow("flow",run_config=LocalRun(),storage=CodeCommit(repo="prefect_flows", path="flows/hello_world.py", commit = "master"),executor=LocalExecutor()) as flow:
name = Parameter('name')
say_hello(name)
flow.run(name='world') # "Hello, world!"
flow.run(name='Marvin') # "Hello, Marvin!"
Does anyone has any clue?
k

Kevin Kho

05/25/2021, 4:39 PM
Hi @Raúl Mansilla! How big is your flow?
z

Zanie

05/25/2021, 4:40 PM
Hey @Raúl Mansilla -- it looks like you're encountering a bug we're actively trying to troubleshoot. This is an issue where the task run id is missing from the context during execution. Some more information about your flow/execution environment will help us narrow down the source of this. Are you using Prefect Server or Cloud?
r

Raúl Mansilla

05/25/2021, 4:41 PM
Im using prefect server @Zanie
z

Zanie

05/25/2021, 4:47 PM
So I've tested your flow locally with 
from prefect import task, Flow, Parameter
from prefect.executors import LocalExecutor
from prefect.run_configs import LocalRun
from prefect.storage import Local


@task()
def say_hello(name):
    print("Hello, {}!".format(name))


with Flow(
    "flow",
    run_config=LocalRun(),
    storage=Local(),
    executor=LocalExecutor(),
) as flow:
    name = Parameter("name")
    say_hello(name)

if __name__ == "__main__":
    flow.run(name="world")  # "Hello, world!"
    flow.run(name="Marvin")  # "Hello, Marvin!"
Running 
python flow.py
runs the flow using 
flow.run(...)
Then, I registered your flow with the CLI 
prefect register -p ~/flow.py --project default
I can then create a flow run from the UI successfully (as long as I provide a parameter)
r

Raúl Mansilla

05/25/2021, 4:49 PM
Yes me too, but when I try to get it from CodeCommit
the error comes
z

Zanie

05/25/2021, 4:49 PM
So I'm pretty sure your error is coming from loading the flow as a script
We pull the flow file from CodeCommit then execute it to retrieve the 
Flow
object; when we do this 
flow.run()
is called because you have it in the script
It's then attempting to run the flow multiple times while setting up for a flow run
The 
if __name__ == "__main__":
check should protect against that
(that will only evaluate to 
True
when called directly with 
python script.py
)
r

Raúl Mansilla

05/25/2021, 4:52 PM
I need to put this in the codecommit script right?
z

Zanie

05/25/2021, 4:52 PM
Yep
r

Raúl Mansilla

05/25/2021, 4:55 PM
do you need to check my register
command?
what do I have to write afer if name == “__main__“:
z

Zanie

05/25/2021, 4:59 PM
Your flow can look exactly like mine but instead of 
storage=Local()
you can define your CodeCommit storage
r

Raúl Mansilla

05/25/2021, 4:59 PM
the flow.run() ? … I´m a little stuck
ok
oh yeah, I didn´t see the end of what you´ve wrote!
z

Zanie

05/25/2021, 5:01 PM
You can leave 
flow.run()
out of your script entirely once you're using the backend to orchestrate your flow instead of running it locally; but if you leave it in you need that 
if ...
block
r

Raúl Mansilla

05/25/2021, 5:03 PM
Ok, I´m getting the same error after the changes
z

Zanie

05/25/2021, 5:04 PM
Can you share your entire flow again? You'll want to have done all of the following: • Updated your local flow • Run 
prefect register
• Commit your new flow to CodeCommit @ master • Run your flow from the UI
r

Raúl Mansilla

05/25/2021, 5:05 PM
yes
the flow 
from prefect import task, Flow, Parameter
from prefect.executors import LocalExecutor
from prefect.run_configs import LocalRun
from prefect.storage import CodeCommit
@task()
def say_hello(_name_):
print("Hello, {}!".format(_name_))
with Flow("flow",_run_config_=LocalRun(),_storage_=CodeCommit(_repo_="prefect_flows", _path_="flows/hello_world.py", _commit_ = "master"),_executor_=LocalExecutor()) as flow:
name = Parameter('name')
say_hello(name)
if __name__ == "__main__":
flow.run(_name_='world') # "Hello, world!"
flow.run(_name_='Marvin') # "Hello, Marvin!"
I’ve updated and commit it to master
register the flow 
prefect register --project raulTesting -p codecommit_test.py -n "flow" -l "aws_test" --force
this is the file I use to register the flow: 
import prefect
from prefect import Flow
from prefect.storage import CodeCommit
flow = Flow("flow")
flow.storage = CodeCommit(repo="prefect_flows", path="flows/hello_world.py", commit = "master")
flow.register(project_name='raulTesting', labels=["aws_test"])
z

Zanie

05/25/2021, 5:16 PM
Are you using both that registration file and the 
prefect register
command? You'll want to just choose one of those methods
And in your registration file there you're defining an empty 
Flow("flow")
object with no tasks; you'd want to import it from your flow file so 
from flows.hello_world import flow
r

Raúl Mansilla

05/25/2021, 5:17 PM
sorry, codecommit_test.py is the code
I saw you after register command
the prefect register command is for registering the file with the code that points to codecommit
how should I do it then? Myabe I´m doing something wrong when registering the flow
z

Zanie

05/25/2021, 5:24 PM
Here's a really minimal example
https://github.com/madkinsz/example-git-flow/blob/master/flow.py
You'll want the file you register to be identical to the file in your repo
r

Raúl Mansilla

05/25/2021, 5:25 PM
ah ok, maybe the Flow() line is not necessary
z

Zanie

05/25/2021, 5:25 PM
With that line, you are defining an empty flow as your flow rather than the one with your tasks
You should be able to use the code from the repo I just linked and switch to code commit instead
Here's another example https://github.com/madkinsz/example-git-flow/blob/self-defined-flow/flow.py
r

Raúl Mansilla

05/25/2021, 5:36 PM
ok, let my check the examples out…
so the file I register should look like the file that is on codecommit…
z

Zanie

05/25/2021, 5:38 PM
Yep!
In the future, we'll support more dynamic registration patterns but currently when you register your flow we actually build the task DAG from the flow object in the script and store that metadata in the backend.
Typically, people using CodeCommit/GitHub storage are registering their flows in CI so they are always identical
r

Raúl Mansilla

05/25/2021, 5:41 PM
the CI is the next step…
so now it seems to be working somehow
codecommit file: 
from prefect import task, Flow, Parameter
from prefect.executors import LocalExecutor
from prefect.run_configs import LocalRun
from prefect.storage import CodeCommit
@task()
def say_hello(name):
print("Hello, {}!".format(name))
with Flow("flow",run_config=LocalRun(),storage=CodeCommit(repo="prefect_flows", path="flows/hello_world.py", commit = "master"),executor=LocalExecutor()) as flow:
name = Parameter('name')
say_hello(name)
if __name__ == "__main__":
flow.run(name='world') # "Hello, world!"
flow.run(name='Marvin') # "Hello, Marvin!"
the file that I´m registering manually in the server: 
from prefect import task, Flow, Parameter
from prefect.executors import LocalExecutor
from prefect.run_configs import LocalRun
from prefect.storage import CodeCommit
@task(log_stdout=True)
def say_hello(name):
print("Hello, {}!".format(name))
with Flow("flow",run_config=LocalRun(),storage=CodeCommit(repo="prefect_flows", path="flows/hello_world.py", commit = "master"),executor=LocalExecutor()) as flow:
name = Parameter('name')
say_hello(name)
if __name__ == "__main__":
flow.run(name='world') # "Hello, world!"
flow.run(name='Marvin') # "Hello, Marvin!"
z

Zanie

05/25/2021, 5:44 PM
Yep!
r

Raúl Mansilla

05/25/2021, 5:44 PM
the command for register it: 
prefect register --project raulTesting -p codecommit_test.py -l "aws_test" --force
Ok, I thought that the tasks and flow should only be in the codecommit file while in the “register file” should be only the “pointer” to the codecommit file
so in CI you copy the commited file into the server for register it
and so they´re the same
z

Zanie

05/25/2021, 5:47 PM
When you're setting storage for a flow to a code repository you are saying: "This flow file also exists at this location so it can be retrieved later"
Typically you'd register flows on merge to your main branch (no copying to be done) e.g. https://github.com/jcrist/prefect-github-example/blob/master/.github/workflows/ci.yaml#L41
r

Raúl Mansilla

05/25/2021, 6:06 PM
thanks a lot @Zanie I will try to undesrtand it correctly and put it into a CI workflow
#prefect-serverJoin Slack
Powered by