Add deferrable support to `LambdaInvokeFunctionOperator`

[eladkal]

Body

#33327 added deferrable option to LambdaCreateFunctionOperator we should have similar functionality in LambdaInvokeFunctionOperator

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[vincbeck]

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

[gopidesupavan]

Happy to look Into this: 😀

[eladkal]

Happy to look Into this: 😀

Assigned

[vincbeck]

@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

[eladkal]

@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?

[vincbeck]

@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

[vincbeck]

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

[gopidesupavan]

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?