metadata-service [orchestrator]: improve stale metadata detection (#42962)

Author: alafanechere

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/__init__.py

@@ -189,7 +189,7 @@
     ),
     ScheduleDefinition(job=generate_connector_test_summary_reports, cron_schedule="@hourly"),
     ScheduleDefinition(
-        cron_schedule="0 8 * * *",  # Daily at 8am US/Pacific
+        cron_schedule="0 * * * *",  # Every hour
         execution_timezone="US/Pacific",
         job=generate_stale_gcs_latest_metadata_file,
     ),

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/assets/github.py

@@ -7,16 +7,17 @@
 import hashlib
 import os
 
-import dateutil
-import humanize
 import pandas as pd
+import yaml
 from dagster import OpExecutionContext, Output, asset
 from github import Repository
 from orchestrator.logging import sentry
+from orchestrator.models.metadata import LatestMetadataEntry, MetadataDefinition, PartialMetadataDefinition
 from orchestrator.ops.slack import send_slack_message
 from orchestrator.utils.dagster_helpers import OutputDataFrame, output_dataframe
 
 GROUP_NAME = "github"
+TOOLING_TEAM_SLACK_TEAM_ID = "S077R8636CV"
 
 
 def _get_md5_of_github_file(context: OpExecutionContext, github_connector_repo: Repository, path: str) -> str:
@@ -34,6 +35,11 @@ def _get_md5_of_github_file(context: OpExecutionContext, github_connector_repo:
     return base_64_value
 
 
+def _get_content_of_github_file(context: OpExecutionContext, github_connector_repo: Repository, path: str) -> str:
+    context.log.debug(f"retrieving contents of {path}")
+    return github_connector_repo.get_contents(path)
+
+
 @asset(required_resource_keys={"github_connectors_directory"}, group_name=GROUP_NAME)
 @sentry.instrument_asset_op
 def github_connector_folders(context):
@@ -65,75 +71,76 @@ def github_metadata_file_md5s(context):
     return Output(metadata_file_paths, metadata={"preview": metadata_file_paths})
 
 
-def _should_publish_have_ran(datetime_string: str) -> bool:
-    """
-    Return true if the datetime is 2 hours old.
-
-    """
-    dt = dateutil.parser.parse(datetime_string)
-    now = datetime.datetime.now(datetime.timezone.utc)
-    two_hours_ago = now - datetime.timedelta(hours=2)
-    return dt < two_hours_ago
-
-
-def _to_time_ago(datetime_string: str) -> str:
+@asset(required_resource_keys={"github_connector_repo", "github_connectors_metadata_files"}, group_name=GROUP_NAME)
+def github_metadata_definitions(context):
     """
-    Return a string of how long ago the datetime is human readable format. 10 min
+    Return a list of all metadata definitions hosted on our github repo
     """
-    dt = dateutil.parser.parse(datetime_string)
-    return humanize.naturaltime(dt)
+    github_connector_repo = context.resources.github_connector_repo
+    github_connectors_metadata_files = context.resources.github_connectors_metadata_files
 
+    metadata_definitions = []
+    for metadata_file in github_connectors_metadata_files:
+        metadata_raw = _get_content_of_github_file(context, github_connector_repo, metadata_file["path"])
+        metadata_dict = yaml.safe_load(metadata_raw.decoded_content)
+        metadata_definitions.append(
+            LatestMetadataEntry(
+                metadata_definition=MetadataDefinition.parse_obj(metadata_dict), last_modified=metadata_file["last_modified"]
+            )
+        )
 
-def _is_stale(github_file_info: dict, latest_gcs_metadata_md5s: dict) -> bool:
-    """
-    Return true if the github info is stale.
-    """
-    not_in_gcs = latest_gcs_metadata_md5s.get(github_file_info["md5"]) is None
-    return not_in_gcs and _should_publish_have_ran(github_file_info["last_modified"])
+    return Output(metadata_definitions, metadata={"preview": [md.json() for md in metadata_definitions]})
 
 
-@asset(required_resource_keys={"slack", "latest_metadata_file_blobs"}, group_name=GROUP_NAME)
-def stale_gcs_latest_metadata_file(context, github_metadata_file_md5s: dict) -> OutputDataFrame:
+@asset(required_resource_keys={"slack"}, group_name=GROUP_NAME)
+def stale_gcs_latest_metadata_file(context, github_metadata_definitions: list, metadata_definitions: list) -> OutputDataFrame:
     """
     Return a list of all metadata files in the github repo and denote whether they are stale or not.
 
     Stale means that the file in the github repo is not in the latest metadata file blobs.
     """
-    human_readable_stale_bools = {True: "🚨 YES!!!", False: "No"}
-    latest_gcs_metadata_file_blobs = context.resources.latest_metadata_file_blobs
-    latest_gcs_metadata_md5s = {blob.md5_hash: blob.name for blob in latest_gcs_metadata_file_blobs}
-
-    stale_report = [
-        {
-            "stale": _is_stale(github_file_info, latest_gcs_metadata_md5s),
-            "github_path": github_path,
-            "github_md5": github_file_info["md5"],
-            "github_last_modified": _to_time_ago(github_file_info["last_modified"]),
-            "gcs_md5": latest_gcs_metadata_md5s.get(github_file_info["md5"]),
-            "gcs_path": latest_gcs_metadata_md5s.get(github_file_info["md5"]),
-        }
-        for github_path, github_file_info in github_metadata_file_md5s.items()
-    ]
+    latest_versions_on_gcs = {
+        metadata_entry.metadata_definition.data.dockerRepository: metadata_entry.metadata_definition.data.dockerImageTag
+        for metadata_entry in metadata_definitions
+        if metadata_entry.metadata_definition.data.supportLevel != "archived"
+    }
+
+    now = datetime.datetime.now(datetime.timezone.utc)
+    latest_versions_on_github = {
+        metadata_entry.metadata_definition.data.dockerRepository: metadata_entry.metadata_definition.data.dockerImageTag
+        for metadata_entry in github_metadata_definitions
+        if metadata_entry.metadata_definition.data.supportLevel
+        != "archived"  # We give a 2 hour grace period for the metadata to be updated
+        and datetime.datetime.strptime(metadata_entry.last_modified, "%a, %d %b %Y %H:%M:%S %Z").replace(tzinfo=datetime.timezone.utc)
+        > now - datetime.timedelta(hours=2)
+    }
 
-    stale_metadata_files_df = pd.DataFrame(stale_report)
+    stale_connectors = []
+    for docker_repository, github_docker_image_tag in latest_versions_on_github.items():
+        gcs_docker_image_tag = latest_versions_on_gcs.get(docker_repository)
+        if gcs_docker_image_tag != github_docker_image_tag:
+            stale_connectors.append(
+                {"connector": docker_repository, "master_version": github_docker_image_tag, "gcs_version": gcs_docker_image_tag}
+            )
 
-    # sort by stale true to false, then by github_path
-    stale_metadata_files_df = stale_metadata_files_df.sort_values(
-        by=["stale", "github_path"],
-        ascending=[False, True],
-    )
+    stale_connectors_df = pd.DataFrame(stale_connectors)
 
     # If any stale files exist, report to slack
     channel = os.getenv("STALE_REPORT_CHANNEL")
-    any_stale = stale_metadata_files_df["stale"].any()
-    if channel and any_stale:
-        only_stale_df = stale_metadata_files_df[stale_metadata_files_df["stale"] == True]
-        pretty_stale_df = only_stale_df.replace(human_readable_stale_bools)
-        stale_report_md = pretty_stale_df.to_markdown(index=False)
-        send_slack_message(context, channel, stale_report_md, enable_code_block_wrapping=True)
-
-    stale_metadata_files_df.replace(human_readable_stale_bools, inplace=True)
-    return output_dataframe(stale_metadata_files_df)
+    any_stale = len(stale_connectors_df) > 0
+    if channel:
+        if any_stale:
+            stale_report_md = stale_connectors_df.to_markdown(index=False)
+            send_slack_message(context, channel, f"🚨 Stale metadata detected! (cc. <!subteam^{TOOLING_TEAM_SLACK_TEAM_ID}>)")
+            send_slack_message(context, channel, stale_report_md, enable_code_block_wrapping=True)
+        else:
+            message = f"""
+            Analyzed {len(github_metadata_definitions)} metadata files on our master branch and {len(metadata_definitions)} latest metadata files hosted in GCS.All MD5 hashes of these files.
+            All MD5 hashes of our metadata files on master match the latest metadata files on GCS.
+            No stale metadata: GCS metadata are up to date with metadata hosted on GCS.
+            """
+            send_slack_message(context, channel, message)
+    return output_dataframe(stale_connectors_df)
 
 
 @asset(required_resource_keys={"github_connector_nightly_workflow_successes"}, group_name=GROUP_NAME)

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/assets/metadata.py

@@ -130,7 +130,7 @@ def metadata_definitions(context: OpExecutionContext) -> List[LatestMetadataEntr
         metadata_entry = LatestMetadataEntry(
             metadata_definition=metadata_def,
             icon_url=icon_url,
-            last_modified=blob.last_modified,
+            last_modified=blob.updated.isoformat(),
             etag=blob.etag,
             file_path=blob.name,
             bucket_name=blob.bucket.name,

airbyte-ci/connectors/metadata_service/orchestrator/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "orchestrator"
-version = "0.1.4"
+version = "0.1.5"
 description = ""
 authors = ["Ben Church <[email protected]>"]
 readme = "README.md"