https://flyte.org logo
Join Slack
Powered by
This message was deleted.
# ask-ai
e
This message was deleted.
a
The Flyte's documentation I have ingested does not mention how to automatically rerun all containers in a pod if a specific log output (like "current=3.00000") is not generated by one of the non-primary containers. If you can clarify whether you want this logic to be handled within the Flyte workflow (e.g., via task retries, custom exit codes, or workflow conditionals) or at the Kubernetes pod/container level, it may help in suggesting a Flyte-native approach or a workaround. If this doesn't help, please reach out to the Flyte support team for further assistance. 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
yes, I want to handle by flyte workflow @ancient-wolf-19325
a
The Flyte documentation I have ingested does not mention how to implement logic within a Flyte workflow that checks for a specific log output from a non-primary container and conditionally reruns all containers in the pod if that output is missing. You might want to provide more details about how your containers communicate their status to the Flyte task (for example, by writing to a shared file, setting an output variable, or using exit codes), as Flyte tasks typically determine success or failure based on the primary container's exit status and outputs, not on parsing logs from sidecar containers. If this doesn't help, please reach out to the Flyte support team for further assistance. 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
I want the logic that logs will check once logs found it should fail if that current=3.00000 is not found then I will use retries option will it work @ancient-wolf-19325
a
Yes, if you implement logic in your primary container to check for the presence of "current=3.00000" in the logs (or in a shared file), and have the task fail (e.g., by raising an exception) if it is not found, then Flyte's retries option will work as expected. The task will be retried according to the number of retries you specify in the @task decorator, as long as the failure is a recoverable/system error or you raise a FlyteRecoverableException for user errors (source, source). Would you like a step-by-step example of how to implement this logic in your Flyte task? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
yes, provide code @ancient-wolf-19325
a
It seems likely that you can implement this by having your primary container check for the log entry (e.g., by reading a shared file or log), and raise an exception if "current=3.00000" is not found. Then, use the retries parameter in your Flyte task to enable automatic retries. Here’s a simplified example:
Copy code
from flytekit import task, workflow

@task(retries=3)
def check_log_and_fail_if_missing(log_path: str) -> None:
    with open(log_path, "r") as f:
        logs = f.read()
    if "current=3.00000" not in logs:
        raise Exception("Required log entry not found: current=3.00000")
    # Otherwise, task succeeds

@workflow
def my_workflow(log_path: str):
    check_log_and_fail_if_missing(log_path=log_path)
You would need to ensure the log file is accessible to the primary container (e.g., via a shared volume). If the log entry is missing, the task fails and Flyte retries it up to 3 times as specified (source). Would you like more detail on handling shared volumes or customizing the pod spec? Sources: - https://www.union.ai/docs/v2/flyte/user-guide/task-configuration/retries-and-timeouts/ 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
update the code sys error if it not found 3.00000 @ancient-wolf-19325
a
It seems likely that you can raise a system error by using a custom exception or by simply raising a standard Exception, which Flyte will treat as a system error and trigger system-level retries if configured. Here’s an updated example:
Copy code
from flytekit import task, workflow

@task(retries=3)
def check_log_and_fail_if_missing(log_path: str) -> None:
    with open(log_path, "r") as f:
        logs = f.read()
    if "current=3.00000" not in logs:
        # Raise a standard Exception, which Flyte treats as a system error
        raise Exception("SYSTEM ERROR: Required log entry not found: current=3.00000")
    # Otherwise, task succeeds

@workflow
def my_workflow(log_path: str):
    check_log_and_fail_if_missing(log_path=log_path)
My understanding is that raising Exception will be treated as a system error by Flyte, and system-level retries (if configured in the platform) will be triggered (source). Would you like more detail on configuring system-level retries? Sources: - https://docs.flyte.org/en/latest/api/flytekit/design/authoring.html#exception-handling 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
if it fails will it restarts @ancient-wolf-19325
a
Yes, if the task fails (for example, because "current=3.00000" is not found and an Exception is raised), Flyte will automatically restart (retry) the task up to the number of retries you specify in the @task decorator (e.g., retries=3). Each retry will rerun the entire task in a new pod (source, source). Would you like more detail on how retries work or how to configure system-level retries? Sources: - Retries and timeouts | Union.ai Docs - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
If you have set 
retries=3
in your Flyte task but retries are not happening when the log check fails, the most likely reason is that Flyte only retries tasks for certain types of failures. Specifically, Flyte will only retry on: - System errors (e.g., infrastructure/network issues) - User errors that are explicitly marked as recoverable by raising 
FlyteRecoverableException
in Python tasks If your container's entrypoint script exits with a nonzero exit code (as in your script: 
exit $RESULT
), Flyte treats this as a user error, but unless you are using a Python task and raise 
FlyteRecoverableException
, it may not count toward the user retry budget. For container tasks, Flyte generally retries on system errors, not on all user-level failures (source, source). Possible reasons for no retry: - The failure is not classified as recoverable/system by Flyte. - The platform/system retry budget is not configured or is exhausted. - The container task type may not honor user retry settings for non-recoverable errors. What you can do: - Ensure your task type supports user retries for nonzero exit codes. - For Python tasks, raise 
FlyteRecoverableException
for retryable errors. - For container tasks, check Flyte platform/system retry settings. Would you like a step-by-step breakdown of how Flyte classifies errors and when retries are triggered? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
Then I want to set it customly so that it should retry @ancient-wolf-19325
a
For Flyte container tasks, you cannot directly control error classification from the shell script—Flyte will only retry on system errors by default, and nonzero exit codes from your entrypoint are treated as user errors, which are not retried unless the platform is configured for unified retry budgets (source, source). To force retries for your use case, you can: - Ask your Flyte administrator to enable the "simplified retry behavior" (
ignore-retry-cause: true
in FlytePropeller config), which counts all failures (user and system) toward the retry budget (source). - Alternatively, wrap your logic in a Python Flyte task and raise 
FlyteRecoverableException
on failure, which will trigger retries as expected. Would you like a code example for the Python approach or more detail on platform configuration? Sources: - Build indestructible pipelines with Flyte - Simplify retry behaviour RFC 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
just make it retry when it didn't found 6.00000
just make it retry when it didn't found 6.00000 @ancient-wolf-19325
a
To ensure Flyte retries when "6.000000" is not found, you must make your task fail with a retryable error. For Python tasks, raise a FlyteRecoverableException if the log check fails; for container tasks, you need platform support for unified retry budgets (
ignore-retry-cause: true
) or ensure the failure is classified as a system error (source, source). Python task example:
Copy code
from flytekit import task, workflow
from flytekit.exceptions.user import FlyteRecoverableException

@task(retries=3)
def check_log(log_path: str):
    with open(log_path) as f:
        if "6.000000" not in f.read():
            raise FlyteRecoverableException("6.000000 not found, retrying...")

@workflow
def wf(log_path: str):
    check_log(log_path=log_path)
Would you like a container task workaround or more detail on platform settings? Sources: - Build indestructible pipelines with Flyte - https://github.com/flyteorg/flyte/blob/master/rfc/system/3902-simplify-retry-behaviour.md 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
FlyteRecoverableException("6.000000 not found, retrying...") NameError: name 'FlyteRecoverableException' is not defined @ancient-wolf-19325
a
You need to import FlyteRecoverableException from flytekit.exceptions.user. Here is the corrected code:
Copy code
from flytekit import task, workflow
from flytekit.exceptions.user import FlyteRecoverableException

@task(retries=3)
def check_log(log_path: str):
    with open(log_path) as f:
        if "6.000000" not in f.read():
            raise FlyteRecoverableException("6.000000 not found, retrying...")

@workflow
def wf(log_path: str):
    check_log(log_path=log_path)
As per flytekit.exceptions.user.py, FlyteRecoverableException must be imported from flytekit.exceptions.user. Would you like more detail on exception handling in Flyte? Sources: - flytekit.exceptions.user.py 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
def check_for_success(): log_path="/dev/shm/njest.log" with open(log_path, "r") as f: logs = f.read() if "Current=6.00000" not in logs: #raise Exception("SYSTEM ERROR: Required log entry not found: current=3.00000", file=sys.stderr) #raise Exception("Required log entry not found: current=6.00000", file=sys.stderr) raise FlyteRecoverableException("6.000000 not found, retrying...") File "/root/recompute/src/tasks/recompute_task.py", line 669, in recompute error_desc=str(e.description), AttributeError: 'FlyteRecoverableException' object has no attribute 'description' Message: AttributeError: 'FlyteRecoverableException' object has no attribute 'description' @ancient-wolf-19325
a
The error AttributeError: 'FlyteRecoverableException' object has no attribute 'description' occurs because FlyteRecoverableException does not have a description attribute—only the message you pass to it (source). You should use str(e) to get the error message, not e.description. Example fix:
Copy code
try:
    check_for_success()
except FlyteRecoverableException as e:
    error_desc = str(e)  # Not e.description
Would you like more detail on Flyte exception handling? Sources: - flytekit.exceptions.user.py 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
logger = logging.getLogger(name) def check_for_success(): log_path="/dev/shm/njest.log" with open(log_path, "r") as f: logs = f.read() if "Current=6.00000" not in logs: #raise Exception("SYSTEM ERROR: Required log entry not found: current=3.00000", file=sys.stderr) #raise Exception("Required log entry not found: current=6.00000", file=sys.stderr) raise FlyteRecoverableException("6.000000 not found, retrying...") def dynamic_pod_spec_recompute( claim_name: str, execution_id: str, enable_aurix: bool, adaptive_image: Optional[str], njest_image: Optional[str], recompute_image: Optional[str], aurix_image: Optional[str], ) -> V1PodSpec: """Dynamic_pod_spec_recompute function is used to set the dynamic pvc claim_name using with_overrides function.""" containers = [ V1Container( name="primary", image_pull_policy="Always", ), V1Container( name="recompute", image=recompute_image, security_context=V1SecurityContext(privileged=True), resources=V1ResourceRequirements( requests={"cpu": "26", "memory": "40Gi"}, limits={"cpu": "28", "memory": "40Gi"} ), volume_mounts=[ V1VolumeMount(name="gcs-volume", mount_path="/input/meas", sub_path="chopped-data"), V1VolumeMount(name="gcs-volume", mount_path="/logs", sub_path="logs"), ], command=["/bin/bash"], args=[ "-c", """ echo '#!/usr/bin/env bash sudo ip link add recomp_br0 type bridge sudo ip link set up recomp_br0 sudo ip link add link recomp_br0 name vlan10 type vlan id 10 sudo ip link set up vlan10 ip addr add 10.10.0.89/16 dev vlan10 ip addr add 10.10.1.6/16 dev vlan10 ip addr add 10.10.1.10/16 dev vlan10 ip addr add 10.10.1.34/16 dev vlan10 ip addr add 10.10.1.90/16 dev vlan10 ip addr add 10.10.1.91/16 dev vlan10 ip route add 239.10.0.2 dev vlan10 scope link sudo ip link add link recomp_br0 name vlan20 type vlan id 20 sudo ip link set up vlan20 ip addr add 10.20.0.89/16 dev vlan20 ip addr add 10.20.1.10/16 dev vlan20 ip addr add 10.20.1.15/16 dev vlan20 ip addr add 10.20.1.97/16 dev vlan20 ip route add 239.20.0.2 dev vlan20 scope link sudo ip link add link recomp_br0 name vlan21 type vlan id 21 sudo ip link set up vlan21 ip addr add 10.21.1.101/16 dev vlan21 ip addr add 10.21.1.124/16 dev vlan21 ip addr add 10.21.1.125/16 dev vlan21 ip addr add 10.21.1.126/16 dev vlan21 ip addr add 10.21.1.127/16 dev vlan21 ip route add 239.21.0.2 dev vlan21 scope link' > /recompute/install/setup_network.sh && chmod +x /recompute/install/setup_network.sh && export DEB_PKG='touch /usr/adaptive/setup_fsi_ccplex_com.sh && chmod +x /usr/adaptive/setup_fsi_ccplex_com.sh && /usr/adaptive/network_setup.sh -i vethAV0 && until [ -f /dev/shm/api_started ]; do sleep 1; done && sleep 10 && touch /dev/shm/adaptive_started' && echo '#!usr/bin/env bash #!/bin/bash # Set ulimit for core dumps ulimit -c unlimited # Set core dump pattern echo '/tmp/dumpfiles/core-recompute-%e.%p.%h.%t' > /proc/sys/kernel/core_pattern # Wait for adaptive_started file until [ -f /dev/shm/adaptive_started ]; do sleep 1 done # Navigate to recompute install directory and setup network cd /recompute/install ./setup_network.sh # Signal recompute started touch /dev/shm/recompute_started # Display network interfaces ip a NV_BIN_ARGS="" if [[ "$NV_BIN_CONVERSION" == "true" ]]; then echo "Running Recompute with NV Bin File" NV_BIN_ARGS="--nv_bin_file /input/meas/nv/roadcast_debug.bin --nv_config_file nv_config/mapping_apis.json" else echo "Running Recompute without NV Bin File" fi # Start recompute service and tcpdump LD_LIBRARY_PATH=$(pwd)/lib:$LD_LIBRARY_PATH bin/recompute --parquet_measurement_path /input/meas/rov --someip_config_path config/ $NV_BIN_ARGS & recompute_pid=$! tcpdump -i recomp_br0 -s0 -v port 40000 or port 30490 -w /logs/rov.pcap & tcpdump_pid=$! # if [[ "$AURIX_ENABLED" == "true" ]]; then # tcpdump -i vlan10 -i vlan20 -w /logs/rov.pcap & # tcpdump_pid=$! # else # tcpdump -i eth0 -s0 -v port 40000 or port 30490 -w /logs/rov.pcap & # tcpdump_pid=$! # fi # Wait for recompute service to finish wait $recompute_pid exit_code=$? # Kill tcpdump process kill $tcpdump_pid 2>/dev/null # Signal recompute ended touch /dev/recompute_ended # Exit with recompute service exit code exit $exit_code' > /recompute/install/bin/recompute_entrypoint.sh && apt-get install -y tcpdump && chmod +x /recompute/install/bin/recompute_entrypoint.sh && sudo taskset -c 12-19 /recompute/install/bin/recompute_entrypoint.sh && touch /dev/shm/recompute_ended """, ], ), ] if enable_aurix: containers.append( V1Container( name="aurix", image=aurix_image, security_context=V1SecurityContext(privileged=True), volume_mounts=[ V1VolumeMount(name="gcs-volume", mount_path="/home/root/logs", sub_path="logs"), ], command=["/bin/bash"], args=[ "-c", """ echo '#!/bin/sh cd /home/opensutloader ./opensutloaderd --logfile /home/root/logs/loader_trace.log' > /home/opensutloader/simulation.sh && ip netns add aurixa && \ ip link add vethAurix0 type veth peer name vethAurix1 && \ ip link set vethAurix0 netns aurixa && \ ip link set vethAurix1 master recomp_br0 && \ ip link set vethAurix1 up && \ ip netns exec aurixa ip link set dev lo up && \ ip netns exec aurixa ip link set dev vethAurix0 up && cd /home/opensutloader && chmod -R 777 /home/root/logs && chmod -R a+rx /home/opensutloader/simulation.sh && sed -i 's/^PcapOutputFilePath=.*/PcapOutputFilePath=\\/home\\/root\\/logs\\/aurix_trace.pcap/' /home/opensutloader/opensutloader.ini && sed -i '/^SimulationTime=/d' /home/opensutloader/opensutloader.ini && sed -i '2i SimulationTime=1000' /home/opensutloader/opensutloader.ini && sed -i 's/^DeviceName=.*/DeviceName=vethAurix0/' /home/opensutloader/opensutloader.ini && until [ -f /dev/shm/adaptive_started ]; do sleep 1 done export -p > /tmp/vars_file sudo ip netns exec aurixa bash -c "ls -l /proc/self/ns/net && chmod +x /home/opensutloader/simulation.sh && cd /home/opensutloader && source /tmp/vars_file && echo $PWD && cat /home/opensutloader/simulation.sh && taskset -c 0 /home/opensutloader/simulation.sh" & until [ -f /dev/shm/recompute_ended ]; do sleep 1 done exit """, ], ), ) pod_spec = V1PodSpec( image_pull_secrets=[V1LocalObjectReference(name="te13535-artifactory-cred")], node_selector=RecomputeConfig.NODE_SELECTOR_PODSPEC, host_ipc=True, restart_policy="OnFailure", active_deadline_seconds=RecomputeConfig.GKE_POD_ACTIVE_DEADLINE_SECONDS, service_account=RecomputeConfig.SERVICE_ACCOUNT, service_account_name=RecomputeConfig.SERVICE_ACCOUNT, containers=containers, volumes=[ V1Volume( name="gcs-volume", persistent_volume_claim=V1PersistentVolumeClaimVolumeSource(claim_name=claim_name), ), # V1Volume(name="recompute-config", config_map=V1ConfigMapVolumeSource(name="recompute-config-v4")), V1Volume( name="gitlab-secret", secret=V1SecretVolumeSource(secret_name=RecomputeConfig.GITLAB_TOKEN), ), V1Volume( name="gitlab-certificate", secret=V1SecretVolumeSource(secret_name=RecomputeConfig.GITLAB_CERTIFICATES), ), V1Volume( name="artifact-certs", secret=V1SecretVolumeSource(secret_name=RecomputeConfig.ARTIFACT_REPO_CERTS), ), V1Volume( name="artifact-secret", secret=V1SecretVolumeSource(secret_name=RecomputeConfig.ARTIFACT_REPO_TOKEN), ), ], ) return pod_spec @task( task_config=Pod( annotations={"gke-gcsfuse/volumes": "true"}, pod_spec=dynamic_pod_spec_recompute( claim_name="adasrecompute", execution_id="mb007", enable_aurix=False, njest_image="njest:v1", adaptive_image="adaptive:v1", recompute_image="recompute:v1", aurix_image="aurix:v1", ), ), container_image=RecomputeConfig.RECOMPUTE_WF_IMAGE, retries=RecomputeConfig.TASK_RETRIES, secret_requests=[Secret(group=ApiHandler.SECRET_GROUP, key=ApiHandler.SECRET_KEY)], ) def recompute(job_id: str) -> str: """Recompute is performed with the input bin and input parquet files.""" pcap = RecomputeConfig.PCAP_PATH try: #time.sleep(300) subprocess.run("until [ -f /dev/shm/timeout_done ]; do sleep 1; done; sleep 60", shell=True) check_for_success() update_status(job_id=job_id, current_task="RECOMPUTE_TASK") mt_pcap = RecomputeConfig.OUTPUT_MT_PCAP_PATH rov_pcap = RecomputeConfig.OUTPUT_ROV_PCAP_PATH bucket_name = WorkflowUtils.execution_id() files = [mt_pcap, rov_pcap] missing_file = WorkflowUtils.check_files_exist(bucket_name, files) if not missing_file: update_status(job_id=job_id, current_task="RECOMPUTE_TASK", task_state=JobState.COMPLETE) pcap = RecomputeConfig.PCAP_PATH else: raise CustomError( description=f"PCAPs NOT Generated. Missing the file: {', '.join(missing_file)}", error_code=ErrorCode.RECOMPUTE_FILE_MISSING, ) return pcap except Exception as e: logger.exception("Error in RECOMPUTE_TASK") update_status( job_id=job_id, current_task="RECOMPUTE_TASK", task_state=JobState.ERROR, error_desc=str(e.description), error_code=str(e.value), ) raise can you check why retry is not happening? eventhough side container fails? @ancient-wolf-19325
a
Based on the provided code and Flyte's documented behavior, the reason retries are not happening when the sidecar (non-primary) container fails is: Flyte only considers the primary container's exit code to determine task success or failure. If your sidecar container fails but the primary container exits successfully (exit code 0), Flyte will treat the task as successful and will not trigger a retry, regardless of what happens in the sidecar (source). In your setup: - The check_for_success() function runs in the primary container and raises a FlyteRecoverableException if "Current=6.00000" is not found in the log, which should trigger a retry if the exception is raised. - However, if the sidecar container fails but the primary container does not detect this failure (e.g., the log file is still present or the check passes), the task will not fail and thus will not be retried. To ensure retries: - The primary container must explicitly check for the sidecar's success (e.g., by monitoring its process, exit code, or output files/logs). - If the sidecar fails, the primary container must raise an exception (preferably FlyteRecoverableException for user retries) so Flyte can trigger a retry. Summary: Flyte will only retry if the primary container fails (nonzero exit code or uncaught exception in the Python task). Sidecar/container failures are ignored unless the primary container detects and reports them as a failure. Would you like a code example for monitoring the sidecar from the primary container? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
sorry sorry, primary fails also retrying not happen. I set 3 in retry option @ancient-wolf-19325
a
If your primary container fails (raises an exception or exits with a nonzero code) and Flyte still does not retry the task, the most likely reasons are: 1. The error is not classified as recoverable or system error. By default, Flyte only retries on system errors or user errors that are subclasses of FlyteRecoverableException (source, source). 2. Platform/system retry budget is exhausted or not configured. If system retries are not enabled or the maximum is reached, retries will not occur. 3. If you are raising a custom exception (not FlyteRecoverableException), Flyte treats it as non-recoverable and will not retry. To ensure retries: - For Python tasks, raise FlyteRecoverableException for retryable errors. - For container tasks, ensure the process exits with a nonzero code and the platform is configured for system retries or unified retry budgets (
ignore-retry-cause: true
). Would you like a step-by-step checklist to debug why retries are not happening in your case? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
yes pleas @ancient-wolf-19325
a
Here’s a step-by-step checklist to debug why Flyte retries are not happening for your primary container failure: 1. Check Exception Type (Python tasks): - Make sure you are raising 
FlyteRecoverableException
for retryable errors. Standard exceptions (like 
Exception
) are not retried unless platform is configured for unified retry budgets (source). 2. Check Task Decorator: - Confirm that you set 
retries=3
in your 
@task
decorator. 3. Check Platform Retry Configuration: - For container tasks, system retries are controlled by platform config (e.g., 
max-node-retries-system-failures
). If not set, only user retries apply (source). - If you want all failures to count toward the retry budget, ask your admin to set 
ignore-retry-cause: true
in FlytePropeller config (source). 4. Check Task Exit Code (Container tasks): - Ensure your primary container exits with a nonzero code on failure. Flyte only retries if the primary container fails. 5. Check Flyte Version and Logs: - Some retry bugs have been fixed in recent Flyte versions. Check the Flyte UI or logs for error classification and retry attempts. 6. Check for Workflow/Task Overrides: - Make sure there are no workflow-level or launch plan overrides that set retries to 0 or override your task’s retry policy. Would you like more detail on any of these steps? Sources: - Build indestructible pipelines with Flyte - Simplify retry behaviour RFC 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
4. Check Task Exit Code (Container tasks): - Ensure your primary container exits with a nonzero code on failure. Flyte only retries if the primary container fails. can you update the code according to this @ancient-wolf-19325
a
It seems likely that to ensure Flyte retries your container task, your primary container must exit with a nonzero exit code when the required log entry is missing. In your shell script (the entrypoint for the primary container), after checking for "Current=6.00000", you should use exit 1 if not found. Here’s a minimal example:
Copy code
#!/bin/bash

log_path="/dev/shm/njest.log"
if ! grep -q "Current=6.00000" "$log_path"; then
    echo "Current=6.00000 not found, failing for retry"
    exit 1  # Nonzero exit code triggers Flyte retry
fi

# Continue with the rest of your logic if found
This ensures that Flyte detects the failure and, if retries are configured (e.g., 
retries=3
), will rerun the task (source). Would you like a full example of integrating this into your Flyte task or pod spec? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
def check_for_success(): log_path="/dev/shm/njest.log" with open(log_path, "r") as f: logs = f.read() if "Current=6.00000" not in logs: #raise Exception("SYSTEM ERROR: Required log entry not found: current=3.00000", file=sys.stderr) #raise Exception("Required log entry not found: current=6.00000", file=sys.stderr) raise FlyteRecoverableException("6.000000 not found, retrying...") i am using this one right inorder to check @ancient-wolf-19325
a
Yes, your Python function is correct for checking if "Current=6.00000" is present in the log file, and raising FlyteRecoverableException if not found. This will trigger Flyte retries if you set 
retries
in your 
@task
decorator (source, source). Would you like a full Flyte task example using this function? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
then why retry is not happening? can you check which one to use python code or shell script which you provided? @ancient-wolf-19325
a
If you are using a Flyte Python task (i.e., your task logic is in Python and decorated with @task), your Python code with FlyteRecoverableException is correct and should trigger retries if retries is set in the decorator. If retries are not happening, possible reasons include: the exception is not being raised, the log file is always present, or there is a misconfiguration in the task or platform (source). If you are using a Flyte container task (i.e., your task runs a shell script as the entrypoint), you must ensure the shell script exits with a nonzero code (e.g., exit 1) when the log check fails. Flyte only triggers retries for container tasks if the main process exits with a nonzero code (source). Summary: - Use the Python code with FlyteRecoverableException for Python tasks. - Use the shell script with exit 1 for container tasks. Would you like help determining which task type you are using or how to configure retries for your specific case? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
yes please @ancient-wolf-19325
a
To determine which approach to use and ensure retries work: 1. If you are using a Flyte Python task: - Your task function is decorated with @task and runs Python code directly. - Use the Python code with FlyteRecoverableException:
Copy code
from flytekit import task, workflow
     from flytekit.exceptions.user import FlyteRecoverableException

     @task(retries=3)
     def check_for_success():
         log_path = "/dev/shm/njest.log"
         with open(log_path, "r") as f:
             logs = f.read()
         if "Current=6.00000" not in logs:
             raise FlyteRecoverableException("6.00000 not found, retrying...")
- This will trigger retries if the exception is raised (source). 2. If you are using a Flyte container task: - Your task runs a shell script or command as the main entrypoint (not Python code). - You must ensure the shell script exits with a nonzero code if the log check fails:
Copy code
#!/bin/bash
     log_path="/dev/shm/njest.log"
     if ! grep -q "Current=6.00000" "$log_path"; then
         echo "Current=6.00000 not found, failing for retry"
         exit 1
     fi
     # Continue with the rest of your logic if found
- Flyte will only retry if the main process exits with a nonzero code (source). How to check your task type: - If your Flyte task is defined with @task and contains Python code, use the Python approach. - If your Flyte task is configured with a Pod or Container and runs a shell command as entrypoint, use the shell script approach. Would you like help identifying your task type in your Flyte project? Sources: - Build indestructible pipelines with Flyte 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
RPC Failed, with Status: StatusCode.INTERNAL Details: json: cannot unmarshal array into Go struct field Container.containers.args of type string @ancient-wolf-19325 what is this problem
a
This error means that in your Flyte container or pod specification, the args field is expected to be a string, but you provided an array (list) instead. Flyte’s Go backend expects args as a string, not a list, in some contexts (source). How to fix: - If you are using a custom pod/container spec, ensure that args is a string, not a list/array. - For example, use:
Copy code
args="-c 'echo hello'"
instead of:
Copy code
args=["-c", "echo hello"]
Would you like more detail on how to structure your container spec? Sources: - https://github.com/flyteorg/flyte/issues/4985 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
NJEST_COMMAND=[ "-c", """ cat <<EOF > /home/initial_value_test.yml TestSteps: 1: WaitFor: Timeout: 200 Variables: dwVehicleIOQMState.ignitionStatus_in: 3 2: Wait: 1200 EOF export TEST_SUITES="TestSuite: - /home/initial_value_test.yml" cd /home/julia/shared_memory_app/ && touch /dev/shm/api_started && MBTestCaseParser --testcase /home/initial_value_test.yml 2>&1 | tee -a /dev/shm/njest.log & until [ -f /dev/shm/recompute_ended ]; do sleep 1 done exit """, ], can you change this to above one you mentioned? @ancient-wolf-19325
a
It seems likely that you need to provide the args as a single string, not a list of two elements, to avoid the "json: cannot unmarshal array into Go struct field Container.containers.args of type string" error. Here’s how you can update your NJEST_COMMAND:
Copy code
NJEST_COMMAND = """
cat <<EOF > /home/initial_value_test.yml
TestSteps:
  1:
    WaitFor:
      Timeout: 200
      Variables:
        dwVehicleIOQMState.ignitionStatus_in: 3
  2:
    Wait: 1200
EOF
export TEST_SUITES="TestSuite:
- /home/initial_value_test.yml"
cd /home/julia/shared_memory_app/ && touch /dev/shm/api_started  && MBTestCaseParser --testcase /home/initial_value_test.yml 2>&1 | tee -a /dev/shm/njest.log &

until [ -f /dev/shm/recompute_ended ]; do
sleep 1
done

exit
"""

# In your container spec:
args = f'-c \'{NJEST_COMMAND}\''
Or, if you are using a Kubernetes Python client object (like V1Container), set:
Copy code
args = ["-c", NJEST_COMMAND]
only if the Flyte version and plugin support an array. Otherwise, use a single string as above. If you still get the error, use:
Copy code
args = "-c '<your full script here>'"
as a single string. Would you like a full example for your Flyte pod/container spec? Sources: - https://github.com/flyteorg/flyte/blob/master/docs/user_guide/extending/container_interface.md - https://www.union.ai/docs/v2/flyte/_static/public/llms-full.txt Note: I wasn’t able to find highly relevant data sources, but above are a few potentially relevant links. 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
args=TaskCommands.NJEST_COMMAND, i am mentioning like this @ancient-wolf-19325
a
It seems likely that if you are using 
args=TaskCommands.NJEST_COMMAND,
and 
NJEST_COMMAND
is a Python list (e.g., 
<https://github.com/flyteorg/flyte/issues/4985|"-c", "your_script_here"]
), you may encounter the "json: cannot unmarshal array into Go struct field Container.containers.args of type string" error if the Flyte backend or your pod spec expects a string, not a list ([source>). How to fix: - Change 
NJEST_COMMAND
to a single string, for example:
Copy code
NJEST_COMMAND = "-c 'your full shell script here'"
and then use:
Copy code
args=NJEST_COMMAND,
- Or, if your Flyte plugin supports a list, use:
Copy code
args=["-c", "your full shell script here"]
but only if the backend expects an array. If you are still seeing the error, use the single string approach. Would you like a concrete example for your use case? Sources: - https://github.com/flyteorg/flyte/issues/4985 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
a
@ancient-wolf-19325 /bin/bash: line 20: warning: here-document at line 1 delimited by end-of-file (wanted `EOF') this is i am using NJEST_COMMAND = """ cat <<EOF > /home/initial_value_test.yml TestSteps: 1: WaitFor: Timeout: 200 Variables: dwVehicleIOQMState.ignitionStatus_in: 3 2: Wait: 1200 EOF export TEST_SUITES="TestSuite: - /home/initial_value_test.yml" cd /home/julia/shared_memory_app/ && touch /dev/shm/api_started && MBTestCaseParser --testcase /home/initial_value_test.yml 2>&1 | tee -a /dev/shm/njest.log & until [ -f /dev/shm/recompute_ended ]; do sleep 1 done exit """ V1Container( name="njest", image=njest_image, security_context=V1SecurityContext(privileged=True), env=[ V1EnvVar(name="PYTHONUNBUFFERED", value="1"), ], image_pull_policy="Always", command=["/bin/bash"], args=["-c", TaskCommands.NJEST_COMMAND], ) @ancient-wolf-19325
a
The Flyte's documentation I have ingested does not mention how to resolve here-document (
<<EOF ... EOF
) issues in shell scripts passed as Python multi-line strings to Kubernetes containers, but your error 
/bin/bash: line 20: warning: here-document at line 1 delimited by end-of-file (wanted 'EOF')
means that the shell never saw the closing 
EOF
marker, likely due to indentation or whitespace issues in your Python string. To help further, please ensure your here-document markers (
EOF
) are at the very start of the line with no leading spaces, and that the multi-line string is not indented in Python. If possible, paste your actual rendered script (as it appears in the container) to check for whitespace or encoding issues. If this doesn't help, please reach out to the Flyte support team for further assistance. 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.
9 Views
Open in Slack
PreviousNext
Powered by