Michail Melonas
09/01/2021, 2:25 PMsecret = PrefectSecret("SOME_SECRET")
flow.storage = S3(bucket="some-bucket", secrets=["AWS_CREDENTIALS"])
flow.run_config = KubernetesRun(
image="<http://registry.gitlab.com/<some_image>|registry.gitlab.com/<some_image>>",
labels=["prefect-14"],
env={
"PREFECT__LOGGING__LEVEL": "DEBUG",
"SECRET": secret
},
image_pull_secrets=["image-puller"],
)
and then running:
flow.register("ProjectName")
I get:
TypeError: Object of type PrefectSecret is not JSON serializable
Any thoughts on this?Kevin Kho
Michail Melonas
09/01/2021, 2:34 PMKevin Kho
PrefectSecret(…).run()
. Could you try that?Michail Melonas
09/01/2021, 2:54 PMPrefectSecret
run
method solves the TypeError
. However, I can see in the console that it returns the value of SOME_SECRET.Kevin Kho
--env
flag? prefect agent kubernetes start --env SECRET=value
?Michail Melonas
09/01/2021, 3:16 PMfrom prefect.engine.executors import DaskExecutor
import coiled
executor = DaskExecutor(
cluster_class=lambda: coiled.Cluster(**{
"software": "somesoftwareenviron",
"shutdown_on_close": False,
"name": "prefect-executor",
"n_workers": 1,
"credentials": "local",
})
)
flow.executor = executor
I'm trying to provision a Coiled cluster. I have a kubernetes agent sitting on a K8s cluster.Kevin Kho
credentials
there?Michail Melonas
09/01/2021, 3:22 PMcredentials
argument is not the actual credentials, but is used to indicate which set of credentials to rely on
In the docs:Kevin Kho
Michail Melonas
09/01/2021, 3:28 PMPrefectSecret
.Kevin Kho
flow.storage = S3(bucket="some-bucket", secrets=["AWS_CREDENTIALS", "SOME_SECRET"])
and I think this will fetch it from Cloud when Prefect is loaded. Will try it in a bit.Michail Melonas
09/01/2021, 3:44 PMPrefectSecret("SOME_SECRET").run()
.Kevin Kho
Chris L.
09/15/2021, 2:03 AMsecrets=["COILED_TOKEN"]
to Storage, I'm wondering if you passed the secret from prefect.context.secrets
into your custom function for creating the Coiled cluster: i.e.
def coiled_cluster():
import dask
dask.config.set({"coiled.token": prefect.context.secrets.COILED_TOKEN})
return coiled.Cluster(**{
"software": "somesoftwareenviron",
"shutdown_on_close": False,
"name": "prefect-executor",
"n_workers": 1,
})
flow.storage = S3(bucket="some-bucket", secrets=["COILED_TOKEN"])
flow.executor = DaskExecutor(cluster_class=coiled_cluster)
This pattern follows ideas from passing the "main image" (i.e. Docker image specified on run config for the Agent / flow runner) into the DaskExecutor (which manages the task runner) at runtime (see https://docs.prefect.io/orchestration/flow_config/executors.html#specifying-worker-images)
I'm planning to test this pattern in the weekend, but wondering if you've tried this already (before you came to your Docker ENTRYPOINT solution)Michail Melonas
09/15/2021, 6:42 AMKevin Kho
Chris L.
09/15/2021, 3:36 PMKevin Kho