Update slack to use resource and bot

Author: bnchrch

airbyte-ci/connectors/metadata_service/orchestrator/.env.template

@@ -4,4 +4,6 @@ GITHUB_METADATA_SERVICE_TOKEN=""
 NIGHTLY_REPORT_SLACK_WEBHOOK_URL=""
 # METADATA_CDN_BASE_URL="https://connectors.airbyte.com/files"
 DOCKER_HUB_USERNAME=""
-DOCKER_HUB_PASSWORD=""
+DOCKER_HUB_PASSWORD=""
+SLACK_TOKEN = ""
+PUBLISH_UPDATE_CHANNEL="#ben-test"

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

@@ -1,7 +1,8 @@
 #
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
-from dagster import Definitions, ScheduleDefinition, load_assets_from_modules
+from dagster import Definitions, ScheduleDefinition, EnvVar, load_assets_from_modules
+from dagster_slack import SlackResource
 
 from orchestrator.resources.gcp import gcp_gcs_client, gcs_directory_blobs, gcs_file_blob, gcs_file_manager
 from orchestrator.resources.github import github_client, github_connector_repo, github_connectors_directory, github_workflow_runs
@@ -54,6 +55,10 @@
     ]
 )
 
+SLACK_RESOURCE_TREE = {
+    "slack": SlackResource(token=EnvVar("SLACK_TOKEN")),
+}
+
 GITHUB_RESOURCE_TREE = {
     "github_client": github_client.configured({"github_token": {"env": "GITHUB_METADATA_SERVICE_TOKEN"}}),
     "github_connector_repo": github_connector_repo.configured({"connector_repo_name": CONNECTOR_REPO_NAME}),
@@ -79,6 +84,7 @@
 }
 
 METADATA_RESOURCE_TREE = {
+    **SLACK_RESOURCE_TREE,
     **GCS_RESOURCE_TREE,
     "all_metadata_file_blobs": gcs_directory_blobs.configured(
         {"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": METADATA_FOLDER, "match_regex": f".*/{METADATA_FILE_NAME}$"}
@@ -89,6 +95,7 @@
 }
 
 REGISTRY_RESOURCE_TREE = {
+    **SLACK_RESOURCE_TREE,
     **GCS_RESOURCE_TREE,
     "latest_oss_registry_gcs_blob": gcs_file_blob.configured(
         {"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRIES_FOLDER, "gcs_filename": "oss_registry.json"}
@@ -99,6 +106,7 @@
 }
 
 REGISTRY_ENTRY_RESOURCE_TREE = {
+    **SLACK_RESOURCE_TREE,
     **GCS_RESOURCE_TREE,
     "latest_cloud_registry_entries_file_blobs": gcs_directory_blobs.configured(
         {"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": METADATA_FOLDER, "match_regex": f".*latest/cloud.json$"}

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

@@ -91,7 +91,7 @@ def generate_and_persist_registry(
 # Registry Generation
 
 
-@asset(required_resource_keys={"registry_directory_manager", "latest_oss_registry_entries_file_blobs"}, group_name=GROUP_NAME)
+@asset(required_resource_keys={"slack", "registry_directory_manager", "latest_oss_registry_entries_file_blobs"}, group_name=GROUP_NAME)
 def persisted_oss_registry(context: OpExecutionContext) -> Output[ConnectorRegistryV0]:
     """
     This asset is used to generate the oss registry from the registry entries.
@@ -108,7 +108,7 @@ def persisted_oss_registry(context: OpExecutionContext) -> Output[ConnectorRegis
     )
 
 
-@asset(required_resource_keys={"registry_directory_manager", "latest_cloud_registry_entries_file_blobs"}, group_name=GROUP_NAME)
+@asset(required_resource_keys={"slack", "registry_directory_manager", "latest_cloud_registry_entries_file_blobs"}, group_name=GROUP_NAME)
 def persisted_cloud_registry(context: OpExecutionContext) -> Output[ConnectorRegistryV0]:
     """
     This asset is used to generate the cloud registry from the registry entries.

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

@@ -305,7 +305,7 @@ def safe_parse_metadata_definition(metadata_blob: storage.Blob) -> Optional[Meta
 
 
 @asset(
-    required_resource_keys={"all_metadata_file_blobs"},
+    required_resource_keys={"slack", "all_metadata_file_blobs"},
     group_name=GROUP_NAME,
     partitions_def=metadata_partitions_def,
     output_required=False,
@@ -377,7 +377,7 @@ def metadata_entry(context: OpExecutionContext) -> Output[Optional[LatestMetadat
 
 
 @asset(
-    required_resource_keys={"root_metadata_directory_manager"},
+    required_resource_keys={"slack", "root_metadata_directory_manager"},
     group_name=GROUP_NAME,
     partitions_def=metadata_partitions_def,
     auto_materialize_policy=AutoMaterializePolicy.eager(max_materializations_per_minute=MAX_METADATA_PARTITION_RUN_REQUEST),

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/logging/publish_connector_lifecycle.py

@@ -2,7 +2,7 @@
 
 from enum import Enum
 from dagster import OpExecutionContext
-from orchestrator.ops.slack import send_slack_webhook
+from orchestrator.ops.slack import send_slack_message
 
 
 class StageStatus(str, Enum):
@@ -36,6 +36,14 @@ def __str__(self) -> str:
 
 
 class PublishConnectorLifecycle:
+    """
+    This class is used to log the lifecycle of a publishing a connector to the registries.
+
+    It is used to log to the logger and slack (if enabled).
+
+    This is nessesary as this lifecycle is not a single job, asset, resource, schedule, or sensor.
+    """
+
     @staticmethod
     def stage_to_log_level(stage_status: StageStatus) -> str:
         if stage_status == StageStatus.FAILED:
@@ -60,8 +68,7 @@ def log(context: OpExecutionContext, lifecycle_stage: PublishConnectorLifecycleS
         level = PublishConnectorLifecycle.stage_to_log_level(stage_status)
         log_method = getattr(context.log, level)
         log_method(message)
-
-        slack_webhook_url = os.getenv("PUBLISH_UPDATE_SLACK_WEBHOOK_URL")
-        if slack_webhook_url:
+        channel = os.getenv("PUBLISH_UPDATE_CHANNEL")
+        if channel:
             slack_message = f"🤖 {message}"
-            send_slack_webhook(slack_webhook_url, slack_message)
+            send_slack_message(context, channel, slack_message)

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/ops/slack.py

@@ -1,4 +1,8 @@
+import os
+
+from dagster import op, OpExecutionContext
 from slack_sdk import WebhookClient
+from dagster_slack import SlackResource
 
 
 def chunk_messages(report):
@@ -12,10 +16,22 @@ def chunk_messages(report):
     yield msg
 
 
-def send_slack_webhook(webhook_url, report, wrap_in_code_block=False):
+@op
+def send_slack_webhook(webhook_url, report):
     webhook = WebhookClient(webhook_url)
     for msg in chunk_messages(report):
         # Wrap in code block as slack does not support markdown in webhooks
-        if wrap_in_code_block:
-            msg = f"```{msg}```"
-        webhook.send(msg)
+        webhook.send(f"```{msg}```")
+
+
+def send_slack_message(context: OpExecutionContext, channel: str, message: str):
+    """
+    Send a slack message to the given channel.
+
+    Args:
+        context (OpExecutionContext): The execution context.
+        channel (str): The channel to send the message to.
+        message (str): The message to send.
+    """
+    if os.getenv("SLACK_TOKEN"):
+        context.resources.slack.get_client().chat_postMessage(channel=channel, text=message)

airbyte-ci/connectors/metadata_service/orchestrator/poetry.lock

@@ -26,6 +26,7 @@ poetry2setup = "^1.1.0"
 slack-sdk = "^3.21.3"
 poetry = "^1.5.1"
 pydantic = "^1.10.6"
+dagster-slack = "^0.20.2"
 
 
 [tool.poetry.group.dev.dependencies]