Correct data refresh status message next steps (#953)

* Correct status message order next steps

* Add delete index recipe

* Add slack "status" method for prepending model names

* Further message improving, use new status method

* Make parameters more consistent

* Add one last "next" step

* Differentiate between ingestion server steps and all steps

Author: AetherUnbound

ingestion_server/ingestion_server/api.py

@@ -269,7 +269,10 @@ def on_post(self, req, _):
             index_type = target_index.split("-")[0]
             if index_type not in MEDIA_TYPES:
                 index_type = "image"
-            slack.verbose(f"`{index_type}`: Elasticsearch reindex complete")
+            slack.verbose(
+                f"`{index_type}`: Elasticsearch reindex complete | "
+                f"_Next: re-apply indices & constraints_"
+            )
 
             elasticsearch = elasticsearch_connect()
             indexer = TableIndexer(

ingestion_server/ingestion_server/indexer.py

@@ -378,7 +378,7 @@ def point_alias(self, model_name: str, index_suffix: str, alias: str, **_):
         if alias_stat.exists:
             if not alias_stat.is_alias:
                 # Alias is an index, this is fatal.
-                message = f"There is an index named {alias}, cannot proceed."
+                message = f"There is an index named `{alias}`, cannot proceed."
                 log.error(message)
                 slack.error(message)
                 return
@@ -395,20 +395,23 @@ def point_alias(self, model_name: str, index_suffix: str, alias: str, **_):
                     }
                 )
                 message = (
-                    f"Migrated alias {alias} "
-                    f"from index {curr_index} to index {dest_index}."
+                    f"Migrated alias `{alias}` from index `{curr_index}` to "
+                    f"index `{dest_index}` | _Next: delete old index_"
                 )
                 log.info(message)
-                slack.info(message)
+                slack.status(model_name, message)
             else:
                 # Alias is already mapped.
-                log.info(f"Alias {alias} already points to index {dest_index}.")
+                log.info(
+                    f"`{model_name}`: Alias `{alias}` already points to "
+                    f"index `{dest_index}`."
+                )
         else:
             # Alias does not exist, create it.
             self.es.indices.put_alias(index=dest_index, name=alias)
-            message = f"Created alias {alias} pointing to index {dest_index}."
+            message = f"Created alias `{alias}` pointing to index `{dest_index}`."
             log.info(message)
-            slack.info(message)
+            slack.status(model_name, message)
 
         if self.progress is not None:
             self.progress.value = 100  # mark job as completed
@@ -441,7 +444,7 @@ def delete_index(
                     if self.is_bad_request is not None:
                         self.is_bad_request.value = 1
                     message = (
-                        f"Alias {target} might be in use so it cannot be deleted. "
+                        f"Alias `{target}` might be in use so it cannot be deleted. "
                         f"Verify that the API does not use this alias and then use the "
                         f"`force_delete` parameter."
                     )
@@ -455,24 +458,24 @@ def delete_index(
                     if self.is_bad_request is not None:
                         self.is_bad_request.value = 1
                     message = (
-                        f"Index {target} is associated with aliases "
+                        f"Index `{target}` is associated with aliases "
                         f"{target_stat.alt_names}, cannot delete. Delete aliases first."
                     )
                     log.error(message)
                     slack.error(message)
                     return
 
             self.es.indices.delete(index=target)
-            message = f"Index {target} was deleted."
+            message = f"Index `{target}` was deleted - data refresh complete! :tada:"
             log.info(message)
-            slack.info(message)
+            slack.status(model_name, message)
         else:
             # Cannot delete as target does not exist.
             if self.is_bad_request is not None:
                 self.is_bad_request.value = 1
-            message = f"Target {target} does not exist and cannot be deleted."
+            message = f"Target `{target}` does not exist and cannot be deleted."
             log.info(message)
-            slack.info(message)
+            slack.status(model_name, message)
 
         if self.progress is not None:
             self.progress.value = 100

ingestion_server/ingestion_server/ingest.py

@@ -271,7 +271,10 @@ def refresh_api_table(
     """
 
     # Step 1: Get the list of overlapping columns
-    slack.info(f"`{table}`: Starting data refresh | _Next: copying data from upstream_")
+    slack.status(
+        table,
+        "Starting ingestion server data refresh | _Next: copying data from upstream_",
+    )
     downstream_db = database_connect()
     upstream_db = psycopg2.connect(
         dbname=UPSTREAM_DB_NAME,
@@ -319,21 +322,15 @@ def refresh_api_table(
         log.info(f"Running copy-data query: \n{copy_data.as_string(downstream_cur)}")
         downstream_cur.execute(copy_data)
 
-    next_step = (
-        "_Next: {starting data cleaning}_"
-        if table == "image"
-        else "Finished refreshing table"
-    )
-    slack.verbose(f"`{table}`: Data copy complete | {next_step}")
+    next_step = "image data cleaning" if table == "image" else "Elasticsearch reindex"
+    slack.status(table, f"Data copy complete | _Next: {next_step}_")
 
     if table == "image":
         # Step 5: Clean the data
         log.info("Cleaning data...")
         clean_image_data(table)
         log.info("Cleaning completed!")
-        slack.verbose(
-            f"`{table}`: Data cleaning complete | " f"Finished refreshing table"
-        )
+        slack.status(table, "Data cleaning complete | _Next: Elasticsearch reindex_")
 
     downstream_db.close()
     log.info(f"Finished refreshing table '{table}'.")
@@ -378,17 +375,14 @@ def promote_api_table(
                 downstream_cur.execute(remap_constraint)
         log.info("Done remapping constraints! Going live with new table...")
         _update_progress(progress, 99.0)
-        slack.verbose(
-            f"`{table}`: Indices & constraints applied, finished refreshing table | "
-            f"_Next: Elasticsearch reindex_"
-        )
+        slack.status(table, "Indices & constraints applied | _Next: table promotion_")
 
         # Step 8: Promote the temporary table and delete the original
         go_live = get_go_live_query(table, index_mapping)
         log.info(f"Running go-live: \n{go_live.as_string(downstream_cur)}")
         downstream_cur.execute(go_live)
-        slack.verbose(
-            f"`{table}`: Finished table promotion | " f"_Next: Elasticsearch promotion_"
+        slack.status(
+            table, "Finished table promotion | _Next: Elasticsearch promotion_"
         )
 
     downstream_db.close()

ingestion_server/ingestion_server/slack.py

@@ -70,3 +70,13 @@ def info(text: str, summary: str = None) -> None:
 
 def error(text: str, summary: str = None) -> None:
     _message(text, summary, level=Level.ERROR)
+
+
+def status(model: str, text: str) -> None:
+    """
+    Send a message regarding the status of the data refresh.
+
+    Model is required an all messages get prepended with the model.
+    """
+    text = f"`{model}`: {text}"
+    info(text, None)

justfile

@@ -179,6 +179,10 @@ _ing-api data port="50281":
 @promote model="image" suffix="init" alias="image":
     just _ing-api '{"model": "{{ model }}", "action": "PROMOTE", "index_suffix": "{{ suffix }}", "alias": "{{ alias }}"}'
 
+# Delete an index in Elasticsearch
+@delete model="image" suffix="init" alias="image":
+    just _ing-api '{"model": "{{ model }}", "action": "DELETE_INDEX", "index_suffix": "{{ suffix }}"}'
+
 # Run ingestion-server tests locally
 ing-testlocal *args:
     cd ingestion_server && pipenv run ./test/run_test.sh {{ args }}