yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-12-14 11:36:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingest/looker): limit dashboard object lifetime to thread (#9564)

Browse Source
This commit is contained in:
Harshal Sheth 2024-01-04 16:34:08 -05:00 committed by GitHub
parent 8f45d3d0cb
commit 20b7dd3642
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 11 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
metadata-ingestion/src/datahub/ingestion/source/looker/looker_source.py
View File

@ -1044,7 +1044,7 @@ class LookerDashboardSource(TestableSource, StatefulIngestionSourceBase):
self, dashboard_id: str, fields: List[str] self, dashboard_id: str, fields: List[str]
) -> Tuple[ ) -> Tuple[
List[MetadataWorkUnit], List[MetadataWorkUnit],
Optional[Dashboard], Optional[looker_usage.LookerDashboardForUsage],
str, str,
datetime.datetime, datetime.datetime,
datetime.datetime, datetime.datetime,
@ -1095,9 +1095,15 @@ class LookerDashboardSource(TestableSource, StatefulIngestionSourceBase):
) )
workunits.extend(metric_dim_workunits) workunits.extend(metric_dim_workunits)
self.reporter.report_dashboards_scanned() self.reporter.report_dashboards_scanned()
# generate usage tracking object
dashboard_usage = looker_usage.LookerDashboardForUsage.from_dashboard(
dashboard_object
)
return ( return (
workunits, workunits,
dashboard_object, dashboard_usage,
dashboard_id, dashboard_id,
start_time, start_time,
datetime.datetime.now(), datetime.datetime.now(),
@ -1287,7 +1293,7 @@ class LookerDashboardSource(TestableSource, StatefulIngestionSourceBase):
for job in concurrent.futures.as_completed(async_workunits): for job in concurrent.futures.as_completed(async_workunits):
( (
work_units, work_units,
dashboard_object, dashboard_usage,
dashboard_id, dashboard_id,
start_time, start_time,
end_time, end_time,
@ -1299,12 +1305,8 @@ class LookerDashboardSource(TestableSource, StatefulIngestionSourceBase):
self.reporter.report_upstream_latency(start_time, end_time) self.reporter.report_upstream_latency(start_time, end_time)
yield from work_units yield from work_units
if dashboard_object is not None: if dashboard_usage is not None:
looker_dashboards_for_usage.append( looker_dashboards_for_usage.append(dashboard_usage)
looker_usage.LookerDashboardForUsage.from_dashboard(
dashboard_object
)
)
self.reporter.report_stage_end("dashboard_chart_metadata") self.reporter.report_stage_end("dashboard_chart_metadata")