Add deferrable support to `LambdaInvokeFunctionOperator` about airflow HOT 10 CLOSED

Happy to look Into this: 😀

Assigned

from airflow.

But if @gopidesupavan is interested in a similar issue, there is this issue: #40207. It seems the operator EcsRunTaskOperator has a bug with the deferrable mode. It is a good first issue

from airflow.

Indeed, that'd be a cool add-on!

from airflow.

Happy to look Into this: 😀

from airflow.

@gopidesupavan, have you started to work on it? Do you have any lead? I am asking this because I recall now there is no Lambda API to fetch status of a given execution. Which then makes this task impossible. Any pooling mechanism on a Lambda execution is purely impossible because there is no API to retrieve status

from airflow.

@gopidesupavan, have you started to work on it? Do you have any lead? I am asking this because I recall now there is no Lambda API to fetch status of a given execution. Which then makes this task impossible. Any pooling mechanism on a Lambda execution is purely impossible because there is no API to retrieve status

But we do have LambdaFunctionStateSensor
What am I missing?

from airflow.

@gopidesupavan, have you started to work on it? Do you have any lead? I am asking this because I recall now there is no Lambda API to fetch status of a given execution. Which then makes this task impossible. Any pooling mechanism on a Lambda execution is purely impossible because there is no API to retrieve status

But we do have LambdaFunctionStateSensor What am I missing?

LambdaFunctionStateSensor polls the deployment state of the AWS Lambda function until it reaches a target state. It does not poll an execution

from airflow.

Hi @eladkal @vincbeck , actually i was started this task yesterday, agreed what you said is correct. No option to get the execution status of lambda execution. What i am thinking is invoking lambda with async with deferable from the trigger. I thinking of some cases useful, where the lambda execution takes more time say (more than 5min) and InvocationType=RequestResponse. so in this case the trigger can listen on the response from the lambda?.

This is trigger logic am planning:

    def serialize(self) -> tuple[str, dict[str, Any]]:
        return (
            self.__class__.__module__ + "." + self.__class__.__qualname__,
            {
                "function_name": self.function_name,
                "payload": self.payload,
                "log_type": self.log_type,
                "qualifier": self.qualifier,
                "invocation_type": self.invocation_type,
                "client_context": self.client_context,
                "aws_conn_id": self.aws_conn_id,
                "region_name": self.region_name,
                "verify": self.verify,
                "botocore_config": self.botocore_config,
            },
        )

    @cached_property
    def hook(self) -> LambdaHook:
        return LambdaHook(aws_conn_id=self.aws_conn_id,
                          region_name=self.region_name,
                          verify=self.verify,
                          config=self.botocore_config,
                          )

    async def run(self) -> AsyncIterator[TriggerEvent]:
        response = await self.hook.invoke_lambda_async(
            function_name=self.function_name,
            invocation_type=self.invocation_type,
            log_type=self.log_type,
            client_context=self.client_context,
            payload=self.payload,
            qualifier=self.qualifier,
        )
 
        payload = await response.get("Payload").read()
        yield TriggerEvent({"status": "success", "response": response, "payload": payload.decode()})

Do you think something that usable? or is this approach not required?

from airflow.

But if @gopidesupavan is interested in a similar issue, there is this issue: #40207. It seems the operator EcsRunTaskOperator has a bug with the deferrable mode. It is a good first issue
Yes can take a look, looks like Ellis is already working on this.

from airflow.

As mentioned in #40425, this is not possible because AWS Lambda does not provide an API to retrieve status of an execution. Without this capability, it is not possible to make this operator deferrable

from airflow.