revert big list update (#1683)

* Revert "update ingest script to handle big lists (#1680)"

This reverts commit c4bc2cb.

* update execution timeouts defaults for harvest tasks.

Author: dkinzer

cob_datapipeline/catalog_preproduction_oai_harvest_dag.py

@@ -185,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,
@@ -201,6 +201,7 @@
         "DATA": "{{ ti.xcom_pull(task_ids='list_updated_files') | tojson }}",
     }},
     trigger_rule="none_failed_min_one_success",
+    execution_timeout=timedelta(hours=24),
     dag=DAG
 )
 
@@ -215,7 +216,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

@@ -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(
@@ -176,7 +177,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,
@@ -192,6 +193,7 @@
         "DATA": "{{ ti.xcom_pull(task_ids='list_updated_files') | tojson }}",
     }},
     trigger_rule="none_failed_min_one_success",
+    execution_timeout=timedelta(hours=24),
     dag=DAG
 )
 
@@ -206,7 +208,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
 
-ingest_marc_file() {
-  set -eo pipefail  # Ensure failures stop execution inside this function
-  cd tmp/cob_index || exit 1
-  file="$1"
+for file in $data_in
+do
   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,15 +34,4 @@ ingest_marc_file() {
   if [ "$file" != "$processed_file" ]; then
     aws s3 mv s3://$BUCKET/$file s3://$BUCKET/$processed_file
   fi
-}
-
-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 "{}"'
-
+done

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)