Coverage for backpack/cwadapter.py: 100%

1''' Reports timer statistics to AWS CloudWatch Metrics. ''' 

2 

3import datetime 

4import logging 

5from typing import Optional, Dict 

6 

7import boto3 

8import botocore 

9 

10from .timepiece import BaseTimer 

11 

12class CloudWatchTimerAdapter: 

13 ''' Reports timer statistics to AWS CloudWatch Metrics. 

14 

15 The IAM policy associated with the Panorama Application Role of this app should grant 

16 the execution of `cloudwatch:PutMetricData` operation. 

17 

18 Args: 

19 namespace (str): The name of the CloudWatch namespace of this custom metrics. 

20 It can be for example the name of your project. 

21 metric_name (str): The name of the CloudWatch metrics. This can be for example 

22 `frame_processing_time`, if you use CWTachometer to measure frame processing 

23 time statistics. 

24 dimensions (Optional[Dict[str, str]]): Additional CloudWatch metrics dimensions of this 

25 metric. This can be for example the device and application identifier. 

26 region (Optional[str]): The AWS region of the CloudWatch metrics. 

27 boto3_session (Optional[boto3.Session]): The boto3 session to be used for sending the 

28 CloudWatch metrics. If left to None, CWTachometer will use the default session. If the 

29 default session does not have a default region configured, you might get errors. 

30 parent_logger (Optional[logging.Logger]): If you want to connect the logger of this class 

31 to a parent, specify it here. 

32 ''' 

33 

34 def __init__(self, 

35 namespace: str, 

36 metric_name: str, 

37 dimensions: Optional[Dict[str, str]] = None, 

38 region: Optional[str] = None, 

39 boto3_session: Optional[boto3.Session] = None, 

40 parent_logger: Optional[logging.Logger] = None 

41 ): 

42 self.logger = ( 

43 logging.getLogger(self.__class__.__name__) if parent_logger is None else 

44 parent_logger.getChild(self.__class__.__name__) 

45 ) 

46 self.dimensions = CloudWatchTimerAdapter._cw_dimensions(dimensions or {}) 

47 self.namespace = namespace 

48 self.metric_name = metric_name 

49 boto3_session = boto3_session or boto3.Session(region_name=region) 

50 self.cloudwatch = boto3_session.client('cloudwatch') 

51 

52 @staticmethod 

53 def _cw_dimensions(dimensions): 

54 return [{ 'Name': name, 'Value': value } for name, value in dimensions.items()] 

55 

56 def send_metrics(self, timestamp: datetime.datetime, timer: BaseTimer) -> None: 

57 ''' Sends timer statistics to CloudWatch. 

58 

59 This method can be used as a callback in Tachometer instances. 

60 

61 For example:: 

62 

63 cw_adapter = CloudWatchTimerAdapter( 

64 namespace='my_namespace', 

65 metric_name='my_metric', 

66 dimensions={'foo': 'bar'} 

67 ) 

68 tacho = TickerTachometer( 

69 stats_callback=cw_adapter.send_metrics 

70 ) 

71 tacho.tick() 

72 

73 Args: 

74 timestamp (datetime.datetime): The timestamp the statistics refers to. 

75 timer (BaseTimer): The timer that collected the statistics. 

76 ''' 

77 metric_data = { 

78 'MetricName': self.metric_name, 

79 'Dimensions': self.dimensions, 

80 'Timestamp': timestamp.astimezone(datetime.timezone.utc), 

81 'StatisticValues': { 

82 'SampleCount': timer.len(), 

83 'Sum': timer.sum(), 

84 'Minimum': timer.min(), 

85 'Maximum': timer.max() 

86 }, 

87 'Unit': 'Seconds' 

88 } 

89 try: 

90 self.cloudwatch.put_metric_data( 

91 Namespace=self.namespace, 

92 MetricData=[metric_data] 

93 ) 

94 except botocore.exceptions.ClientError as error: 

95 self.logger.warning('Couldn\'t put data for metric %s.%s', 

96 self.namespace, self.metric_name) 

97 self.logger.warning(str(error)) 

98 except AttributeError: 

99 self.logger.warning('CloudWatch client is not available.') 

100 return metric_data