fix(ingest): add logging for mcp diff (#11683)

datahub-project · Oct 23, 2024 · 581a39a · 581a39a
1 parent 35f30b7
commit 581a39a
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 4 deletions.
diff --git a/metadata-ingestion/src/datahub/ingestion/api/source_helpers.py b/metadata-ingestion/src/datahub/ingestion/api/source_helpers.py
@@ -148,6 +148,12 @@ def auto_workunit_reporter(report: "SourceReport", stream: Iterable[T]) -> Itera
         report.report_workunit(wu)
         yield wu
 
+    if report.events_produced == 0:
+        report.warning(
+            title="No metadata was produced by the source",
+            message="Please check the source configuration, filters, and permissions.",
+        )
+
 
 def auto_materialize_referenced_tags_terms(
     stream: Iterable[MetadataWorkUnit],

diff --git a/metadata-ingestion/src/datahub/ingestion/source/fivetran/config.py b/metadata-ingestion/src/datahub/ingestion/source/fivetran/config.py
@@ -160,11 +160,14 @@ class FivetranSourceConfig(StatefulIngestionConfigBase, DatasetSourceConfigMixin
     )
     connector_patterns: AllowDenyPattern = Field(
         default=AllowDenyPattern.allow_all(),
-        description="Regex patterns for connectors to filter in ingestion.",
+        description="Filtering regex patterns for connector ids. "
+        "They're visible in the Fivetran UI under Connectors -> Setup -> Fivetran Connector ID.",
     )
     destination_patterns: AllowDenyPattern = Field(
         default=AllowDenyPattern.allow_all(),
-        description="Regex patterns for destinations to filter in ingestion.",
+        description="Regex patterns for destination ids to filter in ingestion. "
+        "Fivetran destination IDs are usually two word identifiers e.g. canyon_tolerable, and are not the same as the destination database name. "
+        "They're visible in the Fivetran UI under Destinations -> Overview -> Destination Group ID.",
     )
     include_column_lineage: bool = Field(
         default=True,

diff --git a/metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran_log_api.py b/metadata-ingestion/src/datahub/ingestion/source/fivetran/fivetran_log_api.py
@@ -277,6 +277,7 @@ def get_allowed_connectors_list(
         if not connectors:
             # Some of our queries don't work well when there's no connectors, since
             # we push down connector id filters.
+            logger.info("No allowed connectors found")
             return []
 
         with report.metadata_extraction_perf.connectors_lineage_extraction_sec:

diff --git a/metadata-ingestion/src/datahub/ingestion/transformer/system_metadata_transformer.py b/metadata-ingestion/src/datahub/ingestion/transformer/system_metadata_transformer.py
@@ -31,14 +31,14 @@ def auto_system_metadata(
 
 class SystemMetadataTransformer(Transformer):
     def __init__(self, ctx: PipelineContext):
-        self._inner_transfomer = AutoHelperTransformer(
+        self._inner_transformer = AutoHelperTransformer(
             functools.partial(auto_system_metadata, ctx)
         )
 
     def transform(
         self, record_envelopes: Iterable[RecordEnvelope]
     ) -> Iterable[RecordEnvelope]:
-        yield from self._inner_transfomer.transform(record_envelopes)
+        yield from self._inner_transformer.transform(record_envelopes)
 
     @classmethod
     def create(cls, config_dict: dict, ctx: PipelineContext) -> Transformer:

diff --git a/metadata-ingestion/src/datahub/testing/compare_metadata_json.py b/metadata-ingestion/src/datahub/testing/compare_metadata_json.py
@@ -89,10 +89,12 @@ def assert_metadata_files_equal(
     diff = diff_metadata_json(output, golden, ignore_paths, ignore_order=ignore_order)
     if diff and update_golden:
         if isinstance(diff, MCPDiff) and diff.is_delta_valid:
+            logger.info(f"Applying delta to golden file {golden_path}")
             diff.apply_delta(golden)
             write_metadata_file(pathlib.Path(golden_path), golden)
         else:
             # Fallback: just overwrite the golden file
+            logger.info(f"Overwriting golden file {golden_path}")
             shutil.copyfile(str(output_path), str(golden_path))
         return