update ingest script to handle big lists (#1680)

* Update ingest script to handle larger lists of files.

* Add a task execution timeout.  Fail if any task takes longer than 10 hours.

* Fix test.

* Fix tests.

Author: dkinzer

cob_datapipeline/catalog_preproduction_oai_harvest_dag.py

@@ -80,7 +80,8 @@
     "start_date": pendulum.datetime(2018, 12, 13, tz="America/New_York"),
     "on_failure_callback": [slackpostonfail],
     "retries": 0,
-    "retry_delay": timedelta(minutes=10)
+    "retry_delay": timedelta(minutes=10),
+    "execution_timeout": timedelta(hours=10)
 }
 
 DAG = airflow.DAG(
@@ -184,7 +185,7 @@
 
 INDEX_UPDATES_OAI_MARC = BashOperator(
     task_id="index_updates_oai_marc",
-    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh ",
+    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh",
     env={**os.environ, **{
         "AWS_ACCESS_KEY_ID": AIRFLOW_S3.login,
         "AWS_SECRET_ACCESS_KEY": AIRFLOW_S3.password,
@@ -214,7 +215,7 @@
 
 INDEX_DELETES_OAI_MARC = BashOperator(
     task_id="index_deletes_oai_marc",
-    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh ",
+    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh",
     env={**os.environ, **{
         "AWS_ACCESS_KEY_ID": AIRFLOW_S3.login,
         "AWS_SECRET_ACCESS_KEY": AIRFLOW_S3.password,

cob_datapipeline/catalog_production_oai_harvest_dag.py

@@ -176,7 +176,7 @@
 
 INDEX_UPDATES_OAI_MARC = BashOperator(
     task_id="index_updates_oai_marc",
-    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh ",
+    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh",
     env={**os.environ, **{
         "AWS_ACCESS_KEY_ID": AIRFLOW_S3.login,
         "AWS_SECRET_ACCESS_KEY": AIRFLOW_S3.password,
@@ -206,7 +206,7 @@
 
 INDEX_DELETES_OAI_MARC = BashOperator(
     task_id="index_deletes_oai_marc",
-    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh ",
+    bash_command=AIRFLOW_HOME + "/dags/cob_datapipeline/scripts/ingest_marc.sh",
     env={**os.environ, **{
         "AWS_ACCESS_KEY_ID": AIRFLOW_S3.login,
         "AWS_SECRET_ACCESS_KEY": AIRFLOW_S3.password,

cob_datapipeline/scripts/ingest_marc.sh

@@ -18,14 +18,14 @@ gem install bundler
 bundle config set force_ruby_platform true
 bundle install --without=debug
 
-data_in=$(echo $DATA | jq -r '.[]')
-
 if [ -z "$COMMAND" ]; then
   COMMAND=ingest
 fi
 
-for file in $data_in
-do
+ingest_marc_file() {
+  set -eo pipefail  # Ensure failures stop execution inside this function
+  cd tmp/cob_index || exit 1
+  file="$1"
   echo "Indexing file: "$file
   bundle exec cob_index $COMMAND $(aws s3 presign s3://$BUCKET/$file)
   processed_file=$(echo $file | sed 's/new-updated/processed-new-updated/' | sed 's/deleted/processed-deleted/')
@@ -34,4 +34,15 @@ do
   if [ "$file" != "$processed_file" ]; then
     aws s3 mv s3://$BUCKET/$file s3://$BUCKET/$processed_file
   fi
-done
+}
+
+export -f ingest_marc_file
+export COMMAND
+export BUCKET
+export SOLR_AUTH_USER
+export SOLR_AUTH_PASSWORD
+export SOLR_URL
+export ALMAOAI_LAST_HARVEST_FROM_DATE
+
+echo "$DATA" | jq -r '.[]' | xargs -n 1 -I {} bash -c 'igest_marc_file "{}"'
+

tests/catalog_preproduction_oai_harvest_dag_test.py

@@ -12,13 +12,13 @@ def setUp(self):
 
     def test_index_deletes_oai_marc(self):
         task = DAG.get_task("index_deletes_oai_marc")
-        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh "
+        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh"
         self.assertEqual(task.env["COMMAND"], "delete --suppress")
         self.assertEqual(task.bash_command, expected_bash_path)
 
     def test_index_ingest_oai_marc(self):
         task = DAG.get_task("index_updates_oai_marc")
-        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh "
+        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh"
         self.assertEqual(task.env["COMMAND"], "ingest")
         self.assertEqual(task.bash_command, expected_bash_path)
 

tests/catalog_production_oai_harvest_dag_test.py

@@ -12,12 +12,12 @@ def setUp(self):
 
     def test_index_deletes_oai_marc(self):
         task = DAG.get_task("index_deletes_oai_marc")
-        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh "
+        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh"
         self.assertEqual(task.env["COMMAND"], "delete --suppress")
         self.assertEqual(task.bash_command, expected_bash_path)
 
     def test_index_ingest_oai_marc(self):
         task = DAG.get_task("index_updates_oai_marc")
-        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh "
+        expected_bash_path = self.airflow_home + "/dags/cob_datapipeline/scripts/ingest_marc.sh"
         self.assertEqual(task.env["COMMAND"], "ingest")
         self.assertEqual(task.bash_command, expected_bash_path)