Support protocol version 2.0.0 to callback handlers orchestrated by CloudFormation service (#101)

Author: ammokhov

python/rpdk/python/codegen.py

@@ -50,27 +50,30 @@ def __init__(self):
         self.package_name = None
         self.package_root = None
         self._use_docker = True
+        self._protocol_version = "2.0.0"
 
     def _init_from_project(self, project):
         self.namespace = tuple(s.lower() for s in project.type_info)
         self.package_name = "_".join(self.namespace)
         self._use_docker = project.settings.get("use_docker", True)
         self.package_root = project.root / "src"
 
-    def _prompt_for_use_docker(self, project):
+    def _init_settings(self, project):
+        LOG.debug("Writing settings")
         self._use_docker = input_with_validation(
             "Use docker for platform-independent packaging (Y/n)?\n",
             validate_no,
             "This is highly recommended unless you are experienced \n"
             "with cross-platform Python packaging.",
         )
         project.settings["use_docker"] = self._use_docker
+        project.settings["protocolVersion"] = self._protocol_version
 
     def init(self, project):
         LOG.debug("Init started")
 
         self._init_from_project(project)
-        self._prompt_for_use_docker(project)
+        self._init_settings(project)
 
         project.runtime = self.RUNTIME
         project.entrypoint = self.ENTRY_POINT.format(self.package_name)

src/cloudformation_cli_python_lib/callback.py

@@ -95,30 +95,26 @@ class ProgressEvent:
     resourceModels: Optional[List[BaseModel]] = None
     nextToken: Optional[str] = None
 
-    def _serialize(
-        self, to_response: bool = False, bearer_token: Optional[str] = None
-    ) -> MutableMapping[str, Any]:
+    def _serialize(self) -> MutableMapping[str, Any]:
         # to match Java serialization, which drops `null` values, and the
         # contract tests currently expect this also
         ser = {k: v for k, v in self.__dict__.items() if v is not None}
+
         # mutate to what's expected in the response
-        if to_response:
-            ser["bearerToken"] = bearer_token
-            ser["operationStatus"] = ser.pop("status").name
-            if self.resourceModel:
+
+        ser["status"] = ser.pop("status").name
+
+        if self.resourceModel:
+            # pylint: disable=protected-access
+            ser["resourceModel"] = self.resourceModel._serialize()
+        if self.resourceModels:
+            ser["resourceModels"] = [
                 # pylint: disable=protected-access
-                ser["resourceModel"] = self.resourceModel._serialize()
-            if self.resourceModels:
-                ser["resourceModels"] = [
-                    # pylint: disable=protected-access
-                    model._serialize()
-                    for model in self.resourceModels
-                ]
-            del ser["callbackDelaySeconds"]
-            if "callbackContext" in ser:
-                del ser["callbackContext"]
-            if self.errorCode:
-                ser["errorCode"] = self.errorCode.name
+                model._serialize()
+                for model in self.resourceModels
+            ]
+        if self.errorCode:
+            ser["errorCode"] = self.errorCode.name
         return ser
 
     @classmethod

src/cloudformation_cli_python_lib/interface.py

@@ -3,11 +3,9 @@
 import traceback
 from datetime import datetime
 from functools import wraps
-from time import sleep
 from typing import Any, Callable, MutableMapping, Optional, Tuple, Type, Union
 
 from .boto3_proxy import SessionProxy, _get_boto_session
-from .callback import report_progress
 from .exceptions import InternalFailure, InvalidRequest, _HandlerError
 from .interface import (
     Action,
@@ -18,7 +16,6 @@
 )
 from .log_delivery import ProviderLogHandler
 from .metrics import MetricsPublisherProxy
-from .scheduler import cleanup_cloudwatch_events, reschedule_after_minutes
 from .utils import (
     BaseModel,
     Credentials,
@@ -32,7 +29,6 @@
 LOG = logging.getLogger(__name__)
 
 MUTATING_ACTIONS = (Action.CREATE, Action.UPDATE, Action.DELETE)
-INVOCATION_TIMEOUT_MS = 60000
 
 HandlerSignature = Callable[
     [Optional[SessionProxy], Any, MutableMapping[str, Any]], ProgressEvent
@@ -72,38 +68,6 @@ def _add_handler(f: HandlerSignature) -> HandlerSignature:
 
         return _add_handler
 
-    @staticmethod
-    def schedule_reinvocation(
-        handler_request: HandlerRequest,
-        handler_response: ProgressEvent,
-        context: LambdaContext,
-        session: SessionProxy,
-    ) -> bool:
-        if handler_response.status != OperationStatus.IN_PROGRESS:
-            return False
-        # modify requestContext dict in-place, so that invoke count is bumped on local
-        # reinvoke too
-        reinvoke_context = handler_request.requestContext
-        reinvoke_context["invocation"] = reinvoke_context.get("invocation", 0) + 1
-        callback_delay_s = handler_response.callbackDelaySeconds
-        remaining_ms = context.get_remaining_time_in_millis()
-
-        # when a handler requests a sub-minute callback delay, and if the lambda
-        # invocation has enough runtime (with 20% buffer), we can re-run the handler
-        # locally otherwise we re-invoke through CloudWatchEvents
-        needed_ms_remaining = callback_delay_s * 1200 + INVOCATION_TIMEOUT_MS
-        if callback_delay_s < 60 and remaining_ms > needed_ms_remaining:
-            sleep(callback_delay_s)
-            return True
-        callback_delay_min = int(callback_delay_s / 60)
-        reschedule_after_minutes(
-            session,
-            function_arn=context.invoked_function_arn,
-            minutes_from_now=callback_delay_min,
-            handler_request=handler_request,
-        )
-        return False
-
     def _invoke_handler(
         self,
         session: Optional[SessionProxy],
@@ -169,26 +133,23 @@ def test_entrypoint(
     def _parse_request(
         event_data: MutableMapping[str, Any]
     ) -> Tuple[
-        Tuple[Optional[SessionProxy], Optional[SessionProxy], SessionProxy],
+        Tuple[Optional[SessionProxy], Optional[SessionProxy]],
         Action,
         MutableMapping[str, Any],
         HandlerRequest,
     ]:
         try:
             event = HandlerRequest.deserialize(event_data)
-            platform_sess = _get_boto_session(event.requestData.platformCredentials)
             caller_sess = _get_boto_session(event.requestData.callerCredentials)
             provider_sess = _get_boto_session(event.requestData.providerCredentials)
             # credentials are used when rescheduling, so can't zero them out (for now)
-            if platform_sess is None:
-                raise ValueError("No platform credentials")
             action = Action[event.action]
-            callback_context = event.requestContext.get("callbackContext", {})
+            callback_context = event.callbackContext or {}
         except Exception as e:  # pylint: disable=broad-except
             LOG.exception("Invalid request")
             raise InvalidRequest(f"{e} ({type(e).__name__})") from e
         return (
-            (caller_sess, provider_sess, platform_sess),
+            (caller_sess, provider_sess),
             action,
             callback_context,
             event,
@@ -224,66 +185,28 @@ def print_or_log(message: str) -> None:
 
         try:
             sessions, action, callback, event = self._parse_request(event_data)
-            caller_sess, provider_sess, platform_sess = sessions
+            caller_sess, provider_sess = sessions
             ProviderLogHandler.setup(event, provider_sess)
             logs_setup = True
 
             request = self._cast_resource_request(event)
 
             metrics = MetricsPublisherProxy(event.awsAccountId, event.resourceType)
-            metrics.add_metrics_publisher(platform_sess)
             metrics.add_metrics_publisher(provider_sess)
-            # Acknowledge the task for first time invocation
-            if not event.requestContext:
-                report_progress(
-                    platform_sess,
-                    event.bearerToken,
-                    None,
-                    OperationStatus.IN_PROGRESS,
-                    OperationStatus.PENDING,
-                    None,
-                    "",
-                )
-            else:
-                # If this invocation was triggered by a 're-invoke' CloudWatch Event,
-                # clean it up
-                cleanup_cloudwatch_events(
-                    platform_sess,
-                    event.requestContext.get("cloudWatchEventsRuleName", ""),
-                    event.requestContext.get("cloudWatchEventsTargetId", ""),
-                )
-            invoke = True
-            while invoke:
-                metrics.publish_invocation_metric(datetime.utcnow(), action)
-                start_time = datetime.utcnow()
-                error = None
-                try:
-                    progress = self._invoke_handler(
-                        caller_sess, request, action, callback
-                    )
-                except Exception as e:  # pylint: disable=broad-except
-                    error = e
-                m_secs = (datetime.utcnow() - start_time).total_seconds() * 1000.0
-                metrics.publish_duration_metric(datetime.utcnow(), action, m_secs)
-                if error:
-                    metrics.publish_exception_metric(datetime.utcnow(), action, error)
-                    raise error
-                if progress.callbackContext:
-                    callback = progress.callbackContext
-                    event.requestContext["callbackContext"] = callback
-                if event.action in MUTATING_ACTIONS:
-                    report_progress(
-                        platform_sess,
-                        event.bearerToken,
-                        progress.errorCode,
-                        progress.status,
-                        OperationStatus.IN_PROGRESS,
-                        progress.resourceModel,
-                        progress.message,
-                    )
-                invoke = self.schedule_reinvocation(
-                    event, progress, context, platform_sess
-                )
+
+            metrics.publish_invocation_metric(datetime.utcnow(), action)
+            start_time = datetime.utcnow()
+            error = None
+
+            try:
+                progress = self._invoke_handler(caller_sess, request, action, callback)
+            except Exception as e:  # pylint: disable=broad-except
+                error = e
+            m_secs = (datetime.utcnow() - start_time).total_seconds() * 1000.0
+            metrics.publish_duration_metric(datetime.utcnow(), action, m_secs)
+            if error:
+                metrics.publish_exception_metric(datetime.utcnow(), action, error)
+                raise error
         except _HandlerError as e:
             print_or_log("Handler error")
             progress = e.to_progress_event()
@@ -296,6 +219,4 @@ def print_or_log(message: str) -> None:
 
         # use the raw event_data as a last-ditch attempt to call back if the
         # request is invalid
-        return progress._serialize(  # pylint: disable=protected-access
-            to_response=True, bearer_token=event_data.get("bearerToken")
-        )
+        return progress._serialize()  # pylint: disable=protected-access

src/cloudformation_cli_python_lib/resource.py

@@ -54,7 +54,6 @@ class RequestData:
     stackTags: Optional[Mapping[str, Any]] = None
     # platform credentials aren't really optional, but this is used to
     # zero them out to prevent e.g. accidental logging
-    platformCredentials: Optional[Credentials] = None
     callerCredentials: Optional[Credentials] = None
     providerCredentials: Optional[Credentials] = None
     previousResourceProperties: Optional[Mapping[str, Any]] = None
@@ -91,6 +90,7 @@ class HandlerRequest:
     resourceTypeVersion: str
     requestData: RequestData
     stackId: str
+    callbackContext: Optional[MutableMapping[str, Any]] = None
     nextToken: Optional[str] = None
     requestContext: MutableMapping[str, Any] = field(default_factory=dict)