You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "josh-fell (via GitHub)" <gi...@apache.org> on 2024/06/13 03:11:46 UTC
[I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
josh-fell opened a new issue, #40209:
URL: https://github.com/apache/airflow/issues/40209
### Discussed in https://github.com/apache/airflow/discussions/39597
<div type='discussions-op-text'>
<sup>Originally posted by **bejota** May 13, 2024</sup>
Hello,
I'm using HttpSensor to poll SLURM's REST API for job status. I'm using `deferrable` mode with `response_check`. The task only goes to the triggerer if I remove the `response_check`.
Airflow Docker Image `apache/airflow:latest-python3.9`
Airflow Version `2.9.0`
apache-airflow-providers-http `4.10.0`
Does not move to triggerer:
```
check_slurm_job = HttpSensor(
task_id="check_slurm_job",
http_conn_id="slurm_conn_id",
method="get",
endpoint="slurm/v0.0.40/job/999",
response_check=lambda response: "COMPLETED" in response.text,
poke_interval=10,
timeout=60 * 60 * 5,
deferrable=True, # not deferring to triggerer ?
dag=dag,
)
```
Moves to triggerer:
```
check_slurm_job = HttpSensor(
task_id="check_slurm_job",
http_conn_id="slurm_conn_id",
method="get",
endpoint="slurm/v0.0.40/job/999",
# response_check=lambda response: "COMPLETED" in response.text,
poke_interval=10,
timeout=60 * 60 * 5,
deferrable=True, # not deferring to triggerer ?
dag=dag,
)
```
In digging further, I see that HttpTrigger has no `response_check` method. I will attempt to add one, unless there is a better approach to this problem?</div>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
Posted by "kandharvishnu (via GitHub)" <gi...@apache.org>.
kandharvishnu commented on issue #40209:
URL: https://github.com/apache/airflow/issues/40209#issuecomment-2433238538
I would like to work on this issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
Posted by "k4rtikp4til (via GitHub)" <gi...@apache.org>.
k4rtikp4til commented on issue #40209:
URL: https://github.com/apache/airflow/issues/40209#issuecomment-3158739953
@dabla I looked at your PR that got merged, with regards to this issue what was the reason for going with check response to be defined at the connection level?
I am creating the dag files based on user input using jinja and connection is just the root to which I dynamically add endpoints and check response functions. Thus, I am still running into this issue. For anyone who comes across the issue again, this is what I have ended up with for now:
`class DeferredHttpPollingSensor(HttpSensor):
"""
Custom HttpSensor that defers the polling if the response is not ready.
"""
def __init__(self, *args, **kwargs):
"""Initialize the DeferredHttpPollingSensor."""
kwargs["deferrable"] = True
super().__init__(*args, **kwargs)
def execute(self, context: Context):
"""Execute the sensor."""
# Store the start time for timeout tracking
context["ti"].xcom_push(
key="deferred_http_polling_sensor_start", value=datetime.now().isoformat()
)
if self.poke(context):
return True
self.defer(
trigger=TimeDeltaTrigger(delta=timedelta(seconds=self.poke_interval)),
method_name="execute_complete",
)
def execute_complete(self, context: Context, event=None):
"""Execute the complete sensor."""
# Check timeout
start_time_str = context["ti"].xcom_pull(key="deferred_http_polling_sensor_start")
if start_time_str:
start_time = datetime.fromisoformat(start_time_str)
elapsed_time = (datetime.now() - start_time).total_seconds()
if elapsed_time >= self.timeout:
raise AirflowSensorTimeout(
f"Snap. Time is OUT. DAG: {context['dag'].dag_id}, "
f"Task: {context['task'].task_id}, "
f"Duration: {elapsed_time}s, "
f"Timeout: {self.timeout}s"
)
if self.poke(context):
return True
self.defer(
trigger=TimeDeltaTrigger(delta=timedelta(seconds=self.poke_interval)),
method_name="execute_complete",
)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #40209:
URL: https://github.com/apache/airflow/issues/40209#issuecomment-2713042329
I think for the check_response to be taken into account, you need to pass extra_options={"check_response": False},
otherwise the check_response parameter won't be taken into account.
I've created a [PR](https://github.com/apache/airflow/pull/45451) for this to allow to configure this also within extra of your http connection.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on issue #40209:
URL: https://github.com/apache/airflow/issues/40209#issuecomment-3160909935
> Hey [@dabla](https://github.com/dabla) – I looked into your PR that was merged related to this issue. Curious to understand: what was the motivation for moving the response_check logic to the connection level?
>
> In my case, I’m generating DAGs dynamically via Jinja templates based on user input. Each connection acts as a root, and I dynamically attach endpoints and response_check functions per task. So, binding response_check at the connection level isn’t ideal for this setup, and I'm still running into the same issue described here.
>
> For anyone else facing this, here’s the workaround I’m using for now:
>
> `class DeferredHttpPollingSensor(HttpSensor): """ Custom HttpSensor that defers the polling if the response is not ready. """
>
> ```
> def __init__(self, *args, **kwargs):
> """Initialize the DeferredHttpPollingSensor."""
> kwargs["deferrable"] = True
> super().__init__(*args, **kwargs)
>
> def execute(self, context: Context):
> """Execute the sensor."""
> # Store the start time for timeout tracking
> context["ti"].xcom_push(
> key="deferred_http_polling_sensor_start", value=datetime.now().isoformat()
> )
>
> if self.poke(context):
> return True
>
> self.defer(
> trigger=TimeDeltaTrigger(delta=timedelta(seconds=self.poke_interval)),
> method_name="execute_complete",
> )
>
> def execute_complete(self, context: Context, event=None):
> """Execute the complete sensor."""
> # Check timeout
> start_time_str = context["ti"].xcom_pull(key="deferred_http_polling_sensor_start")
> if start_time_str:
> start_time = datetime.fromisoformat(start_time_str)
> elapsed_time = (datetime.now() - start_time).total_seconds()
>
> if elapsed_time >= self.timeout:
> raise AirflowSensorTimeout(
> f"Snap. Time is OUT. DAG: {context['dag'].dag_id}, "
> f"Task: {context['task'].task_id}, "
> f"Duration: {elapsed_time}s, "
> f"Timeout: {self.timeout}s"
> )
>
> if self.poke(context):
> return True
>
> self.defer(
> trigger=TimeDeltaTrigger(delta=timedelta(seconds=self.poke_interval)),
> method_name="execute_complete",
> )`
> ```
> Hey [@dabla](https://github.com/dabla) – I looked into your PR that was merged related to this issue. Curious to understand: what was the motivation for moving the response_check logic to the connection level?
>
> In my case, I’m generating DAGs dynamically via Jinja templates based on user input. Each connection acts as a root, and I dynamically attach endpoints and response_check functions per task. So, binding response_check at the connection level isn’t ideal for this setup, and I'm still running into the same issue described here.
>
> For anyone else facing this, here’s the workaround I’m using for now:
>
> `class DeferredHttpPollingSensor(HttpSensor): """ Custom HttpSensor that defers the polling if the response is not ready. """
>
> ```
> def __init__(self, *args, **kwargs):
> """Initialize the DeferredHttpPollingSensor."""
> kwargs["deferrable"] = True
> super().__init__(*args, **kwargs)
>
> def execute(self, context: Context):
> """Execute the sensor."""
> # Store the start time for timeout tracking
> context["ti"].xcom_push(
> key="deferred_http_polling_sensor_start", value=datetime.now().isoformat()
> )
>
> if self.poke(context):
> return True
>
> self.defer(
> trigger=TimeDeltaTrigger(delta=timedelta(seconds=self.poke_interval)),
> method_name="execute_complete",
> )
>
> def execute_complete(self, context: Context, event=None):
> """Execute the complete sensor."""
> # Check timeout
> start_time_str = context["ti"].xcom_pull(key="deferred_http_polling_sensor_start")
> if start_time_str:
> start_time = datetime.fromisoformat(start_time_str)
> elapsed_time = (datetime.now() - start_time).total_seconds()
>
> if elapsed_time >= self.timeout:
> raise AirflowSensorTimeout(
> f"Snap. Time is OUT. DAG: {context['dag'].dag_id}, "
> f"Task: {context['task'].task_id}, "
> f"Duration: {elapsed_time}s, "
> f"Timeout: {self.timeout}s"
> )
>
> if self.poke(context):
> return True
>
> self.defer(
> trigger=TimeDeltaTrigger(delta=timedelta(seconds=self.poke_interval)),
> method_name="execute_complete",
> )`
> ```
Hello @k4rtikp4til, the issue is still open and my PR didn’t get merged yet, so to me it seems normal that you still run into this issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
Posted by "josh-fell (via GitHub)" <gi...@apache.org>.
josh-fell commented on issue #40209:
URL: https://github.com/apache/airflow/issues/40209#issuecomment-2176299198
Probably what the solution _should_ be. May need to confirm with `HttpOperator` too:
https://github.com/apache/airflow/discussions/39597#discussioncomment-9758322
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Deferrable HttpSensor does not move to Triggerer when using `response_check` [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #40209:
URL: https://github.com/apache/airflow/issues/40209#issuecomment-5384779240
This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org