Is there a way to enforce sequential run for dynamic workflow?

Hi, I am trying to mount a path from the host node to the task container using pod templates. For that I created a pod template on kubernetes using the following yaml:

apiVersion: v1
kind: PodTemplate
metadata:
  name: my-pod-template
  namespace: flyte
template:
  metadata:
    labels:
      pod-template-name: my-pod-template
      pod-template-namespace: flyte
  spec:
    containers:
      - name: default
        image: <http://docker.io/rwgrim/docker-noop|docker.io/rwgrim/docker-noop>
        volumeMounts:
          - name: mnt-volume
            mountPath: /mnt
    volumes:
      - name: mnt-volume
        hostPath:
          path: /mnt

And trying to reference this pod template on the client side with:

@task(pod_template_name="my-pod-template")
def print_files() -> None:
    print(os.listdir("/mnt"))

But the task immediately errors out with PodTemplate does not exist My flyte deployment is on version 1.3 and flytekit on version 1.2.11. Is there anything I am missing or do I need different flyte versions? I was trying the same using PodSpec (flytekitplugins-pod) which worked fine.

Hey all. I'm back with a quick question. I have a situation where a workflow fans out to bunch of nodes, 100+, at some point we're hitting the image pull qps limit and since it takes too long for the pod to start, flyte marks the node as failed and nukes the workflow. So my question is, is there a config I can set to have flyte wait longer for the pod to spin up before giving up?

Hi all, I have a task where I'd like to return a heterogenous dict (doesn't have to be dict, it can be anything - so far I've tried tuple, namedtuple, dataclass, and dict - haven't been able to get any to work unfortunately)

from flytekit import task
from flytekit.types.file import FlyteFile


@task
def create_map_inputs(
    paths: list[str],
    spec: dict,
    gcs_output_dir: str,
) -> list[dict]:
    map_inputs = []
    for path in paths:
        map_inputs.append(
            {
                "bam_file": FlyteFile(path),
                "spec": spec,
                "gcs_output_dir": gcs_output_dir,
            }
        )
    return map_inputs


create_map_inputs(paths=["test.bam"], spec={}, gcs_output_dir="")

The task looks like the following and will be used as input into a map task. Running the above gives the following error:

raise TypeError(
TypeError: Failed to convert return value for var o0 for function create_map_inputs with error <class 'TypeError'>: Object of type FlyteFile is not JSON serializable

Ideally this would return list[dataclass], but I've found that dataclass does not work for my use case (see issue), tuple does not work (Type of Generic List type is not supported, Transformer for type <class 'tuple'> is restricted currently), and any other thing I've tried doesn't work. I have found a way around this, but it's incredibly hacky. Any thoughts?

Has anyone seen an error like this when starting a task?

Request failed with status code 500 failed to create workflow in propeller create not allowed while custom resource definition is terminating

Hi community, is there a way to use gpus in the sandbox cluster?

Hello people, as

pyflyte run --remote

"cannot import tasks and workflows from other files currently" I tried going the way of "use

pyflyte register

to handle those". (quotations taking from the --help output of pyflyte run) With my current setup:

workflow/
|- ClassTest.py
|- example.py

And example.py using ClassTest.py like the following:

from ClassTest import TestClass
from flytekit import task, workflow

@workflow
def wf():
    test_class = TestClass()
    test_class.do_stuff()

I get

ModuleNotFoundError: No module named 'ClassTest'

. What am I missing?

Hi all, Has anyone faced issues with

datacatalog

when task caching is enabled? I'm seeing

Failed to retrieve artifact for get artifact request dataset

because of this error

err: missing entity of type Tag with identifier

using datacatalog version: cr.flyte.org/flyteorg/datacatalog-release:v1.3.0 Any insights into what the missing tag could mean would be great. Thanks! 🙏

Hello all Can you please help me with the below questions pertaining to the

task_executions

table in flyte

AdminDb

? 1. Can you please let me know when are

started_at

and

duration

set to

null

? 2. I am of the impression that {

created_at

and

task_execution_created_at

}, {

updated_at

and

task_execution_updated_at

} should be the same. Can you please confirm this? cc @Yee @Alex Pozimenko (FYI)

Dear all, I ran into an inconsistency between flytectl and FlyteRemote which has been reported before. When trying to create a FlyteRemote with this config

admin:

endpoint: dns:///A.B.C.D:PPPPP

insecure: false

authType: Pkce

insecureSkipVerify: true

flytectl works just fine. But when I try to fetch an execution on a FlyteRemote with this code

project = "myproject"

domain = "development"

execution = "a5cphsfgc57nt6nxbknt"

flyte_config_file = "flyte.config.yaml"

remote = FlyteRemote(config=Config.auto(config_file=flyte_config_file))

flyte_workflow_execution = remote.fetch_execution(project=project, domain=domain, name=execution)

I get the following error

Traceback (most recent call last):
  File "debug_remote.py", line 12, in <module>
    flyte_workflow_execution = remote.fetch_execution(project=project, domain=domain, name=execution)
  File "/opt/micromamba/envs/OHLI/lib/python3.8/site-packages/flytekit/remote/remote.py", line 353, in fetch_execution
    self.client.get_execution(
  File "/opt/micromamba/envs/OHLI/lib/python3.8/site-packages/flytekit/clients/friendly.py", line 582, in get_execution
    super(SynchronousFlyteClient, self).get_execution(
  File "/opt/micromamba/envs/OHLI/lib/python3.8/site-packages/flytekit/clients/raw.py", line 43, in handler
    return fn(*args, **kwargs)
  File "/opt/micromamba/envs/OHLI/lib/python3.8/site-packages/flytekit/clients/raw.py", line 651, in get_execution
    return self._stub.GetExecution(get_object_request, metadata=self._metadata)
  File "/opt/micromamba/envs/OHLI/lib/python3.8/site-packages/grpc/_channel.py", line 1030, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/opt/micromamba/envs/OHLI/lib/python3.8/site-packages/grpc/_channel.py", line 910, in _end_unary_response_blocking
    raise _InactiveRpcError(state)  # pytype: disable=not-instantiable
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.UNAVAILABLE
        details = "failed to connect to all addresses; last error: UNKNOWN: ipv4:A.B.C.D:PPPPP: Peer name A.B.C.D is not in peer certificate"
        debug_error_string = "UNKNOWN:failed to connect to all addresses; last error: UNKNOWN: ipv4:A.B.C.D:PPPPP: Peer name A.B.C.D is not in peer certificate {created_time:"2023-05-18T14:40:11.018710903+01:00", grpc_status:14}"
>

The cluster is running the flyte-binary Helm chart in version 1.3.0 and I tried flytekit 1.2.11, 1.3.0, and 1.6.1, all resulting in the same error message.