feat(registry): Add dev deployment (#38221)

## What
1. Fixes an issue with failing deployments
2. Adds the ability to publish to a dev dagster environment
3. Fixes an issue where successful pipeline runs were marked as a failure

## How
1. Removes poetry as a dependency, it was unnesessary and causing an error with pex
2. Updates GHA and the pipeline to accept `DAGSTER_CLOUD_DEPLOYMENT`
3. All we needed to do was return the result 🤦‍♂️

Author: bnchrch

.github/workflows/metadata_service_deploy_orchestrator_dagger.yml

@@ -2,6 +2,10 @@ name: Connector Ops CI - Metadata Service Deploy Orchestrator
 
 on:
   workflow_dispatch:
+    inputs:
+      deployment_target:
+        description: "The deployment target for the metadata orchestrator (prod or dev)"
+        default: "dev"
   push:
     branches:
       - master
@@ -14,8 +18,9 @@ jobs:
     steps:
       - name: Checkout Airbyte
         uses: actions/checkout@v2
-      - name: Deploy the metadata orchestrator
-        id: metadata-orchestrator-deploy-orchestrator-pipeline
+      - name: Deploy the metadata orchestrator [On merge to master]
+        id: metadata-orchestrator-deploy-orchestrator-pipeline-prod
+        if: github.event_name == 'push'
         uses: ./.github/actions/run-airbyte-ci
         with:
           subcommand: "metadata deploy orchestrator"
@@ -27,3 +32,19 @@ jobs:
           gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}
         env:
           DAGSTER_CLOUD_METADATA_API_TOKEN: ${{ secrets.DAGSTER_CLOUD_METADATA_API_TOKEN }}
+          DAGSTER_CLOUD_DEPLOYMENT: "prod"
+      - name: Deploy the metadata orchestrator [On workflow]
+        id: metadata-orchestrator-deploy-orchestrator-pipeline-branch
+        if: github.event_name == 'workflow_dispatch'
+        uses: ./.github/actions/run-airbyte-ci
+        with:
+          subcommand: "metadata deploy orchestrator"
+          context: "manual"
+          dagger_cloud_token: ${{ secrets.DAGGER_CLOUD_TOKEN_2 }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          docker_hub_username: ${{ secrets.DOCKER_HUB_USERNAME }}
+          docker_hub_password: ${{ secrets.DOCKER_HUB_PASSWORD }}
+          gcp_gsm_credentials: ${{ secrets.GCP_GSM_CREDENTIALS }}
+        env:
+          DAGSTER_CLOUD_METADATA_API_TOKEN: ${{ secrets.DAGSTER_CLOUD_METADATA_API_TOKEN }}
+          DAGSTER_CLOUD_DEPLOYMENT: ${{ inputs.deployment_target }}

airbyte-ci/connectors/metadata_service/orchestrator/README.md

@@ -171,6 +171,14 @@ open http://localhost:3000
 And run the `generate_registry` job
 
 ## Additional Notes
+### How to publish to Dagster Cloud Dev
+Pre-requisites:
+- You need to have `airbyte-ci` installed. You can install it by running `make tools.airbyte-ci-dev.install` from the root of the repository.
+- You need to have the `DAGSTER_CLOUD_METADATA_API_TOKEN` environment variable set to an API token from the Airbyte Dagster Cloud account.
+
+```sh
+DAGSTER_CLOUD_METADATA_API_TOKEN=<SECRET> DAGSTER_CLOUD_DEPLOYMENT="dev" airbyte-ci metadata deploy orchestrator
+```
 
 ### Testing Slack Notifications
 You will need to add the following environment variables to your `.env` file:

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

@@ -33,8 +33,10 @@ def remove_stale_metadata_partitions_op(context):
     partition_name = registry_entry.metadata_partitions_def.name
 
     all_fresh_etags = [blob.etag for blob in all_metadata_file_blobs]
+    context.log.info(f"Found {len(all_fresh_etags)} fresh metadata files found in GCS bucket")
 
     all_etag_partitions = context.instance.get_dynamic_partitions(partition_name)
+    context.log.info(f"Found {len(all_etag_partitions)} existing metadata partitions")
 
     for stale_etag in [etag for etag in all_etag_partitions if etag not in all_fresh_etags]:
         context.log.info(f"Removing stale etag: {stale_etag}")
@@ -55,12 +57,16 @@ def add_new_metadata_partitions_op(context):
     """
     This op is responsible for polling for new metadata files and adding their etag to the dynamic partition.
     """
+    context.log.info("Starting add_new_metadata_partitions_op")
+
     all_metadata_file_blobs = context.resources.all_metadata_file_blobs
+    context.log.info(f"Found {len(all_metadata_file_blobs)} metadata files found in GCS bucket")
+
     partition_name = registry_entry.metadata_partitions_def.name
+    existing_partitions = context.instance.get_dynamic_partitions(partition_name)
+    context.log.info(f"Found {len(existing_partitions)} existing metadata partitions")
 
-    new_files_found = {
-        blob.etag: blob.name for blob in all_metadata_file_blobs if not context.instance.has_dynamic_partition(partition_name, blob.etag)
-    }
+    new_files_found = {blob.etag: blob.name for blob in all_metadata_file_blobs if not blob.etag in existing_partitions}
 
     new_etags_found = list(new_files_found.keys())
     context.log.info(f"New etags found: {new_etags_found}")