fix(ingest/gc): logging and stopping fix

metadata-ingestion/src/datahub/ingestion/source/gc/execution_request_cleanup.py

@@ -158,8 +158,11 @@
                     break
                 params["scrollId"] = document["scrollId"]
             except Exception as e:
-                logger.error(
-                    f"ergc({self.instance_id}): failed to fetch next batch of execution requests: {e}"
+                self.report.failure(
+                    title="failed to fetch next batch of execution requests",
+                    message="failed to fetch next batch of execution requests",
+                    context=str(self.instance_id),
+                    exc=e,
                 )
                 self.report.ergc_read_errors += 1
 
@@ -231,8 +234,11 @@
             self.graph.delete_entity(entry.urn, True)
         except Exception as e:
             self.report.ergc_delete_errors += 1
-            logger.error(
-                f"ergc({self.instance_id}): failed to delete ExecutionRequest {entry.request_id}: {e}"
+            self.report.failure(
+                title="failed to delete ExecutionRequest",
+                message="failed to delete ExecutionRequest",
+                context=str(self.instance_id),
+                exc=e,
             )
 
     def _reached_runtime_limit(self) -> bool:

metadata-ingestion/src/datahub/ingestion/source/gc/soft_deleted_entity_cleanup.py

@@ -163,6 +163,8 @@
                 f"Dry run is on otherwise it would have deleted {urn} with hard deletion"
             )
             return
+        if self._deletion_limit_reached() or self._times_up():
+            return
         self._increment_removal_started_count()
         self.ctx.graph.delete_entity(urn=urn, hard=True)
         self.ctx.graph.delete_references_to_urn(
@@ -203,11 +205,10 @@
         for future in done:
             self._print_report()
             if future.exception():
-                logger.error(
-                    f"Failed to delete entity {futures[future]}: {future.exception()}"
-                )
                 self.report.failure(
-                    f"Failed to delete entity {futures[future]}",
+                    title="Failed to delete entity",
+                    message=futures[future],
+                    context=str(futures[future]),
                     exc=future.exception(),
                 )
             self.report.num_soft_deleted_entity_processed += 1
@@ -274,6 +275,28 @@
         )
         yield from self._get_soft_deleted_queries()
 
+    def _times_up(self) -> bool:
+        if (
+            self.config.runtime_limit_seconds
+            and time.time() - self.start_time > self.config.runtime_limit_seconds
+        ):
+            logger.info(
+                f"Runtime limit of {self.config.runtime_limit_seconds} seconds reached"
+            )
+            return True
+        return False
+
+    def _deletion_limit_reached(self) -> bool:
+        if (
+            self.config.limit_entities_delete
+            and self.report.num_hard_deleted > self.config.limit_entities_delete
+        ):
+            logger.info(
+                f"Limit of {self.config.limit_entities_delete} entities reached"
+            )
+            return True
+        return False
+
     def cleanup_soft_deleted_entities(self) -> None:
         if not self.config.enabled:
             return
@@ -285,24 +308,8 @@
                 self._print_report()
                 while len(futures) >= self.config.futures_max_at_time:
                     futures = self._process_futures(futures)
-                if (
-                    self.config.limit_entities_delete
-                    and self.report.num_hard_deleted > self.config.limit_entities_delete
-                ):
-                    logger.info(
-                        f"Limit of {self.config.limit_entities_delete} entities reached. Stopped adding more."
-                    )
+                if self._deletion_limit_reached() or self._times_up():
                     break
-                if (
-                    self.config.runtime_limit_seconds
-                    and time.time() - self.start_time
-                    > self.config.runtime_limit_seconds
-                ):
-                    logger.info(
-                        f"Runtime limit of {self.config.runtime_limit_seconds} seconds reached. Not submitting more futures."
-                    )
-                    break
-
                 future = executor.submit(self.delete_soft_deleted_entity, urn)
                 futures[future] = urn