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

1# Copyright The Cloud Custodian Authors. 

2# SPDX-License-Identifier: Apache-2.0 

3from .core import Filter 

4from datetime import datetime, timedelta 

5from c7n.utils import type_schema, local_session, chunks 

6from c7n.query import RetryPageIterator 

7 

8 

9class ConsecutiveAwsBackupsFilter(Filter): 

10 """Returns resources where number of consective backups (based on the 

11 periodicity defined in the filter) is equal to/or greater than n units. 

12 This filter supports the resources that use AWS Backup service for backups. 

13 

14 :example: 

15 

16 .. code-block:: yaml 

17 

18 policies: 

19 - name: dynamodb-consecutive-aws-backup-count 

20 resource: dynamodb-table 

21 filters: 

22 - type: consecutive-aws-backups 

23 count: 7 

24 period: days 

25 status: 'COMPLETED' 

26 """ 

27 schema = type_schema('consecutive-aws-backups', count={'type': 'number', 'minimum': 1}, 

28 period={'enum': ['hours', 'days', 'weeks']}, 

29 status={'enum': ['COMPLETED', 'PARTIAL', 'DELETING', 'EXPIRED']}, 

30 required=['count', 'period', 'status']) 

31 permissions = ('backup:ListRecoveryPointsByResource', ) 

32 annotation = 'c7n:AwsBackups' 

33 

34 def process_resource_set(self, resources): 

35 arns = self.manager.get_arns(resources) 

36 

37 client = local_session(self.manager.session_factory).client('backup') 

38 paginator = client.get_paginator('list_recovery_points_by_resource') 

39 paginator.PAGE_ITERATOR_CLS = RetryPageIterator 

40 for r, arn in zip(resources, arns): 

41 r[self.annotation] = paginator.paginate( 

42 ResourceArn=arn).build_full_result().get('RecoveryPoints', []) 

43 

44 def get_date(self, time): 

45 period = self.data.get('period') 

46 if period == 'weeks': 

47 date = (datetime.utcnow() - timedelta(weeks=time)).strftime('%Y-%m-%d') 

48 elif period == 'hours': 

49 date = (datetime.utcnow() - timedelta(hours=time)).strftime('%Y-%m-%d-%H') 

50 else: 

51 date = (datetime.utcnow() - timedelta(days=time)).strftime('%Y-%m-%d') 

52 return date 

53 

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

55 results = [] 

56 retention = self.data.get('count') 

57 expected_dates = set() 

58 for time in range(1, retention + 1): 

59 expected_dates.add(self.get_date(time)) 

60 

61 for resource_set in chunks( 

62 [r for r in resources if self.annotation not in r], 50): 

63 self.process_resource_set(resource_set) 

64 

65 for r in resources: 

66 backup_dates = set() 

67 for backup in r[self.annotation]: 

68 if backup['Status'] == self.data.get('status'): 

69 if self.data.get('period') == 'hours': 

70 backup_dates.add(backup['CreationDate'].strftime('%Y-%m-%d-%H')) 

71 else: 

72 backup_dates.add(backup['CreationDate'].strftime('%Y-%m-%d')) 

73 

74 if expected_dates.issubset(backup_dates): 

75 results.append(r) 

76 return results