Merge pull request #262 from tulibraries/devo-62-update-prod-web-content-indeing-dag

DEVO-62: Update prod web-content indexing dag.

Author: dkinzer

cob_datapipeline/prod_sc_web_content_reindex_dag.py

@@ -1,13 +1,15 @@
 # Airflow DAG to index Web Content into SolrCloud.
 from datetime import datetime, timedelta
+from tulflow import tasks
 import airflow
 from airflow.models import Variable
 from airflow.hooks.base_hook import BaseHook
 from airflow.operators.bash_operator import BashOperator
 from airflow.operators.python_operator import PythonOperator
 from cob_datapipeline.task_slack_posts import web_content_slackpostonsuccess
 from cob_datapipeline.task_sc_get_num_docs import task_solrgetnumdocs
-from tulflow import tasks
+from cob_datapipeline.operators import\
+        PushVariable, DeleteAliasListVariable, DeleteCollectionListVariable
 
 """
 INIT SYSTEMWIDE VARIABLES
@@ -22,11 +24,11 @@
 
 # Get Solr URL & Collection Name for indexing info; error out if not entered
 SOLR_CONN = BaseHook.get_connection("SOLRCLOUD")
-WEB_CONTENT_SOLR_CONFIG = Variable.get("WEB_CONTENT_SOLR_CONFIG", deserialize_json=True)
+SOLR_CONFIG = Variable.get("WEB_CONTENT_SOLR_CONFIG", deserialize_json=True)
 # {"configset": "tul_cob-web-2", "replication_factor": 2}
-CONFIGSET = WEB_CONTENT_SOLR_CONFIG.get("configset")
+CONFIGSET = SOLR_CONFIG.get("configset")
 ALIAS = CONFIGSET + "-prod"
-REPLICATION_FACTOR = WEB_CONTENT_SOLR_CONFIG.get("replication_factor")
+REPLICATION_FACTOR = SOLR_CONFIG.get("replication_factor")
 WEB_CONTENT_BRANCH = Variable.get("WEB_CONTENT_PROD_BRANCH")
 
 # Manifold website creds
@@ -109,6 +111,34 @@
     ALIAS
 )
 
+PUSH_ALIAS = PushVariable(
+    task_id="push_alias",
+    name="WEB_CONTENT_QA_ALIASES",
+    value=ALIAS,
+    dag=DAG)
+
+DELETE_ALIAS = DeleteAliasListVariable(
+    task_id="delete_aliases",
+    solr_conn_id='SOLRCLOUD',
+    list_variable="WEB_CONTENT_QA_ALIASES",
+    skip_from_last=2,
+    skip_included=[ALIAS],
+    dag=DAG)
+
+PUSH_COLLECTION = PushVariable(
+    task_id="push_collection",
+    name="WEB_CONTENT_QA_COLLECTIONS",
+    value=CONFIGSET +"-{{ ti.xcom_pull(task_ids='set_collection_name') }}",
+    dag=DAG)
+
+DELETE_COLLECTIONS = DeleteCollectionListVariable(
+    task_id="delete_collections",
+    solr_conn_id='SOLRCLOUD',
+    list_variable="WEB_CONTENT_QA_COLLECTIONS",
+    skip_from_last=2,
+    skip_included=[CONFIGSET +"-{{ ti.xcom_pull(task_ids='set_collection_name') }}"],
+    dag=DAG)
+
 POST_SLACK = PythonOperator(
     task_id='slack_post_succ',
     python_callable=web_content_slackpostonsuccess,
@@ -122,4 +152,8 @@
 INDEX_WEB_CONTENT.set_upstream(CREATE_COLLECTION)
 GET_NUM_SOLR_DOCS_POST.set_upstream(INDEX_WEB_CONTENT)
 SOLR_ALIAS_SWAP.set_upstream(GET_NUM_SOLR_DOCS_POST)
-POST_SLACK.set_upstream(SOLR_ALIAS_SWAP)
+PUSH_ALIAS.set_upstream(SOLR_ALIAS_SWAP)
+DELETE_ALIAS.set_upstream(PUSH_ALIAS)
+PUSH_COLLECTION.set_upstream(DELETE_ALIAS)
+DELETE_COLLECTIONS.set_upstream(PUSH_COLLECTION)
+POST_SLACK.set_upstream(DELETE_COLLECTIONS)

cob_datapipeline/qa_sc_web_content_reindex_dag.py

@@ -11,7 +11,6 @@
 from cob_datapipeline.operators import\
         PushVariable, DeleteAliasListVariable, DeleteCollectionListVariable
 
-
 """
 INIT SYSTEMWIDE VARIABLES
 

tests/prod_sc_web_content_reindex_dag_test.py

@@ -28,6 +28,10 @@ def test_dag_tasks_present(self):
             "index_web_content",
             "get_num_solr_docs_post",
             "solr_alias_swap",
+            "push_alias",
+            "delete_aliases",
+            "push_collection",
+            "delete_collections",
             "slack_post_succ"
             ])
 
@@ -39,7 +43,11 @@ def test_dag_task_order(self):
             "index_web_content": "create_collection",
             "get_num_solr_docs_post": "index_web_content",
             "solr_alias_swap": "get_num_solr_docs_post",
-            "slack_post_succ": "solr_alias_swap",
+            "push_alias": "solr_alias_swap",
+            "delete_aliases": "push_alias",
+            "push_collection": "delete_aliases",
+            "delete_collections": "push_collection",
+            "slack_post_succ": "delete_collections",
         }
 
         for task, upstream_task in expected_task_deps.items():