logs + comments

Author: aschey-forpeople

apps/bfd-pipeline/bfd-pipeline-idr/extractor.py

@@ -75,6 +75,8 @@ def extract_idr_data(
             return res
 
         previous_batch_complete = progress.batch_complete_ts >= progress.batch_start_ts
+        logger.info("previous batch complete: %s", previous_batch_complete)
+
         compare_timestamp = progress.batch_start_ts if previous_batch_complete else progress.last_ts
 
         idr_query_timer.start()
@@ -179,6 +181,9 @@ def extract_many(self, cls: type[T], sql: str, params: dict[str, DbType]) -> Ite
                     batch = cur.fetchmany(self.batch_size)  # type: ignore[assignment]
                     cursor_fetch_timer.stop()
                 return
+            # Snowflake will throw a reauth error if the pipeline has been running for several hours
+            # but it seems to be wrapped in a ProgrammingError.
+            # Unclear the best way to handle this, it will require a bit more trial and error
             except (ReauthenticationRequest, RetryRequest, ProgrammingError) as ex:
                 logger.warning("received transient error, retrying...", exc_info=ex)
                 if attempt == max_attempts - 1:

apps/bfd-pipeline/bfd-pipeline-idr/pipeline.py

@@ -83,7 +83,13 @@ def extract_and_load(
     batch_start = datetime.now()
     progress = get_progress(connection_string, cls.table(), batch_start)
 
-    logger.info("progress for %s - %s", cls.table(), progress.last_ts if progress else "none")
+    logger.info(
+        "progress for %s - last_ts: %s batch_start_ts: %s batch_complete_ts: %s",
+        cls.table(),
+        progress.last_ts if progress else "none",
+        progress.batch_start_ts if progress else "none",
+        progress.batch_complete_ts if progress else "none",
+    )
     data_iter = data_extractor.extract_idr_data(cls, progress, batch_start)
 
     loader = PostgresLoader(connection_string)