def efs_cw_metrics()

in source/lib/cloudwatch.py [0:0]

• 29 lines of code
• 4 McCabe index (conditional complexity)

    def efs_cw_metrics(self, efs_id, name):
        try:
            cw_client = boto3.client('cloudwatch')
            cw_metrics = {}
            metrics = {
                'BurstCreditBalance': 'Average',
                'PermittedThroughput': 'Average'
            }
            now = datetime.utcnow()
            start_time = now - timedelta(seconds=300)
            end_time = min(now, start_time + timedelta(seconds=3600))  # 5 min window
            for metric in metrics:
                data = cw_client.get_metric_statistics(
                    Namespace='AWS/EFS',
                    MetricName=metric,
                    Dimensions=[{
                        'Name': 'FileSystemId',
                        'Value': efs_id}],
                    Period=300,
                    StartTime=start_time,
                    EndTime=end_time,
                    Statistics=[metrics[metric]])['Datapoints']
                for d in data:
                    key = name + metric
                    value = Decimal(d[metrics[metric]])
                    cw_metrics[key] = value
            return cw_metrics
        except Exception as e:
            self.logger.error("unhandled exception: CloudWatchMetric_efs_cw_metrics", exc_info=1)