No response from request api(longer than 30min) while using SimpleHttpOperator

[yuanke7]

Apache Airflow version

2.1.2

Operating System

CentOS Linux 7

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==2.0.0
apache-airflow-providers-celery==2.0.0
apache-airflow-providers-cncf-kubernetes==2.0.0
apache-airflow-providers-docker==2.0.0
apache-airflow-providers-elasticsearch==2.0.2
apache-airflow-providers-ftp==2.0.0
apache-airflow-providers-google==4.0.0
apache-airflow-providers-grpc==2.0.0
apache-airflow-providers-hashicorp==2.0.0
apache-airflow-providers-http==2.0.0
apache-airflow-providers-imap==2.0.0
apache-airflow-providers-microsoft-azure==3.0.0
apache-airflow-providers-mysql==2.0.0
apache-airflow-providers-postgres==2.0.0
apache-airflow-providers-redis==2.0.0
apache-airflow-providers-sendgrid==2.0.0
apache-airflow-providers-sftp==2.0.0
apache-airflow-providers-slack==4.0.0
apache-airflow-providers-sqlite==2.0.0
apache-airflow-providers-ssh==2.0.0

Deployment

Docker-Compose

Deployment details

No response

What happened

I'm using SimpleHttpOperator to request an api which get reponse longer than 30min often, but task is still in running state even api has already returned a response.
Here is one of my task code:
psd_report = SimpleHttpOperator( task_id='psd_report', method='GET', http_conn_id=WIN_SERVER_1, endpoint='general/psd_report', response_check=lambda response : response.json().get('Success'), response_filter=lambda response : response.json(), execution_timeout=timedelta(hours=1), dag=dag, )
Here is my api response data:
{"success":True}

What you expected to happen

I expected that task using SimpleHttpOperator could switch state from running to success so that downstream task could get into the queue.

How to reproduce

Creat a SimpleHttpOperator task, request an api that need at least 30 min or more to response. Trigger this task you could find even api has returned a response, the task is still in running state.

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[Jorricks]

If had a similar issue recently (with a flask app I created myself).
I was using postman to do some test and I didn't realize I had the content-length header still defined. It caused all my requests to wait indefinitely until I would get a time-out.
I'm just guessing here but is it possible that something is happening here as well?

2. Are you setting any headers?
3. What do you get when you query any other endpoint?

[yuanke7]

If had a similar issue recently (with a flask app I created myself).
I was using postman to do some test and I didn't realize I had the content-length header still defined. It caused all my requests to wait indefinitely until I would get a time-out.
I'm just guessing here but is it possible that something is happening here as well?

1. Are you setting any headers?
2. What do you get when you query any other endpoint?

No I didn't set any headers cuz my own api is really simple , I'll try it, thanks !!!

[yuanke7]

If had a similar issue recently (with a flask app I created myself).
I was using postman to do some test and I didn't realize I had the content-length header still defined. It caused all my requests to wait indefinitely until I would get a time-out.
I'm just guessing here but is it possible that something is happening here as well?

1. Are you setting any headers?
2. What do you get when you query any other endpoint?

Hi Jorricks, I have tried but not work, same issue.

And i send request by postman without setting headers, it gets reponse correctly even longer than 30min.

It has to be the problem from Airflow I guess.

Answer for your second question: I have same response json data with my api, {'success':True/False}

[potiuk]

Please @yuanke7 follow the code of conduct, This is a professional environment, so I'd apreciate if you avoid personal notes like that (I edited your comment).

It might make people uncomfortable and we are very keen on creating a welcoming environment.

[Jorricks]

Can you please share the python code of your DAG?
I'll see if I can replicate the issue.

[yuanke7]

Can you please share the python code of your DAG?
I'll see if I can replicate the issue.

Sure, thanks for your help !!!
`

import pendulum
from airflow import DAG
from proj.config import WIN_SERVER_1, SERVER_1_HEADERS
from proj.utils.dag_related import start_wait
from datetime import timedelta, datetime, time
from airflow.sensors.time_sensor import TimeSensor
from airflow.operators.python import PythonOperator
from airflow.providers.http.operators.http import SimpleHttpOperator

default_args = {
'owner' : 'yuankq1',
'depends_on_past' : False,
}

yest = datetime.date(datetime.today() + timedelta(-7))
dag = DAG(
dag_id='mingle_dag',
default_args=default_args,
description='A mingle dag',
concurrency=8,
max_active_runs=1,
catchup=False,
start_date=datetime(yest.year, yest.month, yest.day, tzinfo=pendulum.timezone("Asia/Shanghai")),
schedule_interval='0 8 * * 1-5',
tags=['yuankq1', 'mingle', 'varies', 'api'], # search dags by tags in web
)

start = PythonOperator(
task_id='start_wait',
python_callable=start_wait,
dag=dag
)

sensor_09 = TimeSensor(
task_id='wait_until_09_00',
target_time=time(9, 0)
)
sensor_13 = TimeSensor(
task_id='wait_until_13_00',
target_time=time(13, 0),
)
sensor_14 = TimeSensor(
task_id='wait_until_14_00',
target_time=time(14, 0)
)
sensor_22 = TimeSensor(
task_id='wait_until_22_00',
target_time=time(22, 0)
)
sensor_10_30 = TimeSensor(
task_id='wait_until_10_30',
target_time=time(10, 30)
)
sensor_15_30 = TimeSensor(
task_id='wait_until_15_30',
target_time=time(15, 30)
)
sensor_17 = TimeSensor(
task_id='wait_until_17',
target_time=time(17, 0)
)
sensor_17_30 = TimeSensor(
task_id='wait_until_17_30',
target_time=time(17, 30)
)

save_hub_report = SimpleHttpOperator(
task_id='save_hub_report',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='general/save_hub_report',
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(minutes=30),
dag=dag,
)

matrix_collection = SimpleHttpOperator(
task_id='matrix_collection',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='general/matrix_collection',
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
dag=dag,
execution_timeout=timedelta(minutes=10),
)

psd_report = SimpleHttpOperator(
task_id='psd_report',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='general/psd_report',
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(hours=1),
dag=dag,
)

def flow_a(order_type: str = None, execute_time: str = None) :
flow_a = SimpleHttpOperator(
task_id=f'flow_a_{order_type}_{execute_time}',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='sap/flow_a',
data={'order_type' : order_type},
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(hours=3),
depends_on_past=True,
dag=dag,
)
return flow_a

flow_a_ww2dc_10_30 = flow_a(order_type='ww2dc', execute_time='10_30')
flow_a_ww2dc_17_30 = flow_a(order_type='ww2dc', execute_time='17_30')
flow_a_other_type_10_30 = flow_a(order_type='other_type', execute_time='10_30')
flow_a_other_type_15_30 = flow_a(order_type='other_type', execute_time='15_30')
flow_a_other_type_17_30 = flow_a(order_type='other_type', execute_time='17_30')

flow_b = SimpleHttpOperator(
task_id='flow_b',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='sap/flow_b',
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(hours=5),
dag=dag,
)

flow_d = SimpleHttpOperator(
task_id='flow_d',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='sap/flow_d',
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(hours=3),
dag=dag,
)

po_change = SimpleHttpOperator(
task_id='po_change',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='sap/po_change',
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(minutes=15),
dag=dag,
)

def flow_a_monitor_func(task_id) :
return SimpleHttpOperator(
task_id=f'{task_id}',
method='GET',
http_conn_id=WIN_SERVER_1,
endpoint='sap/flow_a/monitor_repeat',
headers=SERVER_1_HEADERS,
response_check=lambda response : response.json().get('Success'),
response_filter=lambda response : response.json(),
execution_timeout=timedelta(minutes=5),
dag=dag,
)

flow_a_monitor = flow_a_monitor_func('flow_a_monitor')
flow_a_monitor2 = flow_a_monitor_func('flow_a_monitor2')

start >> matrix_collection
start >> flow_a_monitor
start >> sensor_10_30 >> sensor_13 >> sensor_14 >> sensor_15_30 >> sensor_17 >> sensor_17_30
sensor_10_30 >> flow_a_ww2dc_10_30 >> flow_a_other_type_10_30
sensor_13 >> psd_report
sensor_14 >> save_hub_report
sensor_15_30 >> flow_a_other_type_15_30
sensor_17 >> po_change
sensor_17 >> flow_a_monitor2
sensor_17_30 >> flow_a_ww2dc_17_30 >> flow_a_other_type_17_30
sensor_17_30 >> flow_d >> flow_b
`

[yuanke7]

Can you please share the python code of your DAG?
I'll see if I can replicate the issue.

Hi Jorricks, here is my simple flask API for test which you can see it would response in 35min at least.

@bp.route('/test')

def test_api():
    
      start_ = time.localtime()
    
      min = int(request.args.get('min', 35))
    
      logger.info(f'testing start (min{min})...')
    
      time.sleep(60 * min)
   
      logger.info(f'testing end (min{min}).')
    
      end_ = time.localtime()
    
      return flask.json.dumps({'min': min, 'start':start_, 'end':end_})

[yuanke7]

Can you please share the python code of your DAG? I'll see if I can replicate the issue.

Do you have any idea by now ?

[yuanke7]

Finally it solved by myself.

If you face the same problem while using Airflow or sending long response request and not getting response, use linux command 'curl' to send a http request instead of python's 'requests' pkg.

It has to be the problem of python requests(SimpleHttpOperator based on it), it works fine on my Mac but failed on my linux server which confused me for a long time, pls leave an answer if you know it.

[KCIV]

@yuanke7 Can you provide a deeper example of how this was fixed? your dag/task for example?

I'm running into this scenario where a dag calls to an API endpoint which lasts longer than 5 minutes.

Request.get works in all tests (swagger/browser/python) but used in airflow fails after 5 minutes ignoring retries/timeout arguments.

[yuanke7]

@KCIV Try send your requests with a BashOperator, I wrote an operator based on BashOperator:

class BashHttpOperator(BashOperator):

def __init__(self,
             *,
             server: BaseServer.mro,
             http_conn_id: str = 'http_default',  # Reserve
             method: str = 'GET', # Reserve
             endpoint: str,
             data: Optional[Dict] = None,
             env: Optional[Dict[str, str]] = None,
             output_encoding: str = 'utf-8',
             skip_exit_code: int = 99,
             err_exit_code: int = 500,
             check_response_callable: Callable = lambda res: res.get('Success'),
             **kwargs):
    """
    :param server: BaseServer provide url, port.
    :param endpoint: Url endpoint.
    :param data: A dict data for url.
    :param err_exit_code: Raise exception when this code returned.
    :param err_result_callable: Check response by result data.
    """
    data_str = ' '
    if data:
        for k,v in data.items():
            data_str += f'--data-urlencode "{k}={v}" '
    self.bash_command = f'curl -X GET -G -i  http://{server.HOST}:{server.API_PORT}/{endpoint}' + data_str
    super(BashHttpOperator, self).__init__(bash_command=self.bash_command, env=env, output_encoding=output_encoding,  skip_exit_code=skip_exit_code, **kwargs)
    self.err_exit_code = err_exit_code
    self.check_response_callable = check_response_callable

def execute(self, context):
    env = self.get_env(context)
    result = self.subprocess_hook.run_command(
        command=['bash', '-c', self.bash_command],
        env=env,
        output_encoding=self.output_encoding,
    )
    self.log.info(f'------Status code: {result.exit_code}------')
    if self.skip_exit_code is not None and result.exit_code == self.skip_exit_code:
        raise AirflowSkipException(f"Bash command returned exit code {self.skip_exit_code}. Skipping.")
    elif self.err_exit_code and result.exit_code == self.err_exit_code:
        raise Exception('Internal server error, pls check API logs.')
    elif result.exit_code == 56:
        raise Exception(f'Connect to IP failed !!!')
    elif result.exit_code != 0:
        raise Exception('Bash command failed. The command returned a non-zero exit code.')
    res:dict = json.loads(result.output)
    if self.check_response_callable:
        if not self.check_response_callable(res):
            raise Exception(f'Task failed due to param: check_response_callable (check this param and response for detail), response: {res}')
    return res

This operator works fine for me, hope to help you.

[potiuk]

I have no ide where the default comes from - but maybe you passed it in the operator? The HttpOperator has "extra_options" dictionary that you can pass "requests" parameters (including "timeout" mentioned in the docuementation). Maybe - unknowingly - you pass a value there. You can also pass a None value to force "no timeout" https://docs.python-requests.org/en/latest/user/advanced/#timeouts - unfortunately "requests" does not tell explicitly what default it has, so mabe it is somehow server/system dependent. Maybe a variable of some sort.

Or maybe this is something in your infrastructure that kills the request after 5 minutes of inactivity ? Airlfow requests are coming from the machine where tasks are running - and this might be inside an infrastructure that you have no access to (so even if it works from your local machine, it might not work if you run it via airflow because it is run elsewhere. This usually happens on various deployments when you have firewals/corporate environment. In this case I am afraid you neeed to talk to your infra people.

[potiuk]

Converted it to discussion - as this is not very likely to be Airlfow issue. If you have other findings - you might add them here, but I think the first thing you should do is "exec" to be on the same machine that airflow does and execute (in python) the very same python .requests.get() that Airflow does (and repeat it locally) and see if the difference is "environmental". This shoudl give you a clue where to look for.

[Arti3DPlayer]

It is working 9 hours already, response from server was returned for sure

[potiuk]

Hmm. I have a hypothesis. @Arti3DPlayer - maybe you can help me to verfify it and if it works, then I will be able to come up with some code that will make it works in those situations for SimpleHttpOperator ?

I think it can have a lot to do with the TCP_KEEPALIVE. Curl by default uses TCP_KEEPALIVE and you can disable it with --no-keepalive - on the other hand, requests library does not seem to have tcp_keepalive enabled by default and it requires some custom code to be added for it to be enabled.

NOTE! KeepAlive mentioned in requests library are DIFFERENT keepalives. The keepalive that requests library has enabled by default are HTTP-level Keepalives. They keep connections opened for some time and reuse them when you perform subsequent request to the same server. What I am talking about are TCP_KEEPALIVE which is sending an "empty" packet on TCP periodically requiring and ACK to be sent back by the server, even if there is no actual data being transmitted.

Example of this case and implementation here: https://www.finbourne.com/blog/the-mysterious-hanging-client-tcp-keep-alives

My hypothesis is that you have a Firewall in between that simply closes inactive connections and and TCP_KEEPALIVE prevents them from being closed.

So I have a kind request @Arti3DPlayer - can you please run your curl requests with --no-keepalive flag and see if they will be hanging as well ? If we can confirm that, then I can implemente TCP_KEEPALIVE as integration with the requests libarary.

[potiuk]

I actually went ahead and added #24967 as I think having keep alive is good and I verified in my setting that initially no keepalive is used by requests (as expected). I would love @Arti3DPlayer if you could run the quick test I proposed to confirm if my hypothesis is right.

[Arti3DPlayer]

@potiuk yes, will do shortly

[Arti3DPlayer]

@potiuk I ran two times and curl request with --no-keepalive stuck, while curl request without this option finished in 50 min as it should. I can't test your PR, because I use google cloud composer, but seems we need to enable this option for python requests package by default

[potiuk]

Yeah. Thanks for confirming. That's exactly what I did in #24967