👋 I'm running into an auth problem with flytectl using the ClientSecret (client credentials) method. I've assigned the proper scopes to the oauth2 client and the flyteadmin service isn't complaining about not being able to find the token in the metadata anymore which is nice. However now I'm getting a response to flytectl that I'm missing a scope:

Error: Connection Info: [Endpoint: dns:///flyte.localhost:80, InsecureConnection?: true, AuthMode: ClientSecret]: rpc error: code = Unauthenticated desc = authenticated user doesn't have required scope

LinkerD confirms that it's the flyteadmin container that responds with Unauthenticated. This is quite perplexing and I've verified that the token is sent and it contains all the scopes required:

"scope": "all email offline profile"

I hope people can give me some hints as I've been bashing my head for way to long against this. I'm just trying to have a flytectl register stuff in CI.

Workflow execution seems to start but failing with the error below and I dont see any other stacktrace.

Pod failed. No message received from kubernetes.
[f1943114e818d4b7eaf3-n0-0] terminated with exit code (1). Reason [Error]. Message: 
exec /opt/venv/bin/pyflyte-execute: exec format error
.

Hi guys! Currently using Vertex Pipelines at our company and curious on the benefits of Flyte? Vertex is serverless and requires very little maintenance and no k8 knowledge. What would be your argument to make the shift? We are only a couple of MLE’s.

Hello everybody, When running a workflow with a dynamic workflow, I keep getting this error:

Traceback (most recent call last):
  File "/home/flyte/.venv/lib/python3.8/site-packages/flytekit/bin/entrypoint.py", line 90, in _dispatch_execute
    outputs = _scoped_exceptions.system_entry_point(task_def.dispatch_execute)(ctx, idl_input_literals)
AttributeError: 'PythonFunctionWorkflow' object has no attribute 'dispatch_execute'

Somebody have any idea what it could be?

I notice in our workflow when a task in the dag errors and there are other parallel tasks running that they are aborted. Is it possible to have those tasks that are already running finish so they don't need to be recalculated in the next execution?

Hi Flyte folks, does the Optional type has to be defined like this

@task
def hello(username: Optional[str]) -> str:

because i tried to define like this

@task
def hello(username: str | None) -> str:

but getting

Failed with Unknown Exception <class 'AssertionError'> Reason: Failed to Bind variable username for function sandbox.subworkflow.hello.

first one works though

Cross posting here for visibility . https://flyte-org.slack.com/archives/C01P3B761A6/p1684866651162639

Hello All Can mappable task return a list of values, Like:

@task
def a_mappable_task(a: int) -> List[int]:
    l = [a] * 3
    return l

Is there a way to have my dynamic workflow raise the last error from one of its invoked tasks directly? I want to make sure that the correct error and error code are the outer most errors coming from the workflow

Hi there! I'm interested in the ability to run flyte workflows with priorities so that workflows with a higher priority are executed before those with a lower one when they are running at the same time. I found some information about this in the documentation but did not understand how exactly you can start the workflow by specifying its priority as a parameter. Another question is - how can resources be managed? Let's say I have a CPU-intensive workflow, and it was experimentally revealed that my cluster is capable of processing up to 10 such workflows at the same time, what will happen if we run 20 workflows?

Hi All, I am seeing this error when I am trying to run flyte on k8s. What am I missing?

{"json":{},"level":"warning","msg":"Failed to create cluster resources for namespace [flytesnacks-development] with err: Failed to read config template dir [flytesnacks-development] for namespace [] with err: open : no such file or directory","ts":"2023-05-24T09:40:15Z"}
{"json":{},"level":"warning","msg":"Failed to create cluster resources for namespace [flytesnacks-staging] with err: Failed to read config template dir [flytesnacks-staging] for namespace [] with err: open : no such file or directory","ts":"2023-05-24T09:40:15Z"}
{"json":{},"level":"warning","msg":"Failed to create cluster resources for namespace [flytesnacks-production] with err: Failed to read config template dir [flytesnacks-production] for namespace [] with err: open : no such file or directory","ts":"2023-05-24T09:40:15Z"}
{"json":{},"level":"warning","msg":"Failed cluster resource creation loop with: Failed to read config template dir [flytesnacks-development] for namespace [] with err: open : no such file or directory, Failed to read config template dir [flytesnacks-staging] for namespace [] with err: open : no such file or directory, Failed to read config template dir [flytesnacks-production] for namespace [] with err: open : no such file or directory","ts":"2023-05-24T09:40:15Z"}
{"json":{},"level":"error","msg":"Failed to initialize certificates for Secrets Webhook. client rate limiter Wait returned an error: context canceled","ts":"2023-05-24T09:40:20Z"}
{"json":{},"level":"panic","msg":"Failed to start Propeller, err: failed to create FlyteWorkflow CRD: <http://customresourcedefinitions.apiextensions.k8s.io|customresourcedefinitions.apiextensions.k8s.io> is forbidden: User \"system:serviceaccount:test-apps:test-flyte-role\" cannot create resource \"customresourcedefinitions\" in API group \"<http://apiextensions.k8s.io|apiextensions.k8s.io>\" at the cluster scope","ts":"2023-05-24T09:40:20Z"}

Hey flyte community, been doing some autoscaling stuff with Karpenter and EKS with flyte and performance is really good and fast. Just one question, is there any sort of inbuilt logic within Flyte to allocate tasks/pods to certain provisioners/nodegroups automatically based on CPU/MEM request sizes (instead of using nodeselectors,taints), or is this custom logic we would have to build out ourselves? The reason I ask is because the end-user for our use-case would be a non-developer/non-programmer and so we would like to abstract away as much of the configuration from them as possible. Piping in the pod-templates and pod specs manually into the task defintions or using the with_overrides method works just fine, but unfortunately, even something like this may prove too much for the end-user. Obviously this is probably not the intended workflow for Flyte, but was wondering if anybody had any insight into this. Or if the custom logic is the answer, then that's fine too (I am assuming some sort of wrapper function to convert our users' request sizes into pod specs/taints/tolerations to pipe into the task invocations)

Has someone already tried running or integrating CWL with Flyte?

I was looking through the flyte config structs, and flytestdlib uses a StowConfig. Looking at the stow repo it doesn't look like that project has released in over a year and there are no PRs or issues open. Is stow still a supported part of flyte?

Is it possible to have a FlyteDeck show up even when a task fails? For example, I want to publish a pandera report of data check failures.

How can I add secrets to

pyflyte run

command? When I try to run, I keep getting

Key SECRET_KEY on secret flyte-default-credentials not found.

Hi, guys! I am running my workflows using airflow. And I want to read outputs of the previous execution before running a new one. How to properly do that using flytekit python api? • How to get recent execution for a particular workflow (latest version)? I know that you can fetch_execution but only with known id • How to check the outputs of this executions? I’ve tried the following:

hook = FlyteHook(flyte_conn_id=FLYTE_CONN_ID, project=flyte_project, domain=flyte_domain)
remote = hook.create_flyte_remote()

# workflow = remote.fetch_workflow(flyte_project, flyte_domain, flyte_task)
executions = remote.recent_executions(flyte_project, flyte_domain)
executions_filtered = list(
    filter(
        lambda e: e.spec.launch_plan.name == flyte_task and e.is_done,
        executions,
    )
)
if executions_filtered:
    execution = executions_filtered[0]
    <http://logging.info|logging.info>(f"{execution.id}")
    outputs = execution.outputs
    <http://logging.info|logging.info>(f"{outputs}")

But filtering of executions is not really convenient and the outputs are empty (however they shouldn’t be so)

Hi everyone. Ive been working off the PR that adds a transformer for pydantic basemodels (https://github.com/flyteorg/flytekit/pull/1620) running into an unfortunate issue with flytepaths/directories though. These inherit from os.PathLike which falls outside of pydantic's supported json serialization/deserialization schemas. my understanding is also that flytefiles/directories need to be initialized in a particular way to work (below iv'e implemented a flytedirectory initialized inspired from the dataclass transformer). This works but seems a bit hacky so wanted to hear if anyone else encountered similar challenges. TLDR; how to serialize-deserialize flytedirectories/files in pydantic basemodels (or other objects)

from data_engine import (
    pydantic_transformer,
)  # code from <https://github.com/flyteorg/flytekit/pull/1620>


class Rax(pydantic.BaseModel):
    flytedir: str

    @pydantic.validator("flytedir")
    def validate_hax(cls, v: str) -> str:
        flytedir = make_flytedir(v)
        return str(flytedir.path)

    @classmethod
    def from_json(cls, json: str) -> "Self":
        self = cls.parse_raw(json)
        return self


@flytekit.workflow
def test_wf(rax: Rax) -> str:
    return test(rax=rax)  # type: ignore


@flytekit.task
def test(rax: Rax) -> str:
    return str(os.listdir(rax.flytedir))


def make_flytedir(path: Union[str, os.PathLike]) -> directory.FlyteDirectory:
    context = context_manager.FlyteContext.current_context()
    dimensionality = core_types.BlobType.BlobDimensionality.MULTIPART
    literal = make_literal(uri=path, dimensionality=dimensionality)
    transformer = directory_types.FlyteDirToMultipartBlobTransformer()
    out_dir = transformer.to_python_value(context, literal, directory.FlyteDirectory)
    os.listdir(out_dir)  # the dir isnt synced if we dont do this
    return out_dir


def make_literal(
    uri: Union[str, os.PathLike],
    dimensionality,
) -> literals.Literal:
    scalar = make_scalar(uri, dimensionality)
    return literals.Literal(scalar=scalar)  # type: ignore


def make_scalar(
    uri: Union[str, os.PathLike],
    dimensionality,
) -> literals.Scalar:
    blobtype = core_types.BlobType(format="", dimensionality=dimensionality)
    blob = literals.Blob(metadata=literals.BlobMetadata(type=blobtype), uri=uri)

return literals.Scalar(blob=blob) # type: ignore

noob question: how do i install transformers? i’ve tried

pip install transformers

,

pip3 install transformers

,

python3 -m pip install transformers

, and i could find transformers installed using

pip freeze --local

. it’s also installed in the docker image. but i can’t register in flyte without error

ModuleNotFoundError: No module named 'transformers'

is there a way to set postgres and s3/minio passwords as secrets instead of them being in plain text in a deployment?

When I try to use

with_overrides(pod_template_name="string")

on a NotebookTask, it appears that it doesn't do anything. Is this a Papermill/NotebookTask issue or have I perhaps specified it incorrectly? Setting the

pod_template_name="string

parameter directly on the NotebookTask definition itself works, but not with

with_overrides

. Basically, this works:

nb = NotebookTask(
    pod_template_name="random template name",
)

@dynamic
def function():
    return nb

# No problem

But overriding doesn't work within a dynamic workflow:

@dynamic()
def dynamic_workflow() -> List[float]:

    task_output = nb().with_overrides(
        pod_template_name="string to override"
    )

    return task_output

# Pod template name isn't overriden

I'm not sure if my question belongs here or in #flytekit ... I'm seeing map-tasks commonly fail with the error:

[85]: code:"ContainersNotReady|ImagePullBackOff" message:"containers with unready status: [f0ff4b2cb0d6d44d1907-n1-0-dn0-0-dn10-0-85]|Back-off pulling image \"(sanitized).<http://dkr.ecr.us-east-1.amazonaws.com/flyte-plaster:23.5.19\%22%22|dkr.ecr.us-east-1.amazonaws.com/flyte-plaster:23.5.19\"">

This is part of a workflow that employs a mappable task for one section. The workflow will complete successfully maybe half of the time. The other half, this error occurs specifically for the map task. I can see from logs that some elements of the mapped task are executing successfully, and in this case it is number 85 (out of 100) that failed. These are all executing on the same node (this is by design for mapped tasks, yes?), and they all use the same image, which is clearly already on the node since other elements from the mapped task "list" are executing. Edit: The above was incorrect. I misread/misunderstood the "node" aspect: a map task runs within a single "workflow node", not a single computer-node (e.g. ec2-instance). In fact, the 100 elements of the map task are being executed across 5 different machines. Still, the puzzle remains (see the reply) - the task that supposedly failed due to "ContainersNotRead|ImagePullBackoff" was in fact running when the job failed. So how could the container not be ready, or the image need pulling? Thoughts?

Spoiler: I'm pretty much a Python noob. I've started writing a plugin, deriving my plugin Task from

PythonCustomizedContainerTask[T]

and

PythonTask[PluginConfig]

. I then started using my plugin in the workflow, which works out fine. Now I would like to apply caching and other usual decorators like requests and limit. How can I achieve this, as I don't have the classic

Task(...)

decorator for my plugin?

class GlueTask(PythonCustomizedContainerTask[T], PythonTask[GlueConfig]):
    # following the usual init

class GlueTaskExecutor(ShimTaskExecutor[GlueTask]):
    def execute_from_model(self, tt: task_models.TaskTemplate, **kwargs) -> typing.Any:
        # my plugin execution code

# usage in workflow
GlueTask(name="testing", task_config=GlueConfig(crawler_name="test-crawler"), container_image="bla",cache=True, cache_version="1.0")()

That

cache

parameters don't throw errors, but also don't do caching.

Hi everyone 🙌 In my company we are evaluating Flyte 🦜and we want to deploy an MVP to AWS EKS, with s3 and RDS postgres. We are following the guide flyte-the-hard-way but we are stuck in the 05-deploy-with-helm.md. When we try to install flyte the pod throws the following error:

/go/pkg/mod/gorm.io/gorm@v1.24.1-0.20221019064659-5dd2bb482755/gorm.go:206
[error] failed to initialize database, got error failed to connect to `host=***** user=flyteadmin database=flyteadmin`: server error (FATAL: no pg_hba.conf entry for host "172.32.101.12", user "flyteadmin", database "flyteadmin", no encryption (SQLSTATE 28000))

We have run this command to test the database connection:

kubectl run pgsql-postgresql-client --rm --tty -i --restart='Never' --namespace testdb --image <http://docker.io/bitnami/postgresql:11.7.0-debian-10-r9|docker.io/bitnami/postgresql:11.7.0-debian-10-r9> --env='PGPASSWORD=<Password>' --command -- psql testdb --host <RDS-ENDPOINT-NAME> -U flyteadmin -d flyteadmin -p 5432

And works. The only difference is that when we run it asks for the password. Anyone knows what’s happening here?

Hello Is there any way to use partial functions with map tasks, because it gives an error when I run the script on flyte cluster

ValueError: Only Flyte python task types are supported in map tasks currently, received <class 'functools.partial'>

@workflow def n_workflow(branch_name:str, am:str, ame:str, vi:typing.List[str], Equ:str, na:str, fpr:str, ra:typing.List[str], raa:typing.List[str], maa:int, af:str) -> None: abc = data_preparation(branch_name=branch_name, am=am, ame=ame, vi=vi, Equ=Equ, na=na, fpr=fpr, ra=ra, raa=raa, maa=maa, af=af) task_run(abc=abc) In this workflow, while running it through flyte. I was unable to run sequentially. If I launch this "n_workflow" workflow runs successfully for this "data_preparation" task and I need to rerun this "task_run" task separately it to run the next task "task_run". Can you help me to fix this issue?

Hello I'm encountering error while running py script using pyflyte

Encountered error while executing workflow '<http://ss.wf|ss.wf>':
  Error encountered while executing 'wf':
  Failed to convert inputs of task 'ss.split_preprocess':
  [Errno 2] No such file or directory: '/tmp/flyte-2tev7275/raw/b8cd37c42c4279d44bbc8f2b4626f7c3'

I'm curious to get people's opinions here on how to go about building a file-heavy workflow. It's basically a file-based scatter-gather along 2 dimensions. Inputs get split into smaller files, those are processed and then combined based on similar attributes, more processing, and then finally all combined again into a single output file. To make sense of it all, I defined a custom dataclass that has some str attrs and some FlyteFile attrs. However, I discovered that the FlyteFile nested in the custom dataclass doesn't cross task boundaries (the FlyteFile.path attribute of downstream tasks still points to a location upstream where it was defined). Passing the FlyteFile by itself between tasks does not have this issue (the path is correct relative to the downstream pod's context), all else being equal. Then I tried using a tuple with the same attributes as the custom dataclass, but these can't be added to a list (which is what I was using to pass groups of files around). I was getting

Type of Generic List type is not supported, Transformer for type <class 'tuple'> is restricted currently

So now I'm back to using FlyteDirectory but it's.. awkward since there's no metadata about the files and they're given arbitrary names on the backend unless I explicitly name them somehow. I'm wondering how folks might go about dealing with this? Do I handle all the metadata via the FlyteFile names? Pass some sort of metadata object along with every FlyteDirectory? Maybe there's a way around the custom dataclass limitation, as that would be the most elegant solution. Thanks for reading! I appreciate any insight.

Also @Pryce could chime in here - I think Flyte fits well? https://www.reddit.com/r/mlops/comments/13udl02/incorporating_setting_up_mlops_as_a_ds_at_a/?utm_source=share&utm_medium=ios_app&utm_name=ioscss&utm_content=2&utm_term=1

Hello, I’m trying to run the Getting Started wine classification example on my Flyte setup, but have faced several errors. My setup is consisted of: Jupyter Hub + Flyte on the same K8S cluster. The codes are

from flytekit.remote import FlyteRemote
from flytekit.configuration import Config, PlatformConfig, ImageConfig, SerializationSettings

remote = FlyteRemote(
    config=Config(
        platform=PlatformConfig(
            endpoint="dns:///flyte-backend-flyte-binary-grpc.namespace.svc.cluster.local:8089",
            insecure=True,
            insecure_skip_verify=True,
        )
    ),
    default_project="flytesnacks",
    default_domain="development",
    data_upload_location="<s3a://flyte-storage/>"
)

HYPERPARAMS = {
    "hyperparameters": {"C": 0.1}
}

flyte_entity = training_workflow

flyte_workflow = remote.register_workflow(
    entity=flyte_entity,
    serialization_settings=SerializationSettings(image_config=ImageConfig.auto_default_image()),
    version="v1"
)

launch_plan = LaunchPlan.get_or_create(
    workflow=flyte_workflow,
    name="wine_classification_lp",
    default_inputs=HYPERPARAMS,
)
remote.register_launch_plan(launch_plan, "v1", project="flytesnacks", domain="development")

And the workflow failed right at the first task

get_data

, here is the error I’ve got

[1/1] currentAttempt done. Last Error: USER::rapped(*args, **kwargs)                        │
│                                                                              │
│ /usr/local/lib/python3.8/site-packages/flytekit/bin/entrypoint.py:347 in     │
│ _execute_task                                                                │
│                                                                              │
│ ❱ 347 │   │   _task_def = resolver_obj.load_task(loader_args=resolver_args)  │
│                                                                              │
│ /usr/local/lib/python3.8/site-packages/flytekit/core/utils.py:295 in wrapper │
│                                                                              │
│ ❱ 295 │   │   │   │   return func(*args, **kwargs)                           │
│                                                                              │
│ /usr/local/lib/python3.8/site-packages/flytekit/core/python_auto_container.p │
│ y:235 in load_task                                                           │
│                                                                              │
│ ❱ 235 │   │   task_module = importlib.import_module(name=task_module)  # typ │
│                                                                              │
│ /usr/local/lib/python3.8/importlib/__init__.py:127 in import_module          │
│                                                                              │
│ ❱ 127 │   return _bootstrap._gcd_import(name[level:], package, level)        │
│ in _gcd_import:1011                                                          │
│ in _sanity_check:950                                                         │
╰──────────────────────────────────────────────────────────────────────────────╯
ValueError: Empty module name

Any suggestion on which might have caused this error and how to fix it? I’m suspecting that I haven’t registered the workflow or the launch plan properly.