astro migration

[keegansmith21]

Astro Migration

This PR refactors the codebase for use with The Astronomer Airflow service. Broadly speaking, the changes are as follows:

• Change DAG creation to use the Airflow Taskflow API
• Upgrade to Ariflow 2.7.3
• Removed setup.py and setup.cfg - oaebu-workflows is no longer an installable package
• Created Dockerfile, which inherits from the Astro airflow image. Used to ship to Astro deployment
• Move all workflow files (and their dependent files) into the dag folder (ready for Astro shipping)
• Move tests to the ./tests folder
• Update the API usage to the new Dataset API (Refactor to support Astro observatory-platform#653) that uses BigQuery as a storage service for Release objects
• Update the workflow Release objects to be passed as Xcoms between tasks (a benefit of the taskflow upgrade)
• Collapse the upload tasks into their respective file creation tasks. i.e. upload_downloaded() is absorbed into download(). This is necessary now as we no longer necessarily have persistent disks in Astro deployments.
• Bigquery tasks create the client on behalf of their appropriate publisher, rather than have them all invoke the default environment project.
• Updated telescopes that can have multiple releases to use the new dynamic tasks functionality of the taskflow API

[jdddog]

Hey Keegan, nice work on getting the migration done so quickly. See below for some comments on a few things that could be changed.

[codecov]

Codecov Report

Attention: Patch coverage is 94.78114% with 93 lines in your changes are missing coverage. Please review.

Project coverage is 93.76%. Comparing base (3ddef54) to head (18c8f0e).

Files	Patch %	Lines
...s/google_books_telescope/google_books_telescope.py	89.56%	8 Missing and 11 partials ⚠️
...ags/oaebu_workflows/onix_workflow/onix_workflow.py	96.34%	12 Missing and 7 partials ⚠️
...s/oaebu_workflows/onix_telescope/onix_telescope.py	91.53%	7 Missing and 4 partials ⚠️
...ucl_discovery_telescope/ucl_discovery_telescope.py	95.50%	4 Missing and 5 partials ⚠️
...flows/irus_oapen_telescope/irus_oapen_telescope.py	96.17%	4 Missing and 4 partials ⚠️
...oaebu_workflows/jstor_telescope/jstor_telescope.py	94.59%	5 Missing and 3 partials ⚠️
...s/irus_fulcrum_telescope/irus_fulcrum_telescope.py	95.42%	3 Missing and 4 partials ⚠️
...oaebu_workflows/thoth_telescope/thoth_telescope.py	93.39%	4 Missing and 3 partials ⚠️
...pen_metadata_telescope/oapen_metadata_telescope.py	95.83%	3 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #212      +/-   ##
==========================================
- Coverage   94.33%   93.76%   -0.58%     
==========================================
  Files          18       14       -4     
  Lines        2771     2806      +35     
  Branches      399      353      -46     
==========================================
+ Hits         2614     2631      +17     
- Misses         80       96      +16     
- Partials       77       79       +2     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[jdddog]

Thanks for the changes Keegan, it is looking really good.

Just a few small comments.

We were settings a few parameters in the DAG constructor by default that differ to Airflow defaults (unless we override these in a workflow subclass).

• max_active_runs was set to 1 (Airflow pulls this from airflow.cfg if not specified)
• catchup was set to False (Airflow defaults this to True)

Could you double check that these are all set to how they were and if not, if the new default settings are OK?

We used to set "on_failure_callback": on_failure_callback, in default_args which creates the Slack message on failure, could we set this for each DAG?

Perhaps retries and retry_delay should be keyword argument parameters in create_dag?

[keegansmith21]

Thanks for the changes Keegan, it is looking really good.

Just a few small comments.

We were settings a few parameters in the DAG constructor by default that differ to Airflow defaults (unless we override these in a workflow subclass).

• max_active_runs was set to 1 (Airflow pulls this from airflow.cfg if not specified)
• catchup was set to False (Airflow defaults this to True)

Could you double check that these are all set to how they were and if not, if the new default settings are OK?

We used to set "on_failure_callback": on_failure_callback, in default_args which creates the Slack message on failure, could we set this for each DAG?

Perhaps retries and retry_delay should be keyword argument parameters in create_dag?

Hey Jamie,

Thanks for taking another look at this PR. I've made the following changes:

• Added a max_active_runs=1 default where it was not already set
• Added the on_failure_callback from observatory platform to all dags
• Made the retries and retry_delay parameters arguments for the create_dag function with defaults set to 3 and 5 (minutes) respectively

The catchup was already set for each dag as part of the create_dag defaults.

I also found the default airlfow config args here . But it was kind of hard to find so I much prefer explicitly stating the defaults in create_dag.