Hello, our flows have been successfully running for 1 month and made more than 200 successful runs. A new error has just started yesterday. For the last 6 runs, half of them were not completed and stayed in one of our tasks. There is no log information. According to the task logic, the Kinesis Data Analytics must be checked per 60 seconds. After a while, just waiting without doing anything. There is still an active flow staying in that condition, and I have not killed it to show you. Could you please check it?

Hey @Ismail Cenik, are you on server or cloud?

Hello, We run several times during the development. However, the number of runs decreased because of some QA activities. I realized yesterday. Our Prefect Agent has been installed on our EKS. I am using prefect cloud to manage flow runs.

Did this just happen once? You haven’t tried restarting the flow right?

It happened 3 times in the last 24 hours. I have killed the other two and one of them is still active

This seems like something to do with the EKS cluster. We won’t have any more information on the Prefect side than you see. Maybe you can check the Kubernetes logs?

Prefect just stops to check Kinesis Data Analytics status, only stays like in sleep mode

Can you give me an example flow ID?

a4699a77-2b95-44c6-9eae-2d134ec45d6a

So your log ends with

Sleeping for 60s" }

and then nothing else?

Yes, per 60 seconds, it must look at the status, but just stays on that case several hours

Would you know if there is a timeout to requests on Kinesis Data Analytics?

Let me check

Does the kinesis_client have a timeout feature?

No

I have seen people running long processes (think infinite while loops) on Prefect and it just runs. This causes other issues but I bring this up because a long running task shouldn’t have issues. Based on the code and logs, it does seem to get stuck there.

The Kinesis average running time is around 4-5 minutes and we have 10 Kinesis in the flow

I am wondering if we should change this to use Prefect retries and timeouts instead so that we can timeout if Kinesis takes more than the expected time and hangs

This is the current code

@task(log_stdout=True, max_retries=5, retry_delay=timedelta(seconds=60))
def waitPlanDataLoadingTask(app_name):
    kinesis_client = boto3.client(
        'kinesisanalyticsv2', region_name=region,
        aws_access_key_id=key_id, aws_secret_access_key=access_key)
    while True:
        if isJobDone(app_name, kinesis_client):
            des_response = kinesis_client.describe_application(ApplicationName=app_name, IncludeAdditionalDetails=True)
            if des_response['ApplicationDetail']['ApplicationStatus'] == 'READY':
                return
            stop_response = kinesis_client.stop_application(ApplicationName=app_name, Force=False)
            print(f'Stop KDA: {app_name}, response: {stop_response}')
            return
        print(f"Sleeping for {SLEEP_TIME}s")
        time.sleep(int(SLEEP_TIME))

I’ll look a bit into Kinesis and see if we have to go down this path

No, could you give an example?

This is how to add boto3 logs

I will request TOML change from a different team ... Please double-check it.

[logging]
# Extra loggers for Prefect log configuration
extra_loggers = "['boto3']"

Then I can define the environmental variable on the scripts (Register flow/runf flow) like that

export PREFECT__LOGGING__EXTRA_LOGGERS="['boto3']"

And update the code as

import logging
import sys
for l in ['boto3']:
    logger = logging.getLogger(l)
    logger.setLevel('INFO')
    log_stream = logging.StreamHandler(sys.stdout)
    log_stream.setFormatter(LOG_FORMAT)
    logger.addHandler(log_stream)

are they correct?

You don’t need to update the code. The TOML or environment variable will already work. I think the environment variable in the RunConfig will work so you don’t need to go to the different team. You can do something like

KubernetesRun(env={"PREFECT__LOGGING__EXTRA_LOGGERS": "['boto3']")

. The location of the TOML file though is in the home folder under the

.prefect

folder.

So I will not change anything in TOML, right. In my local prefect, the file is under ".prefect", but for the cloud case, I do not know where it is.

Yes I don’t think you have to. Testing now

I have started the flow, and let you know when we see the same issue

I am not seeing any boto3 logs or debug level logs. Did you add the logging level to be

DEBUG

?

I just added what you provided (

flow.run_config = KubernetesRun(_env_={"PREFECT__LOGGING__LEVEL": "DEBUG","PREFECT__LOGGING__EXTRA_LOGGERS": "['boto3']"})

). Is there anymore update?

I assume you re-registered? What storage are you using?

Yes, I always registered flows when updating, let me send storage info private

So I think what happened is that when you use

DockerStorage

with

KubernetesRun

, the default image pull policy for Kubernetes is

IfNotPresent

meaning that it only pulls the container if it doesn’t have one with the same name. The image gets updated but because it has the same name, Kubernetes doesn’t pull it. Can you add

flow.run_config = KubernetesRun(image_pull_policy="Always")

. This was added in 0.14.18.

I am using a job template

That’s weird. I tried setting up the boto3 logs and it worked locally (I read stuff from S3 to test). At this point, the simplest fix would be including an appropriate timeout like:

@task(log_stdout=True, max_retries=200, retry_delay=timedelta(seconds=60), timeout=600)
def waitPlanDataLoadingTask(app_name):

It’s very likely it’s one of the

describe_application

or

stop_application

that’s hanging. If that Kinesis call is only working sometimes and hangs on others. This

timeout

will retry the task if it hangs. The only thing to note is that it will timeout after a certain number of loop iterations. To work with this, you can also change the max_retries to something very generous so that this process will just keep running.

Hello, we updated our prefect version to the 0.14.20, now observing

Was that causing issues?

Not sure, since we are getting the same error

I don’t expect this to be your specific problem but a lot of weird errors do happen when the flow version is higher than the agent version. The agent should be higher