https://outerbounds.com/ logo
Join Slack
Powered by
Hello, I have an issue I'm running with timeouts. ...
# ask-metaflow
f
Hello, I have an issue I'm running with timeouts. I've noticed that timeouts don't seem to apply (at least not immediately) to s3 transient failures when using the s3.get_many api. In the attached screenshots, you can see my workers are going far above the 30 minute timeout that is applied to them. When using s3.get_many (regardless of the number of files being pulled in a given batch), the transient errors never resolve. I haven't seen one scenario where if the failure is encountered once, it eventually succeeds. This also has a strange impact on the flow itself. I have noticed that the flow goes into a wind-down type mode where less and less workers are being created (I'm running this flow in our K8S environment using Argo), almost as if its trying to wrap up what it can and kill the flow. Any suggestions on how to deal with this? I already am using @timeout(minutes=30) and @catch(flowerror). Neither appear to be working with this issue.
h
that's weird. what version of metaflow are you on?
ok i think i know what the problem is. will have a PR out in a few
f
I'm using 2.18.0. I don't know if this is a related issue, but in that work step, I am running a subprocess to execute problematic processing logic, e.g. sometimes has segmentation faults, etc. My goal here was to isolate the execution so that if there is a problem, the subprocess can be killed and we can resume the other steps without issue. I am noticing that when there is a problem with the subprocess, it seems to kill the step too. I have a try/catch around the subprocess step, and I have the @catch decorator around the step. Regardless the steps end in failure, which kills the whole flow.
h
how are you launching the subprocess?
f
Here is the block of code:
Copy code
wrapper_script = "utils/preprocess_wrapper.py"
            
            <http://logger.info|logger.info>(f"Starting subprocess preprocessing with timeout {timeout_seconds}s")
            
            # Run the preprocessing in subprocess
            process = subprocess.run(
                [
                    "uv", "run", "python",
                    wrapper_script,
                    input_file_path,
                    output_file_path,
                    self.preprocessor_name
                ],
                capture_output=True,
                text=True,
                timeout=timeout_seconds
            )
There is additional logging that follows this, and also logging from within the preprocess_wrapper.py. But none of this shows up in the logs, as you can see above, the Starting subprocess.... is logged out, and then nothing. The step dies.
Also, not sure if this is helpful, but I did try to execute the processing logic for that file locally, and it did execute fine, no problems. Makes it even more difficult to diagnose why it might have failed in that subprocess.
h
what exit code do you get when the step dies from the subprocess failure?
f
None, it never gets that far. The last thing I see in the metaflow logs is "Starting subprocess.... " And then nothing, eventually I see the step has failed. I have extensive logging around and in the subprocess and none of it shows up. So I'm wondering if its failing to setup the subprocess
Copy code
try:
            # Get the wrapper script path - it should be in utils/ relative to where we're running
            wrapper_script = "utils/preprocess_wrapper.py"
            
            <http://logger.info|logger.info>(f"Starting subprocess preprocessing with timeout {timeout_seconds}s")
            
            # Run the preprocessing in subprocess
            process = subprocess.run(
                [
                    "uv", "run", "python",
                    wrapper_script,
                    input_file_path,
                    output_file_path,
                    self.preprocessor_name
                ],
                capture_output=True,
                text=True,
                timeout=timeout_seconds
            )
            
            # Log subprocess output for debugging (both success and failure cases)
            if process.stdout:
                stdout_lines = process.stdout.strip().split('\n')
                for line in stdout_lines:
                    if line.strip():  # Skip empty lines
                        <http://logger.info|logger.info>(f"[Subprocess] {line.strip()}")
            
            if process.stderr:
                stderr_lines = process.stderr.strip().split('\n')
                for line in stderr_lines:
                    if line.strip():  # Skip empty lines
                        if 'ERROR' in line or 'CRITICAL' in line:
                            logger.error(f"[Subprocess] {line.strip()}")
                        elif 'WARNING' in line:
                            logger.warning(f"[Subprocess] {line.strip()}")
                        else:
                            <http://logger.info|logger.info>(f"[Subprocess] {line.strip()}")
            
            # Check if subprocess succeeded
            if process.returncode != 0:
                # Extract just the core error message from stderr if available
                core_error = "Unknown preprocessing error"
                if process.stderr:
                    # Look for the actual error line (usually starts with "ERROR -")
                    lines = process.stderr.strip().split('\n')
                    for line in lines:
                        if 'ERROR -' in line or 'failed:' in line.lower():
                            core_error = line.strip()
                            break
                
                raise Exception(f"Preprocessing subprocess failed (exit code {process.returncode}): {core_error}")
            
            # Read processed data
            if not os.path.exists(output_file_path):
                raise Exception(f"Subprocess completed but output file not found: {output_file_path}")
            
            with open(output_file_path, 'r', encoding='utf-8') as f:
                processed_data = f.read()
            
            <http://logger.info|logger.info>(f"Subprocess preprocessing completed successfully")
            return processed_data
            
        except subprocess.TimeoutExpired:
            error_msg = f"Preprocessing subprocess timed out after {timeout_seconds} seconds"
            logger.error(error_msg)
            raise Exception(error_msg)
            
        except Exception as e:
            logger.error(f"Subprocess preprocessing failed: {e}")
            raise
            
        finally:
            # Clean up temporary files
            try:
                if os.path.exists(input_file_path):
                    os.unlink(input_file_path)
                if os.path.exists(output_file_path):
                    os.unlink(output_file_path)
            except Exception as cleanup_error:
                logger.warning(f"Failed to cleanup temp files: {cleanup_error}")
Here is the full block of code.
h
are you running the flow locally? you could still check the exit code via `$?`in bash for example
f
I'll give that a try.
Is there a way via the metaflow argo-workflows to query a given task and see what its output was? For example if I have a branched task name 't-65fa6189' Can I somehow check the exit code of that specific pod ?
h
i'm not familiar with argo, but chatgpt seems to think it's possible
f
I'll try to poke around and see if I can get exit codes from tasks, but its difficult because the subprocess fails only for 0.1-0.2% of our files, and its difficult to run it at that scale due to what 'local' development is for us, we use a limited environment to work and run jobs which has a 1 hour timeout on k8s access. I don't need 100% processing rate, but I just don't want the failures to fail the whole flow. Anything goes wrong with a branch in that step, I would like the @catch to pick it up and simply notify the next step of it when it comes to that.
h
PR is out for review.. will let you know when we cut a release
regarding the subprocess thing, you can try switching to 
Popen
and stream the logs as they come in, eg:
Copy code
import subprocess

with subprocess.Popen(
    ["uv", "run", "python", ...],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1,
) as p:
    for line in p.stdout:
        print("[child]", line.rstrip())
f
Thank you! I'll keep an eye out for the merge, and I'll also try out Popen. Thanks for the rapid help 🙂
h
fix is in 
2.18.8
f
Thank you! I'll update my version and see if the issue is still there.
11 Views
Open in Slack
PreviousNext
Powered by