Deployment Custom Pull Steps

Show and Tell
, , ,
1

Introduction

You are running Prefect and deploying your code using prefect deploy.
Below are some alternative examples for introducing custom steps into the prefect.yaml using run_shell. What is important to note, is that these are arbitrary - you can take any sort of custom action after the infrastructure / process has been created.

In the following examples, I want to demonstrate using a 3rd party key management tool, and private repository, in lieu of storing credentials as Prefect Secrets.

Example 2 of this article assume you are familiar with Identity concepts for your respective cloud platform and does not provide instruction on configuring - only how it might be utilized in a pull step for your deployment.
This would encompass Azure managed identities, IRSA for AWS, and Service Account Identities for GCP.

Example 1 - Run Shell Script with AZ CLI

This first example below requires azure-cli installed, and authenticated.
It retrieves a private bitbucket access token from An Azure keyvault secretkeyvault, then uses this to clone from a private repository.

 $ cat prefect.yaml
# Welcome to your prefect.yaml file! You can you this file for storing and managing
# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: test
prefect-version: 2.10.15

# build section allows you to manage and build docker images
build: null

# push section allows you to manage if and how this project is uploaded to remote locations
push: null

# pull section allows you to provide instructions for cloning this project in remote locations
pull:
- prefect.deployments.steps.run_shell_script:
    id: get-access-token
    script: az keyvault secret show --name mysecret --vault-name secretkeyvault --query "value" --output tsv
    stream_output: false # We don't want to expose the credential
- prefect.deployments.steps.git_clone:
    repository: https://bitbucket.org/private/azure-private-deployments.git
    branch: master
    access_token: "{{ get-access-token.stdout }}"

Example 2 - Run Shell Python Script

Another similar example for the same purpose, but to demonstrate running a python script. When run as a process locally, this follows the Azure credential chain, and obtains a DefaultAzureCredential from my shell which has been authenticated previously.
If this is being run as a pull step while using ACI (Azure Container Infrastructure) or AKS (Azure Kubernetes), then managed identities can be assigned and used.

The script “retrieve_secrets.py” must be accessible where the pull step is running from.
If this is local in development, it can be anywhere in the file path.
If this is using a container based deployment such as ACI or AKS, then it would need to be copied in during the image build process, OR, pulled in through another pull step such as another git clone, or S3 download.

 $ cat prefect.yaml
# Welcome to your prefect.yaml file! You can you this file for storing and managing
# configuration for deploying your flows. We recommend committing this file to source
# control along with your flow code.

# Generic metadata about this project
name: test
prefect-version: 2.10.15

# build section allows you to manage and build docker images
build: null

# push section allows you to manage if and how this project is uploaded to remote locations
push: null

# pull section allows you to provide instructions for cloning this project in remote locations
pull:
- prefect.deployments.steps.run_shell_script:
    id: get-access-token
    script: python /tmp/retrieve_secrets.py
    stream_output: false # We don't want to expose the credential
- prefect.deployments.steps.git_clone:
    repository: https://bitbucket.org/private/azure-private-deployments.git
    branch: master
    access_token: "{{ get-access-token.stdout }}"

The contents of retrieve_secrets.py for reference:

from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.keyvault.secrets import SecretClient
import sys

# Set up the default credential, which uses the managed identity of the Azure resource (ACI, VM, etc.)
def get_creds(auth_type):
    if auth_type == "managed_identity":
        credential = ManagedIdentityCredential(client_id="abcd-efgh-1111-222") # The clientID for a user-assigned managed identity
    else:
        credential = DefaultAzureCredential(exclude_shared_token_cache_credential=True)
    return credential


def get_secret(credential):
    # Create a secret client using the default credential and the URL to the Key Vault
    secret_client = SecretClient(vault_url="https://mysecretkeyvault.vault.azure.net", credential=credential)
    secret_name = "mysecret"
    # Retrieve the secret
    retrieved_secret = secret_client.get_secret(secret_name)
    print (retrieved_secret.value)


def main():
    if len(sys.argv) == 2 and sys.argv[1] == "managed_identity":
        credential = get_creds("managed_identity")
    else:
        credential = get_creds("default")
    get_secret(credential)

if __name__ == "__main__":
    main()

Results

Note the results aren’t significantly different here - we can see “Running run_shell_script step”. If you are in the process of testing and developing, you COULD set “stream_output: true” to see the results returned and verify on the console. As we are retrieving a sensitive API token, this has been intentionally set to “false”.

Worker 'ProcessWorker af4c3697-fd23-413d-af4d-0fd6d3c8ba2d' started!
09:02:19.176 | INFO    | prefect.flow_runs.worker - Worker 'ProcessWorker af4c3697-fd23-413d-af4d-0fd6d3c8ba2d' submitting flow run '8360ec6e-bb08-4df6-9f40-4490c0cb813c'
09:02:19.824 | INFO    | prefect.flow_runs.worker - Opening process...
09:02:19.923 | INFO    | prefect.flow_runs.worker - Completed submission of flow run '8360ec6e-bb08-4df6-9f40-4490c0cb813c'
Cloning into 'azure-private-deployments'...
09:02:26.001 | INFO    | prefect.deployment - Cloned repository 'https://bitbucket.org/private/azure-private-deployments.git' into 'azure-private-deployments'
09:02:26.850 | INFO    | Flow run 'wonderful-mouse' - Finished in state Completed()
 > Running run_shell_script step...
 > Running git_clone step...
Hello world
09:02:27.434 | INFO    | prefect.flow_runs.worker - Process 7516 exited cleanly.

Using a Secret Block

While this article is premised on a custom step to retrieve credentials elsewhere, below is an example of using a Prefect Secret if it is stored in your workspace, and permission is authorized for your user. This loads the prefect secret “bitcred” as the necessary credential.

pull:
- prefect.projects.steps.git_clone_project:
    repository: https://github.com/org/repo.git
    branch: master
    access_token: "{{ prefect.blocks.secret.bitcred }}"
1 Like