dag2_disk_usage.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. from datetime import timedelta
  2. from airflow.models import DAG,Variable
  3. from airflow.utils.dates import days_ago
  4. from airflow.operators.bash_operator import BashOperator
  5. from airflow.contrib.operators.ssh_operator import SSHOperator
  6. from airflow.contrib.hooks.ssh_hook import SSHHook
  7. default_args = {
  8. 'owner': 'airflow',
  9. 'depends_on_past': False,
  10. 'start_date': days_ago(2),
  11. 'email_on_failure': False,
  12. 'email_on_retry': False,
  13. 'retries': 1,
  14. 'retry_delay': timedelta(minutes=5)
  15. }
  16. orwell_ssh_hook = \
  17. SSHHook(
  18. key_file=Variable.get('hpc_ssh_key_file'),
  19. username=Variable.get('hpc_user'),
  20. remote_host='orwell.hh.med.ic.ac.uk')
  21. woolf_ssh_hook = \
  22. SSHHook(
  23. key_file=Variable.get('hpc_ssh_key_file'),
  24. username=Variable.get('hpc_user'),
  25. remote_host='woolf.med.ic.ac.uk')
  26. eliot_ssh_hook = \
  27. SSHHook(
  28. key_file=Variable.get('hpc_ssh_key_file'),
  29. username=Variable.get('hpc_user'),
  30. remote_host='eliot.med.ic.ac.uk')
  31. igf_lims_ssh_hook = \
  32. SSHHook(
  33. key_file=Variable.get('hpc_ssh_key_file'),
  34. username=Variable.get('hpc_user'),
  35. remote_host='igf-lims.cc.ic.ac.uk')
  36. dag = \
  37. DAG(
  38. dag_id='dag2_disk_usage',
  39. schedule_interval="@hourly",
  40. max_active_runs=1,
  41. catchup=False,
  42. tags=['igf-lims','hpc','eliot','woolf','orwell'],
  43. default_args=default_args)
  44. with dag:
  45. check_orwell_disk = \
  46. SSHOperator(
  47. task_id = 'check_orwell_disk',
  48. dag = dag,
  49. ssh_hook = orwell_ssh_hook,
  50. queue='hpc_4G',
  51. command = 'bash /home/igf/igf_code/IGF-cron-scripts/orwell/orwell_disk_usage.sh '
  52. )
  53. check_eliot_disk = \
  54. SSHOperator(
  55. task_id = 'check_eliot_disk',
  56. dag = dag,
  57. ssh_hook = eliot_ssh_hook,
  58. queue='hpc_4G',
  59. command = 'bash /home/igf/git_repos/IGF-cron-scripts/eliot/eliot_disk_usage.sh '
  60. )
  61. check_woolf_disk = \
  62. SSHOperator(
  63. task_id = 'check_woolf_disk',
  64. dag = dag,
  65. ssh_hook = woolf_ssh_hook,
  66. queue='hpc_4G',
  67. command = 'bash /home/igf/git_repos/IGF-cron-scripts/woolf/woolf_disk_usage.sh '
  68. )
  69. check_igf_lims_disk = \
  70. SSHOperator(
  71. task_id = 'check_igf_lims_disk',
  72. dag = dag,
  73. ssh_hook = igf_lims_ssh_hook,
  74. queue='hpc_4G',
  75. command = 'bash /home/igf/github/IGF-cron-scripts/igf_lims/igf_lims_disk_usage.sh '
  76. )
  77. merge_disk_usage = \
  78. SSHOperator(
  79. task_id = 'merge_disk_usage',
  80. dag = dag,
  81. ssh_hook = eliot_ssh_hook,
  82. queue='hpc_4G',
  83. command = 'bash /home/igf/git_repos/IGF-cron-scripts/eliot/merge_disk_usage.sh '
  84. )
  85. internal_usage = \
  86. SSHOperator(
  87. task_id = 'internal_usage',
  88. dag = dag,
  89. ssh_hook = eliot_ssh_hook,
  90. queue='hpc_4G',
  91. command = 'bash /home/igf/git_repos/IGF-cron-scripts/eliot/internal_usage.sh '
  92. )
  93. merge_disk_usage << [check_orwell_disk,check_eliot_disk,check_woolf_disk,check_igf_lims_disk]
  94. merge_disk_usage >> internal_usage