Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/c7n/filters/costhub.py: 44%

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3from c7n.config import Bag 

4from c7n.manager import resources 

5from c7n.utils import local_session, type_schema, filter_empty 

6from .core import Filter, ListItemResourceManager 

7 

8 

9class RecommendationManager(ListItemResourceManager): 

10 

11 model = Bag(id='recommendationId') 

12 

13 @classmethod 

14 def get_model(cls): 

15 return cls.model 

16 

17 

18class CostHubRecommendation(Filter): 

19 """Cost optimization hub recommendations. 

20 

21 .. code-block:: yaml 

22 

23 - name: cost-ec2-optimize 

24 resource: aws.ec2 

25 filters: 

26 - type: cost-optimization 

27 attrs: 

28 - actionType: Rightsize 

29 - key: recommendationLookbackPeriodInDays 

30 op: gte 

31 value: 10 

32 - key: estimatedMonthlySavings 

33 value: 30 

34 op: gte 

35 

36 """ 

37 

38 schema = type_schema( 

39 'cost-optimization', 

40 efforts={ 

41 'type': 'array', 

42 'items': {'enum': ['VeryLow', 'Low', 'Medium', 'High', 'VeryHigh']}, 

43 }, 

44 action={ 

45 'enum': [ 

46 'Rightsize', 

47 'Stop', 

48 'Upgrade', 

49 'PurchaseSavingsPlans', 

50 'PurchaseReservedInstances', 

51 'MigrateToGraviton', 

52 ] 

53 }, 

54 attrs={'$ref': '#/definitions/filters_common/list_item_attrs'}, 

55 ) 

56 schema_alias = True 

57 resource_type_map = { 

58 'ec2': 'Ec2Instance', 

59 'ebs': 'EbsVolume', 

60 'lambda': 'LambdaFunction', 

61 'ecs-service': 'EcsService', 

62 'asg': 'Ec2AutoScalingGroup', 

63 } 

64 default_action = { 

65 'ec2': 'Rightsize', 

66 'ebs': 'Upgrade', 

67 'ecs-service': 'Rightsize', 

68 'lambda': 'Rightsize', 

69 } 

70 

71 permissions = ('cost-optimization-hub:ListRecommendations',) 

72 annotation_key = "c7n:cost_optimize" 

73 

74 def process(self, resources, event=None): 

75 client = local_session(self.manager.session_factory).client( 

76 'cost-optimization-hub', region_name='us-east-1') 

77 filter_params = filter_empty({ 

78 'actionTypes': [ 

79 self.manager.data.get( 

80 'action', self.default_action[self.manager.type] 

81 ) 

82 ], 

83 'regions': [self.manager.config.region] or None, 

84 'resourceTypes': [self.resource_type_map[self.manager.type]], 

85 }) 

86 r_map = {} 

87 for arn, r in zip(self.manager.get_arns(resources), resources): 

88 r_map[arn] = r 

89 

90 pager = client.get_paginator('list_recommendations') 

91 recommendations = pager.paginate(filter=filter_params).build_full_result() 

92 

93 results = set() 

94 frm = RecommendationManager(self.manager.ctx, data={'filters': self.data.get('attrs', [])}) 

95 

96 for rec in recommendations['items']: 

97 rec_rarn = rec['resourceArn'] 

98 

99 # a few of the recommendation resources use a version 

100 # qualifier which won't match the innate/latest arn coming 

101 # from describe sources (sans qualifier) 

102 if rec_rarn.count(':') == 7: 

103 rec_rarn, _ = rec_rarn.rsplit(':', 1) 

104 

105 if rec_rarn not in r_map: 

106 continue 

107 if not frm.filter_resources([rec], event): 

108 continue 

109 r = r_map[rec_rarn] 

110 r[self.annotation_key] = rec 

111 results.add(rec_rarn) 

112 return [r for rid, r in r_map.items() if rid in results] 

113 

114 @classmethod 

115 def register_resources(klass, registry, resource_class): 

116 if resource_class.type in klass.resource_type_map: 

117 resource_class.filter_registry.register('cost-optimization', klass) 

118 

119 

120resources.subscribe(CostHubRecommendation.register_resources)