https://prefect.io
Join Slack
Hey folks, I'm encountering an issue trying to get a flow to rerun itself when a condition is met. H...
# ask-community
c
Hey folks, I'm encountering an issue trying to get a flow to rerun itself when a condition is met. Here's the code for my flow:
Copy code
rerun_flow = StartFlowRun(
        flow_name="blast_import_flow",
        project_name="python_imports"
        )   
spawn_time_series_import = StartFlowRun(
        project_name="python_imports", 
        flow_name="blast_metric_series_flow",
        wait=True)

with Flow("blast_import_flow", 
        schedule=daily_schedule, 
        state_handlers=[cloud_only_slack_handler]
       ) as flow:

    data_object = get_blast_batch()
    check_api_result = check_api(data_object=data_object)
    transformed_data_objects = transform(data_object=check_api_result)
    imported = import_to_database(data_object_list=transformed_data_objects)
    blast_ids, record_length = push_records_to_summary_table(imported)

    # package ids into the right dict format
    packaged_ids = package_into_params.map(blast_id=blast_ids)

    spawn = spawn_time_series_import.map(
            parameters=packaged_ids)
    with case(record_length, 5): 
        rerun_flow()
The ETL portions of the flow are working just fine, and the 
spawn_time_series_import
is correctly creating its tasks. But 
rerun_flow
isn't - I'm seeing this error:
Copy code
Unexpected error: ClientError([{'path': ['user'], 'message': 'field "user" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.user', 'code': 'validation-failed', 'exception': {'message': 'field "user" not found in type: \'query_root\''}}}])
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/runner.py", line 48, in inner
    new_state = method(self, state, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/engine/task_runner.py", line 860, in get_task_run_state
    logger=self.logger,
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/executors.py", line 298, in run_task_with_timeout
    return task.run(*args, **kwargs)  # type: ignore
  File "/usr/local/lib/python3.7/site-packages/prefect/utilities/tasks.py", line 449, in method
    return run_method(self, *args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/prefect/tasks/prefect/flow_run.py", line 172, in run
    run_link = client.get_cloud_url("flow-run", flow_run_id)
  File "/usr/local/lib/python3.7/site-packages/prefect/client/client.py", line 887, in get_cloud_url
    tenant_slug = self.get_default_tenant_slug(as_user=as_user and using_cloud_api)
  File "/usr/local/lib/python3.7/site-packages/prefect/client/client.py", line 920, in get_default_tenant_slug
    res = self.graphql(query)
  File "/usr/local/lib/python3.7/site-packages/prefect/client/client.py", line 318, in graphql
    raise ClientError(result["errors"])
prefect.utilities.exceptions.ClientError: [{'path': ['user'], 'message': 'field "user" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.user', 'code': 'validation-failed', 'exception': {'message': 'field "user" not found in type: \'query_root\''}}}]
I'm not sure what the difference is, aside from the wait. I figure I don't want a bunch of these telescoping and waiting on each other to finish.
z
Hi @Chris Jordan -- are you using Prefect Server?
c
I'm using prefect cloud, registering these with an agent on a server of mine and running from the web gui
z
Could you give me the output of 
prefect diagnostics
locally and from the agent's machine?
c
locally here being the machine I'm developing on?
z
Yep!
c
local:
Copy code
(python-imports-3.7.6) ┌─[cjordan@WMNAPA-BBO02-20] - [~/work/python_imports] - [2021-02-04 05:56:36]
└─[0] <git:(final_blast bad0de3✈) > prefect diagnostics 
{
  "config_overrides": {
    "cloud": {
      "graphql": true
    },
    "context": {
      "secrets": false
    },
    "debug": true,
    "engine": {
      "executor": {
        "dask": {
          "address": true,
          "local_processes": true
        }
      }
    },
    "flows": {
      "defaults": {
        "storage": {
          "default_class": true
        }
      }
    }
  },
  "env_vars": [
    "PREFECT__CLOUD__AGENT__AUTH_TOKEN"
  ],
  "system_information": {
    "platform": "Darwin-19.6.0-x86_64-i386-64bit",
    "prefect_backend": "server",
    "prefect_version": "0.13.18",
    "python_version": "3.7.6"
  }
}
(python-imports-3.7.6) ┌─[cjordan@WMNAPA-BBO02-20] - [~/work/python_imports] - [2021-02-04 05:56:41]
└─[0] <git:(final_blast bad0de3✈) >
z
So your backend there is Server, not Cloud and you've overridden the API url?
c
yes the backend there is Server, but I'm only developing there
for deployment I push to the server that I'm running agent on
and register there
z
I see 👍
c
here's the diagnostics from that server
Copy code
cjordan@livo:/data/cjordan/python_imports$ prefect diagnostics 
{
  "config_overrides": {},
  "env_vars": [],
  "system_information": {
    "platform": "Linux-4.4.0-1100-aws-x86_64-with-debian-stretch-sid",
    "prefect_backend": "cloud",
    "prefect_version": "0.13.18",
    "python_version": "3.7.6"
  }
}
z
Great thanks! If you're willing to try the newest version of Prefect that'd be helpful, I'll be looking into some sort of breaking change here.
c
that seems plausible. I think there was a reason I didn't upgrade, though I can't recall it now, and I don't want to spend Thursday evening chasing that down
could you talk me through a guess of what's going on, and I can try to debug things a little myself?
z
It's failing here:
Copy code
def get_default_tenant_slug(self, as_user: bool = True) -> str:
        """
        Get the default tenant slug for the currently authenticated user

        Args:
            - as_user (bool, optional): whether this query is being made from a USER scoped token;
                defaults to `True`. Only used internally for queries made from RUNNERs

        Returns:
            - str: the slug of the current default tenant for this user
        """
        if as_user:
            query = {
                "query": {"user": {"default_membership": {"tenant": "slug"}}}
            }  # type: dict
        else:
            query = {"query": {"tenant": {"slug"}}}

        res = self.graphql(query)

        if as_user:
            user = res.get("data").user[0]
            slug = user.default_membership.tenant.slug
        else:
            slug = res.get("data").tenant[0].slug
        return slug
c
if my only option is to upgrade then I'm willing to work through that, but if I can skate by without for now I'd prefer that
z
Upgrading would just help nail down if it's a versioning issue
So the error is that there isn't a "user" field which generally indicates that the database schema isn't matching what the client expects
Usually this is a Prefect Server / Core version mismatch but we generally are careful not to break things like this with Cloud
c
got it, that makes sense
z
Ah hey this may be a token issue
c
I can take a look at upgrading, but the soonest I can devote nontrivial time to it is Tuesday
z
What scope of token are you using?
RUNNER / USER ?
c
one sec looking this up...
z
Which agent type are you using as well?
c
the token I'm using is at least a few versions old, and the note I saved it in only calls it a 
prefect agent token
looks like I'm starting a docker agent
invocation is here
Copy code
prefect agent start \
        --volume "${WORKDIR}/.config.toml:/opt/prefect/config.toml" \
        -t "${token}" \
        docker \
        --show_flow_logs=True
z
Could you try generating a new USER scoped token and using that instead?
c
I'll give that a shot, give me a minute
(lastpass is annoying to sync)
also let me double check here, cause the woding isn't consistent across all the documentation - from https://docs.prefect.io/orchestration/concepts/api.html#authentication, 
USER-scoped API tokens function as personal access tokens.
You seem to use the words interchangably
these are the same thing right?
z
Yeah this is in the middle of a refactor because it's really confusing as is... one sec
c
They're replacing a bridge next to my workplace, and for some time, every time I drive over the old one I think to myself "man this bridge is in really bad shape they should really do something about it" .... as the construction crews are working 30ft away
z
😄
What you want is a token that's not RUNNER scoped 😄
It turns out this is a known issue that was fixed in December (https://github.com/PrefectHQ/prefect/commit/810faa5039422ac2ae5e26de29902fd424c6f2b0)
Since we're not passing 
as_user=False
it's trying to access something that RUNNER tokens don't have permissions to see.
c
Copy code
(python-imports-3.7.6) cjordan@livo:/data/cjordan/python_imports$ screen -r 

    return self.main(*args, **kwargs)                                                           File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/click/core.py", line 782, in main                                                       rv = self.invoke(ctx)                                                                       File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/click/core.py", line 1259, in invoke                                                    return _process_result(sub_ctx.command.invoke(sub_ctx))                                     File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/click/core.py", line 1259, in invoke                                                    return _process_result(sub_ctx.command.invoke(sub_ctx))                                     File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/click/core.py", line 1066, in invoke                                                    return ctx.invoke(self.callback, **ctx.params)                                              File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/click/core.py", line 610, in invoke                                                     return callback(*args, **kwargs)                                                            File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/click/decorators.py", line 21, in new_func
    return f(get_current_context(), *args, **kwargs)
  File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/prefect/cli/agent.py", line 666, in start
    docker_interface=not no_docker_interface,
  File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/prefect/agent/agent.py", line 219, in start
    self._verify_token(self.client.get_auth_token())
  File "/home/users/cjordan/.pyenv/versions/3.7.6/envs/python-imports-3.7.6/lib/python3.7/site-packages/prefect/agent/agent.py", line 178, in _verify_token
    raise AuthorizationError("Provided token does not have a RUNNER scope.")
prefect.utilities.exceptions.AuthorizationError: Provided token does not have a RUNNER scope.
cjordan@livo:/data/cjordan/python_imports$
so it seems like a version upgrade is probably in order
Like I said I can't get to that for a couple of days
z
Aghh. I shall ask someone that understands our tokens. I think you just need to grab a TENANT token using your PAT/USER token
c
got it, thanks
this isn't a pressing issue - if someone gets back to me sometime Monday that would be fantastic
and in the meantime I can spend some time Tuesday working through upgrades
z
You can make a tenant token direectly at https://cloud.prefect.io/team/tokens
c
lemme see if I've got one already
looks like no
one sec, pushing it through the machine...
I'm getting the same runner error. I double checked that the token I just generated has the scope TENANT and that the token my script is pulling out of lastpass is the one that I just generated
er the error is
Provided token does not have a RUNNER scope.
z
Annoying, sorry about that.
It looks like it won't let you use a higher privilege token.
I think you'll either need to patch that line yourself or upgrade -- sorry it's broken on older versions 😕
c
can't fault you for not having fixed it fast enough for me to get the upgrade last time I did an upgrade
but this has been really helpful
z
👍 0.14 has all sorts of fun stuff in it
c
I'm glad to know that there is a fix, and I'll have the time to apply it soon
39 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by