Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(ingest/gc): add limit, add actual loop for iterating over batches #11809

Merged
merged 4 commits into from
Nov 6, 2024
Merged

fix(ingest/gc): add limit, add actual loop for iterating over batches #11809

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
69e7893
fix(ingest/gc): add limit, add actual loop for iterating over batches
anshbansal Nov 6, 2024
a4541fd
break from loop
anshbansal Nov 6, 2024
96f2c33
Fix iterator
treff7es Nov 6, 2024
739bf01
Add fix if seconds timout is not set
treff7es Nov 6, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions metadata-ingestion/src/datahub/ingestion/source/gc/soft_deleted_entity_cleanup.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,14 @@ class SoftDeletedEntitiesCleanupConfig(ConfigModel):
default=None,
description="Query to filter entities",
)
limit_entities_delete: Optional[int] = Field(
10000, description="Max number of entities to delete."
)

runtime_limit_seconds: Optional[int] = Field(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Think this makes more sense as minutes

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

probably true in reality but dev-ops is pretty used to dealing with seconds for lots of things, feel free to change it later but not a blocker imho

None,
description="Runtime limit in seconds",
)


@dataclass
Expand Down Expand Up @@ -122,6 +130,10 @@ def delete_entity(self, urn: str) -> None:
return

self.ctx.graph.delete_entity(urn=urn, hard=True)
self.ctx.graph.delete_references_to_urn(
urn=urn,
dry_run=False,
)

def delete_soft_deleted_entity(self, urn: str) -> None:
assert self.ctx.graph
Expand All @@ -145,6 +157,7 @@ def delete_soft_deleted_entity(self, urn: str) -> None:

def cleanup_soft_deleted_entities(self) -> None:
assert self.ctx.graph
start_time = time.time()

deleted_count_retention = 0
urns = self.ctx.graph.get_urns_by_filter(
Expand All @@ -158,7 +171,26 @@ def cleanup_soft_deleted_entities(self) -> None:

futures = {}
with ThreadPoolExecutor(max_workers=self.config.max_workers) as executor:
num_urns_submitted = 0
for urn in urns:
num_urns_submitted += 1
if (
self.config.limit_entities_delete
and num_urns_submitted > self.config.limit_entities_delete
):
logger.info(
f"Limit of {self.config.limit_entities_delete} entities reached. Stopping"
)
break
if (
self.config.runtime_limit_seconds
and time.time() - start_time > self.config.runtime_limit_seconds
):
logger.info(
f"Runtime limit of {self.config.runtime_limit_seconds} seconds reached. Stopping"
)
break

future = executor.submit(self.delete_soft_deleted_entity, urn)
futures[future] = urn

Expand Down
Loading