Thanks to help from the prefect team :hattip:, I h...
# prefect-community
Thanks to help from the prefect team hattip, I have a prefect server running in Kubernetes with the experimental helm chart. I also have dask gateway running. To connect up the bits, I need to get the prefect agent running in the cluster, and submitting jobs to dask gateway. I can get the kubernetes agent running locally against the (port-forwarded) server in K8. But: • What config do I need to point it at the cluster FQDN of the prefect server, when running in a container? • It would seem that the kubernetes agent might not be what I want anyway — as I want to submit flows to the dask gateway, not as K8 jobs. Or perhaps I can configure the agent container environment as specified here? ? • I’m a bit unsure about the relation between the agent and the server. Is there any doc explaining? Is the server purely passive, and the agent is the active component? (Then what is
doing?) Some explanation as to how the pieces interact (and/or how I should best go about getting them to interact in my scenario) would be great. Thanks!
👏 3
Hi Shaun, this is great to hear. A few assorted answers: • The agent submits your jobs (where the
runs), while dask runs tasks (where the
runs). So whether you want to use the k8s agent or not depends on if you want your entire flow to run in k8s, or just the tasks (if you're used to dask, you can think of this as where does your dask client live (inside or outside the cluster), while the workers managed by dask-gateway are always inside the cluster). I personally would recommend using the k8s agent, as it keeps everything uniformly in k8s. • To configure your jobs to use dask-gateway, you'd want to configure a
on your
, complete with
to configure the cluster. Something like:
Copy code
executor = DaskExecutor(
        # any args you need to pass to configure your cluster
    adapt_kwargs={"minimum": min_workers, "maximum": max_workers},
flow.environment.executor = executor
• The server stores jobs that should be run, and never pushes anything out. Agents run elsewhere, and poll the server for any jobs they should run (the server never reaches out actively to agents). Towel is for periodic background jobs to manage server resources, you shouldn't need to worry about that. I agree, we definitely could document better here.
I'll let @josh chime in more with configuring core to work with server on k8s, since he's had it working with k8s in the past.
I actually have it working in k8s as we speak using the helm chart @Shaun Cutts made 😉
🚀 1
💯 1
@Jim Crist-Harif — Thanks for your answer. And thanks for your help, @josh 🙂 — so when the agent gets a scheduled flow from the server, the executor is already “baked in”. Can I configure with env variables rather than have to submit with cloudpickle with every flow?
My goal is to have the agent submit to dask gateway by default if nothing else is specified on flows that have been send to the server.
WRT the agent -> server communication: the only thing you need to ensure for core to work with the server running in k8s is that it can talk to the Apollo endpoint of the server. So you can set the API on your agent either through the `start`/`install` commands with the
flag or you can just set the
env var to that endpoint
You're asking if you can configure defaults for every flow, rather than relying on configuring per-flow? If so, you can currently configure the default executor class and cluster class, but not cluster kwargs. However! Dask-gateway supports configuring default kwargs itself as part of the dask config, so you can tell prefect to use dask-gateway, then configure your defaults via dask's config in your image/environment.
The dask config you'd want to set would be
, for prefect you'd want to configure
All can be set via environment variables, or as static config files in your images.
@josh — great
was what I was looking for I think (though this is “server” not “cloud”). @Jim Crist-Harif — ok. So I can set that up as environment config in the agent container?
That would go in your flow run images, not in the agent. The agent submits new jobs, which will create executors from their local config.
Yeah @Shaun Cutts the config on the API is a bit confusing because it interprets the value based on some settings. So you can set
to your location to also set it. The default backend is set to cloud so setting the CLOUD__API does the same trick. Setting the backend to server and the server endpoint is more explicit actually so you might want to go that way
@Jim Crist-Harif Ah — so I need to tell the agent to put that into the environment of the flow images it creates. I see it has --env config for that. Hmm…
@josh 👍
Oh wait, nvm, it looks like flow's always default to configuring an executor themselves when they're created (defaults to
. So the prefect config would have to go in your users environments when they're registering flows. The dask config could still go in the runtime environments.
Ah … hmm… would you be interested in a PR to the K8 agent that (with option) allows it to override “LocalExecutor” with another? … I guess I could just write appropriate wrappers in user environment as well …
That wouldn't happen on the agent, since the agent has nothing to do with executors. Hmmmm. We're currently rethinking agents and environments, so I'm a bit hesitant to add more options here right now, but we can definitely work to support this use case in the future. Is this a blocker for you currently?
[And just to confirm I can get the dask config to be set up in the images using --env options on the k8 agent?] As to being a blocker — I guess not. I don’t want to have to expose dask gateway credentials in the user environment. But that part I could setup using the agent, right? And then I could write a wrapper for the other parts for flow creation.
All the dask-gateway stuff can be statically configured by you, yes. You could do it with the
option, by adding it to the default job yaml spec, or by setting the config in the images themselves - any of these would work. The only thing you'd need to ensure on the user side is that they configure flow environments that should use dask with a
on their environments before registering.
Great — thanks, Jim! I think I’ve got enough to work with (until I get snarled up)!
Great! Please reach out if you run into any issues, I'm happy to help. Glad dask-gateway is working for you!
@josh -- I think you meant PREFECT__SERVER___API (not …__ENDPOINT) as above, right? With the latter, I can’t connect. Now with the former, (set to http://prefect-server-apollo.prefect-server.svc.cluster.local:4200) I’m getting error below: … which is different than previously. Does “PREFECT__CLOUD__AGENT__AGENT_ADDRESS” have to be set — out of the box it comes out as http//8080. Is this for healthcheck?
Copy code
Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/urllib3/", line 160, in _new_conn
    (self._dns_host, self.port), self.timeout, **extra_kw
  File "/usr/local/lib/python3.6/site-packages/urllib3/util/", line 84, in create_connection
    raise err
  File "/usr/local/lib/python3.6/site-packages/urllib3/util/", line 74, in create_connection
ConnectionRefusedError: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/urllib3/", line 677, in urlopen
  File "/usr/local/lib/python3.6/site-packages/urllib3/", line 392, in _make_request
    conn.request(method, url, **httplib_request_kw)
  File "/usr/local/lib/python3.6/http/", line 1287, in request
    self._send_request(method, url, body, headers, encode_chunked)
  File "/usr/local/lib/python3.6/http/", line 1333, in _send_request
    self.endheaders(body, encode_chunked=encode_chunked)
  File "/usr/local/lib/python3.6/http/", line 1282, in endheaders
    self._send_output(message_body, encode_chunked=encode_chunked)
  File "/usr/local/lib/python3.6/http/", line 1042, in _send_output
  File "/usr/local/lib/python3.6/http/", line 980, in send
  File "/usr/local/lib/python3.6/site-packages/urllib3/", line 187, in connect
    conn = self._new_conn()
  File "/usr/local/lib/python3.6/site-packages/urllib3/", line 172, in _new_conn
    self, "Failed to establish a new connection: %s" % e
urllib3.exceptions.NewConnectionError: <urllib3.connection.HTTPConnection object at 0x7f4b536d77f0>: Failed to establish a new connection: [Errno 111] Connection refused

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/site-packages/requests/", line 449, in send
  File "/usr/local/lib/python3.6/site-packages/urllib3/", line 727, in urlopen
    method, url, error=e, _pool=self, _stacktrace=sys.exc_info()[2]
  File "/usr/local/lib/python3.6/site-packages/urllib3/util/", line 439, in increment
    raise MaxRetryError(_pool, url, error or ResponseError(cause))
urllib3.exceptions.MaxRetryError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f4b536d77f0>: Failed to establish a new connection: [Errno 111] Connection refused',))

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/bin/prefect", line 8, in <module>
  File "/usr/local/lib/python3.6/site-packages/click/", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/click/", line 782, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.6/site-packages/click/", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.6/site-packages/click/", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.6/site-packages/click/", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.6/site-packages/click/", line 610, in invoke
    return callback(*args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/click/", line 21, in new_func
    return f(get_current_context(), *args, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/prefect/cli/", line 295, in start
  File "/usr/local/lib/python3.6/site-packages/prefect/agent/kubernetes/", line 88, in __init__
  File "/usr/local/lib/python3.6/site-packages/prefect/agent/", line 153, in __init__
    self.client = Client(, api_token=token)
  File "/usr/local/lib/python3.6/site-packages/prefect/client/", line 124, in __init__
    tenant_info = self.graphql({"query": {"tenant": {"id"}}})
  File "/usr/local/lib/python3.6/site-packages/prefect/client/", line 281, in graphql
  File "/usr/local/lib/python3.6/site-packages/prefect/client/", line 237, in post
  File "/usr/local/lib/python3.6/site-packages/prefect/client/", line 401, in _request
    session=session, method=method, url=url, params=params, headers=headers
  File "/usr/local/lib/python3.6/site-packages/prefect/client/", line 319, in _send_request
    response = <|>(url, headers=headers, json=params, timeout=30)
  File "/usr/local/lib/python3.6/site-packages/requests/", line 578, in post
    return self.request('POST', url, data=data, json=json, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/requests/", line 530, in request
    resp = self.send(prep, **send_kwargs)
  File "/usr/local/lib/python3.6/site-packages/requests/", line 643, in send
    r = adapter.send(request, **kwargs)
  File "/usr/local/lib/python3.6/site-packages/requests/", line 516, in send
    raise ConnectionError(e, request=request)
requests.exceptions.ConnectionError: HTTPConnectionPool(host='localhost', port=4200): Max retries exceeded with url: / (Caused by NewConnectionError('<urllib3.connection.HTTPConnection object at 0x7f4b536d77f0>: Failed to establish a new connection: [Errno 111] Connection refused',))
@josh @Chris White — AHA — so I go it working with this config:
Copy code
          value: server
        - name: PREFECT__SERVER__API
          value: <http://prefect-server-apollo.prefect-server.svc.cluster.local:4200>
        - name: PREFECT__CLOUD__API
          value: <http://prefect-server-apollo.prefect-server.svc.cluster.local:4200>
I think this means that PREFECT__BACKEND isn’t always respected…?
Prefect backend needs to be set to server on any client process that needs to talk to server (as opposed to cloud), so it makes sense that you’d need to set it on the agent image 👍
@Chris White The issue is that it needs both
. If I specify just the former (together with
), it fails.
That’s because there is no
Copy code
The config looks like this:
Copy code
backend = "cloud" or "server"

host = "<http://localhost>"
port = "4200"
endpoint = "${}:${server.port}"

api = "${${backend}.endpoint}"
endpoint = "<>"
graphql = "${cloud.api}/graphql"
Where ultimately
is what is used however it is interpreted off of each backend’s respective
Hmm… but I tried “ENDPOINT” as well AH — so If you specify
, and leave the default
, it doesn’t work, even when backend is set to server. But it works if
is not present. Ok — so that makes sense in terms of what you wrote, but is confusing if you are trying to override values from “default configuration” as I was. Perhaps better would be to have something like:
Copy code
host = "<http://localhost>"
port = "4200"
endpoint = "${}:${server.port}"

endpoint = "<>"

endpoint = "${${backend}.endpoint}"
graphql = "${api.endpoint}/graphql"
and use
as the canonical settings….
Yeah IIRC it was kept under
for some backwards compatibility reasons but I’m starting to think we can alleviate that in the near future especially because we are already doing a core version check on flows in the agent to determine which version of prefect they were registered on. aka anything under something like
would use the old config and so forth
Thanks josh!