dag2_disk_usage.py 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  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. ## ARGS
  8. default_args = {
  9. 'owner': 'airflow',
  10. 'depends_on_past': False,
  11. 'start_date': days_ago(2),
  12. 'email_on_failure': False,
  13. 'email_on_retry': False,
  14. 'retries': 1,
  15. 'retry_delay': timedelta(minutes=5)
  16. }
  17. ## SSH HOOK
  18. orwell_ssh_hook = \
  19. SSHHook(
  20. key_file=Variable.get('hpc_ssh_key_file'),
  21. username=Variable.get('hpc_user'),
  22. remote_host=Variable.get('orwell_server_hostname'))
  23. woolf_ssh_hook = \
  24. SSHHook(
  25. key_file=Variable.get('hpc_ssh_key_file'),
  26. username=Variable.get('hpc_user'),
  27. remote_host=Variable.get('woolf_server_hostname'))
  28. eliot_ssh_hook = \
  29. SSHHook(
  30. key_file=Variable.get('hpc_ssh_key_file'),
  31. username=Variable.get('hpc_user'),
  32. remote_host=Variable.get('eliot_server_hostname'))
  33. igf_lims_ssh_hook = \
  34. SSHHook(
  35. key_file=Variable.get('hpc_ssh_key_file'),
  36. username=Variable.get('hpc_user'),
  37. remote_host=Variable.get('igf_lims_server_hostname'))
  38. ## DAG
  39. dag = \
  40. DAG(
  41. dag_id='dag2_disk_usage',
  42. schedule_interval="@hourly",
  43. max_active_runs=1,
  44. catchup=False,
  45. tags=['igf-lims','hpc','eliot','woolf','orwell'],
  46. default_args=default_args)
  47. with dag:
  48. ## TASK
  49. check_orwell_disk = \
  50. SSHOperator(
  51. task_id='check_orwell_disk',
  52. dag=dag,
  53. ssh_hook=orwell_ssh_hook,
  54. queue='hpc_4G',
  55. command='bash /home/igf/igf_code/IGF-cron-scripts/orwell/orwell_disk_usage.sh ')
  56. ## TASK
  57. check_eliot_disk = \
  58. SSHOperator(
  59. task_id='check_eliot_disk',
  60. dag=dag,
  61. ssh_hook=eliot_ssh_hook,
  62. queue='hpc_4G',
  63. command='bash /home/igf/git_repos/IGF-cron-scripts/eliot/eliot_disk_usage.sh ')
  64. ## TASK
  65. check_woolf_disk = \
  66. SSHOperator(
  67. task_id='check_woolf_disk',
  68. dag=dag,
  69. ssh_hook=woolf_ssh_hook,
  70. queue='hpc_4G',
  71. command='bash /home/igf/git_repos/IGF-cron-scripts/woolf/woolf_disk_usage.sh ')
  72. ## TASK
  73. check_igf_lims_disk = \
  74. SSHOperator(
  75. task_id='check_igf_lims_disk',
  76. dag=dag,
  77. ssh_hook=igf_lims_ssh_hook,
  78. queue='hpc_4G',
  79. command='bash /home/igf/github/IGF-cron-scripts/igf_lims/igf_lims_disk_usage.sh ')
  80. ## TASK
  81. merge_disk_usage = \
  82. SSHOperator(
  83. task_id='merge_disk_usage',
  84. dag=dag,
  85. ssh_hook=eliot_ssh_hook,
  86. queue='hpc_4G',
  87. command='bash /home/igf/git_repos/IGF-cron-scripts/eliot/merge_disk_usage.sh ')
  88. ## TASK
  89. internal_usage = \
  90. SSHOperator(
  91. task_id='internal_usage',
  92. dag=dag,
  93. ssh_hook=eliot_ssh_hook,
  94. queue='hpc_4G',
  95. command='bash /home/igf/git_repos/IGF-cron-scripts/eliot/internal_usage.sh ')
  96. ## PIPELINE
  97. merge_disk_usage << \
  98. [check_orwell_disk,
  99. check_eliot_disk,
  100. check_woolf_disk,
  101. check_igf_lims_disk]
  102. merge_disk_usage >> internal_usage