Skip to content

implement CloudWatch metrics #69

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tobywf merged 5 commits into from
Dec 4, 2019
Merged

implement CloudWatch metrics #69

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
645cd62
implement cloudwatch metrics
jaymccon Nov 28, 2019
175b5af
use utc time for metrics
jaymccon Dec 4, 2019
2948e0e
cleaner if block
jaymccon Dec 4, 2019
1d728d7
update test to take ResourceModel argument
jaymccon Dec 4, 2019
4852b20
flake8 under python 3.8 catches C901 on a different line to < 3.8
jaymccon Dec 4, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 71 additions & 42 deletions src/cloudformation_cli_python_lib/metrics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,17 +11,21 @@

LOG = logging.getLogger(__name__)

METRIC_NAMESPACE_ROOT = "AWS/CloudFormation"


def format_dimensions(dimensions: Mapping[str, str]) -> List[Mapping[str, str]]:
return [{"Name": key, "Value": value} for key, value in dimensions.items()]


class MetricPublisher:
def __init__(self, namespace: str, session: Session) -> None:
self.namespace = namespace
def __init__(self, account_id: str, resource_type: str, session: Session) -> None:
suffix = resource_type.replace("::", "/")
self.namespace = f"{METRIC_NAMESPACE_ROOT}/{account_id}/{suffix}"
self.resource_type = resource_type
self.client = session.client("cloudwatch")

def _publish_metric( # pylint: disable-msg=too-many-arguments
def publish_metric( # pylint: disable-msg=too-many-arguments
self,
metric_name: MetricTypes,
dimensions: Mapping[str, str],
Expand All @@ -34,9 +38,9 @@ def _publish_metric( # pylint: disable-msg=too-many-arguments
Namespace=self.namespace,
MetricData=[
{
"MetricName": metric_name,
"MetricName": metric_name.name,
"Dimensions": format_dimensions(dimensions),
"Unit": unit,
"Unit": unit.name,
"Timestamp": str(timestamp),
"Value": value,
}
Expand All @@ -46,51 +50,76 @@ def _publish_metric( # pylint: disable-msg=too-many-arguments
except ClientError as e:
LOG.error("An error occurred while publishing metrics: %s", str(e))


class MetricsPublisherProxy:
def __init__(self) -> None:
self._publishers: List[MetricPublisher] = []

def add_metrics_publisher(self, publisher: MetricPublisher) -> None:
self._publishers.append(publisher)

def publish_exception_metric(
self, timestamp: datetime.datetime, action: Action, error: Any
) -> None:
dimensions: Mapping[str, str] = {
"DimensionKeyActionType": action,
"DimensionKeyExceptionType": str(type(error)),
"DimensionKeyResourceType": self.namespace,
}

self._publish_metric(
metric_name=MetricTypes.HandlerException,
dimensions=dimensions,
unit=StandardUnit.Count,
value=1.0,
timestamp=timestamp,
)
for publisher in self._publishers:
dimensions: Mapping[str, str] = {
"DimensionKeyActionType": action.name,
"DimensionKeyExceptionType": str(type(error)),
"DimensionKeyResourceType": publisher.resource_type,
}
publisher.publish_metric(
metric_name=MetricTypes.HandlerException,
dimensions=dimensions,
unit=StandardUnit.Count,
value=1.0,
timestamp=timestamp,
)

def publish_invocation_metric(
self, timestamp: datetime.datetime, action: Action
) -> None:
dimensions = {
"DimensionKeyActionType": action,
"DimensionKeyResourceType": self.namespace,
}

self._publish_metric(
metric_name=MetricTypes.HandlerInvocationCount,
dimensions=dimensions,
unit=StandardUnit.Count,
value=1.0,
timestamp=timestamp,
)
for publisher in self._publishers:
dimensions = {
"DimensionKeyActionType": action.name,
"DimensionKeyResourceType": publisher.resource_type,
}
publisher.publish_metric(
metric_name=MetricTypes.HandlerInvocationCount,
dimensions=dimensions,
unit=StandardUnit.Count,
value=1.0,
timestamp=timestamp,
)

def publish_duration_metric(
self, timestamp: datetime.datetime, action: Action, milliseconds: float
) -> None:
dimensions = {
"DimensionKeyActionType": action,
"DimensionKeyResourceType": self.namespace,
}

self._publish_metric(
metric_name=MetricTypes.HandlerInvocationDuration,
dimensions=dimensions,
unit=StandardUnit.Milliseconds,
value=milliseconds,
timestamp=timestamp,
)
for publisher in self._publishers:
dimensions = {
"DimensionKeyActionType": action.name,
"DimensionKeyResourceType": publisher.resource_type,
}
publisher.publish_metric(
metric_name=MetricTypes.HandlerInvocationDuration,
dimensions=dimensions,
unit=StandardUnit.Milliseconds,
value=milliseconds,
timestamp=timestamp,
)

def publish_log_delivery_exception_metric(
self, timestamp: datetime.datetime, error: Any
) -> None:
for publisher in self._publishers:
dimensions: Mapping[str, str] = {
"DimensionKeyActionType": "ProviderLogDelivery",
"DimensionKeyExceptionType": str(type(error)),
"DimensionKeyResourceType": publisher.resource_type,
}
publisher.publish_metric(
metric_name=MetricTypes.HandlerException,
dimensions=dimensions,
unit=StandardUnit.Count,
value=1.0,
timestamp=timestamp,
)
62 changes: 47 additions & 15 deletions src/cloudformation_cli_python_lib/resource.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import json
import logging
from datetime import datetime
from functools import wraps
from time import sleep
from typing import Any, Callable, MutableMapping, Optional, Tuple, Type, Union
Expand All @@ -18,6 +19,7 @@
ProgressEvent,
)
from .log_delivery import ProviderLogHandler
from .metrics import MetricPublisher, MetricsPublisherProxy
from .scheduler import CloudWatchScheduler
from .utils import (
BaseResourceModel,
Expand Down Expand Up @@ -95,14 +97,8 @@ def schedule_reinvocation(
# 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:
LOG.info(
"Scheduling re-invoke locally after %s seconds, with Context {%s}",
callback_delay_s,
reinvoke_context,
)
sleep(callback_delay_s)
return True
LOG.info("Scheduling re-invoke with Context {%s}", reinvoke_context)
callback_delay_min = int(callback_delay_s / 60)
CloudWatchScheduler(boto3_session=session).reschedule_after_minutes(
function_arn=context.invoked_function_arn,
Expand Down Expand Up @@ -175,8 +171,7 @@ def test_entrypoint(
def _parse_request(
self, event_data: MutableMapping[str, Any]
) -> Tuple[
Optional[SessionProxy],
boto3.Session,
Tuple[Optional[SessionProxy], Optional[boto3.Session], boto3.Session],
BaseResourceHandlerRequest,
Action,
MutableMapping[str, Any],
Expand All @@ -185,6 +180,7 @@ def _parse_request(
try:
event = HandlerRequest.deserialize(event_data)
caller_creds = event.requestData.callerCredentials
provider_creds = event.requestData.providerCredentials
platform_creds = event.requestData.platformCredentials
request: BaseResourceHandlerRequest = UnmodelledRequest(
clientRequestToken=event.bearerToken,
Expand All @@ -199,22 +195,45 @@ def _parse_request(
aws_secret_access_key=platform_creds.secretAccessKey,
aws_session_token=platform_creds.sessionToken,
)
provider_sess = None
if provider_creds:
provider_sess = boto3.Session(
aws_access_key_id=provider_creds.accessKeyId,
aws_secret_access_key=provider_creds.secretAccessKey,
aws_session_token=provider_creds.sessionToken,
)
action = Action[event.action]
callback_context = event.requestContext.get("callbackContext", {})
Credentials(**event_data["requestData"]["platformCredentials"])
except Exception as e: # pylint: disable=broad-except
LOG.exception("Invalid request")
raise InvalidRequest(f"{e} ({type(e).__name__})") from e
return caller_sess, platform_sess, request, action, callback_context, event
return (
(caller_sess, provider_sess, platform_sess),
request,
action,
callback_context,
event,
)

@_ensure_serialize
def __call__(
# TODO: refactor to reduce branching and locals
@_ensure_serialize # noqa: C901
def __call__( # pylint: disable=too-many-locals # noqa: C901
self, event_data: MutableMapping[str, Any], context: LambdaContext
) -> MutableMapping[str, Any]:
try:
ProviderLogHandler.setup(event_data)
parsed = self._parse_request(event_data)
caller_sess, platform_sess, request, action, callback, event = parsed
sessions, request, action, callback, event = self._parse_request(event_data)
caller_sess, provider_sess, platform_sess = sessions
metrics = MetricsPublisherProxy()
metrics.add_metrics_publisher(
MetricPublisher(event.awsAccountId, event.resourceType, platform_sess)
)
if provider_sess:
metrics.add_metrics_publisher(
MetricPublisher(
event.awsAccountId, event.resourceType, provider_sess
)
)

@tobywf tobywf Dec 2, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this looks like a case to me where breaking from the Java interface could be beneficial? we could backport changes or simply say it's different; the metrics stuff is old and the log delivery was patched in hastily from what i can see. for example, seems like the only difference between the publishers is the session. that means the proxy could take the awsAccountId and resourceType, and maybe add_metrics_publisher would just be called with credentials to add another one?

it's up to you. either one of us could do this in a separate PR, too. more of a brainstorm, since it's perfectly functional as is.

@jaymccon jaymccon Dec 4, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

let's get this merged to get the functionality available and can improve in future pr's

# Acknowledge the task for first time invocation
if not event.requestContext:
report_progress(
Expand All @@ -235,7 +254,20 @@ def __call__(
)
invoke = True
while invoke:
progress = self._invoke_handler(caller_sess, request, action, callback)
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
Expand Down
Loading