Coverage for portality/tasks/prune_es_backups.py: 52%

1from portality import models 

2from portality.core import app 

3from portality.lib import dates 

4 

5from portality.tasks.redis_huey import long_running, schedule 

6from portality.decorators import write_required 

7from portality.background import BackgroundTask, BackgroundApi 

8 

9from esprit.raw import Connection 

10from esprit.snapshot import ESSnapshotsClient 

11 

12from datetime import datetime, timedelta 

13# FIXME: update for index-per-type 

14 

15 

16class PruneESBackupsBackgroundTask(BackgroundTask): 

17 

18 __action__ = "prune_es_backups" 

19 

20 def run(self): 

21 """ 

22 Execute the task as specified by the background_job 

23 :return: 

24 """ 

25 job = self.background_job 

26 

27 # Connection to the ES index 

28 conn = Connection(app.config.get("ELASTIC_SEARCH_HOST"), index='_snapshot') 

29 

30 snap_ttl = app.config.get('ELASTIC_SEARCH_SNAPSHOT_TTL', 366) 

31 snap_thresh = datetime.utcnow() - timedelta(days=snap_ttl) 

32 job.add_audit_message('Deleting backups older than {}'.format(dates.format(snap_thresh))) 

33 

34 client = ESSnapshotsClient(conn, app.config['ELASTIC_SEARCH_SNAPSHOT_REPOSITORY']) 

35 client.prune_snapshots(snap_ttl, self.report_deleted_closure(job)) 

36 

37 def cleanup(self): 

38 """ 

39 Cleanup after a successful OR failed run of the task 

40 :return: 

41 """ 

42 pass 

43 

44 @staticmethod 

45 def report_deleted_closure(job): 

46 """ Report the deletion to the background task audit log """ 

47 def _report_deleted_callback(snapshot, status_code, result): 

48 job.add_audit_message('Deleted snapshot {0}, Status code: {1}, Successful: {2}'.format(snapshot.name, status_code, result)) 

49 

50 return _report_deleted_callback 

51 

52 @classmethod 

53 def prepare(cls, username, **kwargs): 

54 """ 

55 Take an arbitrary set of keyword arguments and return an instance of a BackgroundJob, 

56 or fail with a suitable exception 

57 

58 :param username: The user this job will run under 

59 :param kwargs: arbitrary keyword arguments pertaining to this task type 

60 :return: a BackgroundJob instance representing this task 

61 """ 

62 

63 # first prepare a job record 

64 job = models.BackgroundJob() 

65 job.user = username 

66 job.action = cls.__action__ 

67 return job 

68 

69 @classmethod 

70 def submit(cls, background_job): 

71 """ 

72 Submit the specified BackgroundJob to the background queue 

73 

74 :param background_job: the BackgroundJob instance 

75 :return: 

76 """ 

77 background_job.save() 

78 prune_es_backups.schedule(args=(background_job.id,), delay=10) 

79 

80 

81@long_running.periodic_task(schedule("prune_es_backups")) 

82@write_required(script=True) 

83def scheduled_prune_es_backups(): 

84 user = app.config.get("SYSTEM_USERNAME") 

85 job = PruneESBackupsBackgroundTask.prepare(user) 

86 PruneESBackupsBackgroundTask.submit(job) 

87 

88 

89@long_running.task() 

90@write_required(script=True) 

91def prune_es_backups(job_id): 

92 job = models.BackgroundJob.pull(job_id) 

93 task = PruneESBackupsBackgroundTask(job) 

94 BackgroundApi.execute(task)