Hi Flyte community, Does using the pod template setting for task registration completely override the pod template that Flyte defaults to or does it add to it? I'd like to add the following setting so my raw container Flyte task pods can refer to the their corresponding settings (based on stage/app):

envFrom:
      - configMapRef:
          name: special-config
      - secretRef:
          name: special-config

https://raw.githubusercontent.com/kubernetes/website/main/content/en/examples/pods/pod-configmap-envFrom.yaml

Hey everyone I'm busy working trying to deploy Flyte to k8s hosted on Hetzner but struggling as existing deployment patterns seem to be geared toward AWS / GCP. Flyte looks awesome but at this point I can't even deploy it well enough to test it in our environment. I've tried using the

flyte-binary

helm chart. Also having trouble working around this: https://github.com/flyteorg/flyte/issues/3769 Could use a hand if anyone has one to spare.

I am currently trying to combine dynamic overrides with ray to dynamically determine the ray cluster size when running a task. But when the task under the dynamic runs, it is considered a python task and never creates a ray cluster.

@dynamic
def request_train_resources_task(
    compressed_dataset: FlyteFile,
    train_parameters: TrainParameters,
) -> TrainOutputs:
    task_config = RayJobConfig(
        head_node_config=HeadNodeConfig(
            ray_start_params={
                "block": "true",
            }
        ),
        worker_node_config=[
            WorkerNodeConfig(
                group_name="ray-group",
                replicas=train_parameters.num_nodes - 1,
            )
        ],
    )
    resources_request = Resources(
        cpu=str(train_parameters.cpus_per_gpu * train_parameters.num_gpus),
        mem=f"{train_parameters.memory_per_gpu * train_parameters.num_gpus}G",
        gpu=str(train_parameters.num_gpus),
        ephemeral_storage=f"{train_parameters.ephemeral_storage}G",
    )
    print(f"task_config: {task_config}")
    print(f"resources_request: {resources_request}")
    return pl_mnist_train_task(
        compressed_dataset=compressed_dataset,
        train_parameters=train_parameters,
    ).with_overrides(
        task_config=task_config,
        requests=resources_request,
    )

Hi, is it possible to send Platform Events to something like external Kafka cluster or is AWS SQS/SNS the only available option right now?

I've deployed flyte using kubeadm along with spark-k8s-operator which run successful this example, did try run the example from flytesnacks and got some issues (logs in 🧵). Someone know about

Non-spark-on-k8s command provided

?

[need help] I'm trying to setup a gpu slurm cluster. this is the last 2 lines in my

/etc/slurm/slurm.conf

NodeName=localhost Gres=gpu:1 CPUs=4 RealMemory=15006 Sockets=1 CoresPerSocket=2 ThreadsPerCore=2 State=UNKNOWN
PartitionName=debug Nodes=ALL Default=YES MaxTime=INFINITE State=UP

this is the

/etc/slurm/gres.conf

AutoDetect=nvml
NodeName=localhost Name=gpu Type=tesla  File=/dev/nvidia0 COREs=0

after changed the config, I restarted my slurm cluster and type

slurmd -C

but it doesn't show that I have gpu. CC @Adrian Loy @Bernd Doser @Buğra Gedik

I have a dynamic task, that has a few chained tasks (using >> operator), when running I get the following error`RuntimeExecutionError: max number of system retry attempts [51/50] exhausted. Last known status message: [system] unable to read futures file, maybe corrupted, caused by: [system] Failed to read futures protobuf file., caused by: path/to/futures.pb: [LIMIT_EXCEEDED] limit exceeded. 22mb > 10mb.` Any thoughts on what might be going on here? Thanks!

Hi all, does Recover on dynamic task work properly, we have a dynamic task that returns a Dict with following

sample_name: sample_workflow_results

. When some sample fails and we recover, unfortunately the output mapped dict is all shuffled up and no longer matches with the sample_name key. For example we expect the certain dynamic task returns

filename -> filename-suffix

map like below

{
	"counter_1":1,
	"counter_2":2,
	"counter_3":3,
	"counter_4":4,
	"counter_5":5,
	"counter_6":6
}

and i forcefully fail some tasks and some successfully completes. Then when i recover i got this instead

{
	"counter_1":5,
	"counter_2":6,
	"counter_3":3,
	"counter_4":4,
	"counter_5":5,
	"counter_6":4
}

example dynamic task is below

@dynamic(cache=True, cache_version=get_workflow_version())
def compute_average_for_chunks(
    chunks: Dict[str, FlyteFile], infile: FlyteFile
) -> Dict[str, int]:
    suffix_dict = {}

    for chunk_name, chunk_path in chunks.items():
        infile_suffix = get_file_suffix_task(chunk_path=chunk_path, infile=infile)
        suffix_dict[chunk_name] = infile_suffix

    return suffix_dict

Hey folks.. so I'm exploring how to set up the CI/CD pipeline now that I have some workflows running locally. In order to do this I was experimenting with the flyte-sandbox docker image to register the new version of the workflows and so on. I'm running into an error though

flytectl get workflows --admin.endpoint <remote_host>:30080 -p <project_name> -d development

is successful

docker run --rm --entrypoint flytectl <http://cr.flyte.org/flyteorg/flyte-sandbox|cr.flyte.org/flyteorg/flyte-sandbox> get workflows --admin.endpoint <remote_host>:30080 -p <project_name> -d development

throws an error

time="2022-06-27T20:18:25Z" level=info msg="[0] Couldn't find a config file []. Relying on env vars and pflags."
{"json":{},"level":"error","msg":"failed to initialize token source provider. Err: failed to fetch auth metadata. Error: rpc error: code = Unavailable desc = connection error: desc = \"transport: authentication handshake failed: tls: first record does not look like a TLS handshake\"","ts":"2022-06-27T20:18:26Z"}
{"json":{},"level":"warning","msg":"Starting an unauthenticated client because: can't create authenticated channel without a TokenSourceProvider","ts":"2022-06-27T20:18:26Z"}
{"json":{},"level":"info","msg":"Initialized Admin client","ts":"2022-06-27T20:18:26Z"}
Error: rpc error: code = Unavailable desc = connection error: desc = "transport: authentication handshake failed: tls: first record does not look like a TLS handshake"
{"json":{},"level":"error","msg":"rpc error: code = Unavailable desc = connection error: desc = \"transport: authentication handshake failed: tls: first record does not look like a TLS handshake\"","ts":"2022-06-27T20:18:27Z"}

Hi, wondering what people are using for viewing/archiving container logs on self managed on prem k8s... did anyone find a nice setup yet, since flyte can't collect and archive task logs for you?