Starting flow instead of running it

[adammrozik]

Hi!

My goal is to while I call flow.run function to receive only some kind of id/state of the flow that has been started, but not actually have to wait for the result. Do I understand correctly that Agent is a thing that can help me solve it?

I have manually deployed kubernetes agent based on the YAML received from prefect kubernetes install. I have only added option to runAsUser 1000, as my cluster does not allow running on root.

I have also Dask Cluster running on the same K8S cluster. Dask cluster itself is working, but I am having trouble with using the agent. After deploying it with kubectl apply, I thought that run_config option would be able to find it:

    executor = DaskExecutor(address=f"tcp://my_dask_address:1234")

    with Flow("Dask Kubernetes Flow", executor=executor) as flow:
        run_command(**kwargs)

    # Setup Executor and run the flow
    flow.run_config = KubernetesRun(cpu_request=2, memory_request="4Gi")
    flow.run()

Is there something I am missing or something more that should be done?

Also, my Kubernetes Agent crashes on its own with the following error:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 465, in _request
    json_resp = response.json()
  File "/usr/local/lib/python3.6/site-packages/requests/models.py", line 900, in json
    return complexjson.loads(self.text, **kwargs)
  File "/usr/local/lib/python3.6/json/__init__.py", line 354, in loads
    return _default_decoder.decode(s)
  File "/usr/local/lib/python3.6/json/decoder.py", line 339, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
  File "/usr/local/lib/python3.6/json/decoder.py", line 357, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/bin/prefect", line 8, in <module>
    sys.exit(cli())
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.6/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/prefect/cli/agent.py", line 288, in start
    start_agent(KubernetesAgent, image_pull_secrets=image_pull_secrets, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/prefect/cli/agent.py", line 109, in start_agent
    agent.start()
  File "/usr/local/lib/python3.6/site-packages/prefect/agent/agent.py", line 224, in start
    self.client.attach_headers({"X-PREFECT-AGENT-ID": self._register_agent()})
  File "/usr/local/lib/python3.6/site-packages/prefect/agent/agent.py", line 193, in _register_agent
    agent_config_id=self.agent_config_id,
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 1701, in register_agent
    tenant_id=self.active_tenant_id,
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 234, in active_tenant_id
    self._init_tenant()
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 263, in _init_tenant
    tenant_info = self.graphql({"query": {"tenant": {"id"}}})
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 304, in graphql
    retry_on_api_error=retry_on_api_error,
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 220, in post
    retry_on_api_error=retry_on_api_error,
  File "/usr/local/lib/python3.6/site-packages/prefect/client/client.py", line 473, in _request
    raise ClientError("Malformed response received from API.") from exc
prefect.utilities.exceptions.ClientError: Malformed response received from API.

Can it be a result of running it without root privileges?

[zanieb]

Hey @adammrozik

The run_config is for flows that are registered with a backend (flow.register()) then run from the UI or API. flow.run() is always for local runs of flows.

Re your agent, are you using Prefect Server? It looks like you may be missing a default tenant.

[adammrozik]

Hey Michael! Thanks for the quick answer! Right now we are not using prefect server. Is it mandatory to use it for this kind of work? I think I am confused with what is the actual role of Kubernetes agent if dask is doing all of the tasks czw., 4 mar 2021, 17:41 użytkownik Michael Adkins <[email protected]> napisał:

…

Hey @adammrozik <https://github.com/adammrozik> The run_config is for flows that are registered with a backend ( flow.register()) then run from the UI or API. flow.run() is always for local runs of flows. Re your agent, are you using Prefect Server? It looks like you may be missing a default tenant. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#4208 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ARPF4NRM4I5MLBBJJA25YD3TB6Z4DANCNFSM4YTRTFKQ> .

[zanieb]

Prefect Server isn't mandatory -- you can use Prefect Cloud instead. One of the backends is required to use Agents though since they query an API for flows to run.

The Agents allow you to schedule execution of your flows on different infrastructure than your development machine. For example, a workflow could look like

• Write a flow
• Push to github
• CI registers the flow with Prefect Cloud
• A user runs the flow from Prefect Cloud
• The flow run is picked up by a KubernetesAgent
• The flow runs as a Kubernetes Job
• The K8s container runs dask to manage data flow
• The user can watch the flow run's status in realtime at the UI