Hi. Is there any method to pass env_file to

pyflyte run --remote

command? I was wondering if I could add

--env-file

option like

docker run

command.

Hi everyone, I was trying to test integration between feast and minio s3 store using flytekit pipeline I have faced this error could anybody help

"name": "flytekit", "levelname": "WARNING", "message": "Unsupported Type <class 'feast.repo_config.RepoConfig'> found, Flyte will default to use PickleFile as the transport. Pickle can only be used to send objects between the exact same version of Python, and we strongly recommend to use python type that flyte support."}

this is the code

import os
import boto3
import logging
from flytekit import task
from feast.infra.offline_stores.file import FileOfflineStoreConfig
from feast.infra.online_stores.sqlite import SqliteOnlineStoreConfig
from feast.repo_config import RepoConfig
logger = logging.getLogger(__file__)

# execution on demo cluster
os.environ["FEAST_S3_ENDPOINT_URL"] = ENDPOINT = "<http://192.168.1.117:30002>"
os.environ["AWS_ACCESS_KEY_ID"] = "minio"
os.environ["AWS_SECRET_ACCESS_KEY"] = "miniostorage"

# flytectl config set s3.endpoint_url http://<minio_host>:<minio_port>
# flytectl config set s3.access_key <minio_access_key>
# flytectl config set s3.secret_key <minio_secret_key>

bucket_name = "my-s3-bucket"
registry_path = "data/registry.db"
online_store_path = "data/online_store.db"

@task
def create_bucket(
    bucket_name: str, registry_path: str, online_store_path: str
) -> RepoConfig:
    client = boto3.client(
        "s3",
        aws_access_key_id="minio",
        aws_secret_access_key="miniostorage",
        use_ssl=False,
        endpoint_url="<http://192.168.1.117:30002>",
    )
    return RepoConfig(
        registry=f"<s3://my-s3-bucket/data/online_store.db>",
        project="my_project",
        provider="local",
        offline_store=FileOfflineStoreConfig(),
        online_store=SqliteOnlineStoreConfig(path=online_store_path),
        entity_key_serialization_version=2,
    )

Hello, is there any example/documentation for flyte-binary charts values for deployment in GCP?

#ask-the-community I am still stuck on the flytectl vs flytekit.remote not showing consisten behaviour. flytectl get projects works fine with clientsecret method but flytekit.remote throws the below error {"asctime": "2023-01-31 124443,934", "name": "flytekit.remote", "levelname": "WARNING", "message": "This feature is still in beta. Its interface and UX is subject to change."} {"asctime": "2023-01-31 124443,997", "name": "flytekit.cli", "levelname": "ERROR", "message": "Unauthenticated RPC error <_InactiveRpcError of RPC that terminated with:\n\tstatus = StatusCode.UNAUTHENTICATED\n\tdetails = \"token parse error [JWT_VERIFICATION_FAILED] Could not retrieve id token from metadata, caused by: rpc error: code = Unauthenticated desc = Request unauthenticated with IDToken\"\n\tdebug_error_string = \"UNKNOWN:Error received from peer ipv410.183.180.0443 {created_time:\"2023-01-31T124443.996825038+02:00\", grpc_status:16, grpc_message:\"token parse error [JWT_VERIFICATION_FAILED] Could not retrieve id token from metadata, caused by: rpc error: code = Unauthenticated desc = Request unauthenticated with IDToken\"}\"\n>, refreshing credentials and retrying\n"} ^CProcess ForkProcess-1: Traceback (most recent call last): File "/home/samuel/.local/lib/python3.8/site-packages/flytekit/clients/raw.py", line 150, in handler return fn(*args, **kwargs) File "/home/samuel/.local/lib/python3.8/site-packages/flytekit/clients/raw.py", line 448, in list_workflows_paginated return self._stub.ListWorkflows(resource_list_request, metadata=self._metadata) File "/home/samuel/.local/lib/python3.8/site-packages/grpc/_channel.py", line 946, in call return _end_unary_response_blocking(state, call, False, None) File "/home/samuel/.local/lib/python3.8/site-packages/grpc/_channel.py", line 849, in _end_unary_response_blocking raise _InactiveRpcError(state) grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAUTHENTICATED details = "token parse error [JWT_VERIFICATION_FAILED] Could not retrieve id token from metadata, caused by: rpc error: code = Unauthenticated desc = Request unauthenticated with IDToken" debug_error_string = "UNKNOWN:Error received from peer ipv410.183.180.0443 {created_time:"2023-01-31T124443.996825038+02:00", grpc_status:16, grpc_message:"token parse error [JWT_VERIFICATION_FAILED] Could not retrieve id token from metadata, caused by: rpc error: code = Unauthenticated desc = Request unauthenticated with IDToken"}"

How can i retrieve tasks in a particular workflow in flyte using flytectl

the examples shown in flytectl docs only show tasks in project.... My project has 2 worklows and I need to get the tasks only for 1 workflow

Hello, good morning! I was wondering if the team is planning to hold office hours today? My team has a quick question on max task nodes and figured it might be a good place to address it. Thanks!! @Dan Rammer (hamersaw) @Samhita Alla

My task that requires secrets management is currently stuck on queue state, likely due to certificate issues on webhook. Logs on Flyte propeller:

plugin execution, caused by: failed to execute handle for plugin [container]: [InternalError] failed to create resource, caused by: Internal error occurred: failed calling webhook "<http://flyte-pod-webhook.flyte.org|flyte-pod-webhook.flyte.org>": failed to call webhook: Post "<https://flyte-pod-webhook.flyte.svc:443/mutate--v1-pod?timeout=10s>": x509: certificate has expired or is not yet valid: current time 2023-02-01T05:22:42Z is after 2022-11-24T05:12:16Z

The cert is last updated 1 year ago, so it probably expired. How can i rotate the cert?

cc: @Arief Rahmansyah

Hi Team, I was in process to install flyte on GKE, I see the manual for installation on GCP (https://docs.flyte.org/en/latest/deployment/gcp/manual.html) is removed from the doc and it redirect to a new deployment docs which doesn't have clear steps as before. Have we changed the way of installation ? If installation process is as before then this new doc is really difficult to follow

Hello team, While applying a map function on spark rdd, job failed unexpectedly. flyte Spark job is executed on local machine. Any suggestion on how to resolve this failure.

flytekit.exceptions.scopes.FlyteScopedUserException: An error occurred while calling z:org.apache.spark.api.python.PythonRDD.runJob.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4.0 failed 1 times, most recent failure: Lost task 0.0 in stage 4.0 (TID 3) (localhost executor driver): org.apache.spark.SparkException: Python worker exited unexpectedly (crashed)


INFO:py4j.clientserver:Closing down clientserver connection

Hi all - wondering if having multiple optional inputs for tasks (with defaults provided) is supported? Will drop a minimal example in 🧵, but essentially we have tasks with one default provided that serialize just fine but tasks with more than one default provided that don't serialize, with the error

flytekit.exceptions.user.FlyteAssertion: Input was not specified for: optional_int of type simple: INTEGER

. Issues #1312 [BUG][flytekit] Default input argument not being considered in PythonInstanceTask and #3046 both seem related. Has anyone else seen this before? (flytekit 1.3.0)

I would also have a question with regards to spot instances. We have a subworkflow (workflow within a workflow) which starts tasks on spot nodes (

interruptible=True

). The task then fails with

Last Error: USER::Pod was terminated in response to imminent node shutdown.

The last log from the node where this task is run on is:

Deleting node <node-id> because it does not exist in the cloud provider

So we assume the node is reclaimed by GCP. Does anyone happen to know whether there might be a difference in how interruptible task failures are handled in subworkflows?

Hello, What’s the mechanism to set and retrieve environment variables defined in Flyte?

Use cases:
ENV = dev, beta, or prod
Custom constant e.g. PERSIST_BUCKET_PREFIX = 'hbomax-datascience-deployment', -> s3_bucket = f'{PERSIST_BUCKET_PREFIX}-{ENV}'

So I need both ENV and PERSIST_BUCKET_PREFIX as env vars.

Hi team.. I’m seeing my map tasks get evicted from k8s even with the

interruptible=False

flag.. which ends up dragging the task on much longer than it should (since it retries). Is there something different for map tasks?

Hello, Could you explain the mechanism to set and retrieve globally defined variables so that they are available to all workflows and tasks in Flyte? Ideally, they can be changed dynamically through Flyte Admin, not baked into the docker image as env vars. E.g. an S3 bucket name for all the flyte projects, etc.

Hi people! I have not found a lot of documentation about the Prometheus metrics made available by Flyte. Are there metrics about tasks such as individual task completion times, durations, etc? Much like we have for workflows with metrics like "flytepropellerallworkflowsuccess_duration_ms_count". These informations are on Flyte console so I wonder if they are also available by metrics or if at least there is a plan to implement that. Also, are there metrics about CPU and memory usage by workflow/task?

Hi team.. our Flyte deploy is acting up today. Flyteadmin and Flytescheduler pods keep going down randomly. We’ve not touched the deployment itself, hence quite confused about how to root cause this. In the scheduler logs we see this

rpc error: code = Unavailable desc = connection error: desc = "transport: Error while dialing dial tcp 172.20.71.119:81: connect: connection refused"

Not sure what component it’s trying to connect to… and not sure what’s causing the flakiness… Replicas are starting and stopping somewhat arbitrarily. Any insight would be appreciated

Hi Flyte community! Sharing here a resource that has been very useful in the past for Kubernetes troubleshooting and which remains largely up to date: https://learnk8s.io/a/a-visual-guide-on-troubleshooting-kubernetes-deployments/troubleshooting-kubernetes.en_en.v3.pdf The good thing is that it follows a bottom-up kind of methodology to explore possible problems at each resource in the K8s hierarchy. Most of the times you won't have to go through all the verifications there, and sometimes the problem will be somewhere else 🫤 but it's a good starting point. Feel free to share any other resource that you consider useful for this purpose. Thanks!

Hey all, I'm trying to experiment with "fast registration" via pyflyte but am unsuccessful and having a hard time finding docs that work. The docs tell me to use

--fast

in conjunction with

--image

to do a fast register of local code that will be "overlayed" onto to the specified image (as I understand it). But in my trials I find that: 1. --fast is not a recognized option, eventually realized it is a command (without dashes), are the docs are wrong? 2. Then it complains "No such option --image". Again, are docs are wrong? Or I'm misreading something. I was trying to modify a command that works without "fast" into one that runs with "fast" to see if I can avoid building and pushing a new image (the image I specify in this case already exists). The command looks like this, where I've inserted $FAST_REG into a command that otherwise works to serialize against a given image:

pyflyte --pkgs plaster.genv2 serialize $FAST_REG --image <my-ecr-registry-path>/<my-image-name>:<tag> --local-source-root . workflows -f /tmp/_pb_outputs

...where the $FAST_REG is where I'm attempting to insert "fast" and run my local code overlayed with what is in the image I'm specifying. I'm probably misunderstanding something, and the docs aren't helping me. Thanks for any tips!

Hi Flyte community, question about the general design of dynamic workflow. Why is dynamic workflow executed in a k8s pod instead of in the flyte propeller? Won’t this lead to potential performance issue due to the overhead of spinning up new k8s pods?

Hi there! I'm just discovering flyte and have question regarding pipelines, which I didn't find the answer to yet: Is it possible to achieve this: • Task A ◦ checks external source periodically for updates ◦ if external source changes, provide different outputs ◦ invalidate all produced artifacts (e.g. checkpoints, metrics, ...) of the whole pipeline • Task B (depends on Task A) ◦ restart when Task A changes its outputs The use-case would be that a user changes the training-configuration and thus the training has to be restarted. Also, all produced checkpoints should be invalidated.

Maybe I think inside the wrong box. Would the "flyte-way" be this: • new config -> new "experiment" • keep previous experiments • ability to delete any experiment (ability to roll-back in time) ?

How would I implement "dynamic" workflows like: • network training is running • I add new training data • the new data gets automatically integrated into the training-process or is this out of scope for pipelines and I have to work with my own events?

I found the

@dynamic

decorator. Is it true that I can use it to: • run a thread that e.g. looks for new data • if new data is found: ◦ spawn a new subgraph ▪︎ rebuild the dataset ▪︎ cancel the current training ▪︎ (re)start training task ?

How can I stop execution of a subgraph within my

@dynamic

workflow python function?

Hello, good morning! We recently had an issue during registration with a large workflow (it has >100 task nodes, and calls two sub-workflows within the main workflow). The docker images build fine, and it also seemingly packages okay, but in the middle of the registration process we get a

context deadline exceeded

rpc error. It looks like this: The error:

------------------------------------------------------------------------ --------- --------------------------------------------------- 
| /tmp/register4200568562/16_workflow.cx_topic_model_workflow_2.pb       | Failed  | Error registering file due to rpc error: code =   |
|                                                                        |         | DeadlineExceeded desc = context deadline exceeded |
 ------------------------------------------------------------------------ --------- ---------------------------------------------------

In the doc linked below, it's mentioned that the flytectl wait deadline is 15 seconds: https://docs.flyte.org/en/latest/community/troubleshoot.html#troubles-with-flytectl-commands-with-auth-enabled We looked around but weren't able to find the setting to increase that wait time. Any guidance here would be much appreciated!! Thanks so much Also, here's the packaging success message (we have a 37-node wf in this specific message, but it's the same registration issue):

Packaging app_module.workflow.cx_topic_model_workflow -> 36_app_module.workflow.cx_topic_model_workflow_3.pb
Successfully packaged 37 flyte objects into /root/flyte-package.tgz

Hi there, what would be involved in being able to pass information through the type system to the data persistence layer to apply tags to objects that get stored in blob stores via flytekit? Eg. say I'd like to output a FlyteFile of some type and give it a tag so when it gets stored in s3 (or gcp) it will have that tag (eg. it helps associate the object with a project or department). At the moment, it seems the data persistence layer interface really doesn't know anything except paths of objects to/from destinations. Would it be feasible to plumb tags through to the persistence layer via type annotations (for backends that support tags)?