Acionar DAGs do Serviço gerenciado para Apache Airflow com funções do Cloud Run e a API REST do Airflow

Airflow gerenciado (Geração 3) | Airflow gerenciado (Geração 2) | Airflow gerenciado (Geração 1 legada)

Nesta página, descrevemos como usar funções do Cloud Run para acionar DAGs do Serviço gerenciado para Apache Airflow em resposta a eventos.

O Apache Airflow foi projetado para executar DAGs em uma programação regular, mas também é possível acionar DAGs em resposta a eventos. Uma maneira de fazer isso é usar as funções do Cloud Run para acionar DAGs gerenciados do Airflow quando um evento especificado ocorrer.

Você também pode:

O exemplo neste guia demonstra uma função que aciona um DAG em resposta a um evento:

  1. Você configura acionadores para sua função no Cloud Run functions.
  2. Quando a função é acionada, ela faz uma solicitação para acionar um DAG pela API REST do Airflow do seu ambiente do Airflow Gerenciado. A solicitação contém o identificador e o tipo do evento, além da carga útil dele.
  3. O Airflow processa essa solicitação e executa o DAG especificado nela. O DAG gera os dados que foram transmitidos a ele pela função.

Antes de começar

Esta seção lista as etapas preparatórias.

Verificar a configuração de rede do ambiente

Essa solução não funciona em configurações de IP privado e VPC Service Controls porque não é possível configurar a conectividade das funções do Cloud Run com o servidor da Web do Airflow nessas configurações.

Ativar as APIs do projeto

Console

Ative as APIs do Airflow gerenciado e do Cloud Run functions, se alguma delas ainda não estiver ativada.

Funções necessárias para ativar APIs

Para ativar APIs, você precisa da permissão serviceusage.services.enable. Se você criou o projeto, provavelmente já tem essa permissão com o papel de Proprietário (roles/owner). Caso contrário, é possível receber essa permissão com o papel de Administrador do Service Usage (roles/serviceusage.serviceUsageAdmin). Saiba como conceder papéis.

Ativar as APIs

gcloud

Ative as APIs do Airflow gerenciado e do Cloud Run functions, se alguma delas ainda não estiver ativada:

Funções necessárias para ativar APIs

Para ativar APIs, você precisa da permissão serviceusage.services.enable. Se você criou o projeto, provavelmente já tem essa permissão com o papel de Proprietário (roles/owner). Caso contrário, é possível receber essa permissão com o papel de Administrador do Service Usage (roles/serviceusage.serviceUsageAdmin). Saiba como conceder papéis. 

gcloud services enable cloudfunctions.googleapis.com composer.googleapis.com

Ativar a API REST do Airflow

Dependendo da sua versão do Airflow, faça o seguinte:

Permitir chamadas de API para a API REST do Airflow usando o controle de acesso à rede do servidor da Web

O Cloud Run functions pode acessar a API REST do Airflow usando um endereço IPv4 ou IPv6.

Se você não tiver certeza de qual será o intervalo de IP de chamada, use uma opção de configuração padrão em Controle de acesso ao servidor da Web, que é All IP addresses have access (default), para não bloquear acidentalmente as funções do Cloud Run. Você pode configurar o acesso à rede do servidor da Web mais tarde.

Ver o URL do servidor da Web do Airflow

Este exemplo faz solicitações da API REST para o endpoint do servidor da Web do Airflow. Use a parte do URL da interface da Web do Airflow antes de .appspot.com no código da função do Cloud.

Console

  1. No console Google Cloud , acesse a página Ambientes.

    Acessar "Ambientes"

  2. Clique no nome do seu ambiente.

  3. Na página Detalhes do ambiente, acesse a guia Configuração do ambiente.

  4. O URL do servidor da Web do Airflow está listado no item da IU da Web do Airflow.

gcloud

Execute este comando:

gcloud composer environments describe ENVIRONMENT_NAME \
    --location LOCATION \
    --format='value(config.airflowUri)'

Substitua:

  • ENVIRONMENT_NAME pelo nome do ambiente
  • LOCATION pela região em que o ambiente está localizado;

Acessar o client_id do proxy do IAM

Para fazer uma solicitação ao endpoint de API REST do Airflow, a função exige o ID do cliente do proxy do Identity and Access Management que protege o servidor da Web do Airflow.

O Airflow Gerenciado não fornece essas informações diretamente. Em vez disso, faça uma solicitação não autenticada no servidor da Web do Airflow e capture o ID do cliente do URL de redirecionamento:

cURL 

curl -v AIRFLOW_URL 2>&1 >/dev/null | grep -o "client_id\=[A-Za-z0-9-]*\.apps\.googleusercontent\.com"

Substitua AIRFLOW_URL pelo URL da interface da Web do Airflow.

Na saída, procure a string após client_id. Exemplo:

client_id=836436932391-16q2c5f5dcsfnel77va9bvf4j280t35c.apps.googleusercontent.com

Python

Salve o código a seguir em um arquivo chamado get_client_id.py. Preencha os valores de project_id, location e composer_environment. Em seguida, execute o código no Cloud Shell ou no ambiente local. 

# This script is intended to be used with Composer 1 environments
# In Composer 2, the Airflow Webserver is not in the tenant project
# so there is no tenant client ID
# See https://cloud.google.com/composer/docs/composer-2/environment-architecture
# for more details
import google.auth
import google.auth.transport.requests
import requests
import six.moves.urllib.parse

# Authenticate with Google Cloud.
# See: https://cloud.google.com/docs/authentication/getting-started
credentials, _ = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
authed_session = google.auth.transport.requests.AuthorizedSession(credentials)

# project_id = 'YOUR_PROJECT_ID'
# location = 'us-central1'
# composer_environment = 'YOUR_COMPOSER_ENVIRONMENT_NAME'

environment_url = (
    "https://composer.googleapis.com/v1beta1/projects/{}/locations/{}"
    "/environments/{}"
).format(project_id, location, composer_environment)
composer_response = authed_session.request("GET", environment_url)
environment_data = composer_response.json()
composer_version = environment_data["config"]["softwareConfig"]["imageVersion"]
if "composer-1" not in composer_version:
    version_error = (
        "This script is intended to be used with Composer 1 environments. "
        "In Composer 2, the Airflow Webserver is not in the tenant project, "
        "so there is no tenant client ID. "
        "See https://cloud.google.com/composer/docs/composer-2/environment-architecture for more details."
    )
    raise (RuntimeError(version_error))
airflow_uri = environment_data["config"]["airflowUri"]

# The Composer environment response does not include the IAP client ID.
# Make a second, unauthenticated HTTP request to the web server to get the
# redirect URI.
redirect_response = requests.get(airflow_uri, allow_redirects=False)
redirect_location = redirect_response.headers["location"]

# Extract the client_id query parameter from the redirect.
parsed = six.moves.urllib.parse.urlparse(redirect_location)
query_string = six.moves.urllib.parse.parse_qs(parsed.query)
print(query_string["client_id"][0])

Fazer upload de um DAG para o ambiente

Faça o upload de um DAG para seu ambiente. O exemplo a seguir mostra a configuração de execução do DAG recebida. Você acionará esse DAG a partir de uma função que vai criar neste guia depois.

import datetime

import airflow
from airflow.operators.bash_operator import BashOperator

with airflow.DAG(
        'composer_sample_trigger_response_dag',
        start_date=datetime.datetime(2026, 1, 1),
        # Not scheduled, trigger only
        schedule=None) as dag:

    # Print the dag_run's configuration, which includes information about the
    # Cloud Storage object change.
    print_gcs_info = BashOperator(
        task_id='print_gcs_info', bash_command='echo {{ dag_run.conf }}}}')

Implantar uma função que aciona o DAG

É possível implantar uma função usando sua linguagem preferida compatível com o Cloud Run Functions ou o Cloud Run. Neste tutorial, demonstramos uma função do Cloud implementada em Python e Java.

Especificar parâmetros de configuração da função

  • Gatilho: selecione um ou vários gatilhos do Eventarc para sua função.

    Para mais informações sobre como criar gatilhos, consulte Criar gatilhos com o Eventarc. Por exemplo, é possível acionar funções do Cloud Storage usando o Eventarc.

  • Conta de serviço: a conta de serviço especificada para o gatilho precisa ter permissões suficientes para acionar DAGs em ambientes do Airflow Gerenciado.

    Recomendamos seguir o princípio de menor privilégio e conceder apenas o papel de Usuário do Composer (composer.user). Para mais informações sobre como configurar permissões, consulte Papéis e permissões para destinos do Cloud Run.

  • Ponto de entrada da função:

    • (Python) Ao adicionar código para este exemplo, selecione o ambiente de execução Python 3.10 ou mais recente e especifique trigger_dag como o ponto de entrada.

Adicionar requisitos

Especifique as dependências no arquivo requirements.txt:

requests-toolbelt==1.0.0
google-auth==2.38.0
google-cloud-pubsub==2.28.0

Adicionar código da função

Coloque o código a seguir no arquivo main.py e faça as seguintes substituições:

  • Substitua o valor da variável client_id pelo valor client_id recebido anteriormente.

  • Substitua o valor da variável webserver_id pelo ID do projeto de locatário, que faz parte do URL da interface da Web do Airflow antes de .appspot.com. Você já recebeu o URL da interface da Web do Airflow.

  • Especifique a versão da API REST do Airflow usada:

    • Se você usa a API REST estável do Airflow, defina a variável USE_EXPERIMENTAL_API como False.
    • Se você usa a API REST experimental do Airflow, não é necessário fazer alterações. A variável USE_EXPERIMENTAL_API já está definida como True.


from google.auth.transport.requests import Request
from google.oauth2 import id_token
import requests


IAM_SCOPE = "https://www.googleapis.com/auth/iam"
OAUTH_TOKEN_URI = "https://www.googleapis.com/oauth2/v4/token"
# If you are using the stable API, set this value to False
# For more info about Airflow APIs see https://cloud.google.com/composer/docs/access-airflow-api
USE_EXPERIMENTAL_API = True


def trigger_dag(data, context=None):
    """Makes a POST request to the Composer DAG Trigger API

    When called via Google Cloud Functions (GCF),
    data and context are Background function parameters.

    For more info, refer to
    https://cloud.google.com/functions/docs/writing/background#functions_background_parameters-python

    To call this function from a Python script, omit the ``context`` argument
    and pass in a non-null value for the ``data`` argument.

    This function is currently only compatible with Composer v1 environments.
    """

    # Fill in with your Composer info here
    # Navigate to your webserver's login page and get this from the URL
    # Or use the script found at
    # https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/composer/rest/get_client_id.py
    client_id = "YOUR-CLIENT-ID"
    # This should be part of your webserver's URL:
    # {tenant-project-id}.appspot.com
    webserver_id = "YOUR-TENANT-PROJECT"
    # The name of the DAG you wish to trigger
    dag_name = "composer_sample_trigger_response_dag"

    if USE_EXPERIMENTAL_API:
        endpoint = f"api/experimental/dags/{dag_name}/dag_runs"
        json_data = {"conf": data, "replace_microseconds": "false"}
    else:
        endpoint = f"api/v1/dags/{dag_name}/dagRuns"
        json_data = {"conf": data}
    webserver_url = "https://" + webserver_id + ".appspot.com/" + endpoint
    # Make a POST request to IAP which then Triggers the DAG
    make_iap_request(webserver_url, client_id, method="POST", json=json_data)


# This code is copied from
# https://github.com/GoogleCloudPlatform/python-docs-samples/blob/main/iap/make_iap_request.py
# START COPIED IAP CODE
def make_iap_request(url, client_id, method="GET", **kwargs):
    """Makes a request to an application protected by Identity-Aware Proxy.
    Args:
      url: The Identity-Aware Proxy-protected URL to fetch.
      client_id: The client ID used by Identity-Aware Proxy.
      method: The request method to use
              ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                https://github.com/requests/requests/blob/master/requests/api.py
                If no timeout is provided, it is set to 90 by default.
    Returns:
      The page body, or raises an exception if the page couldn't be retrieved.
    """
    # Set the default timeout, if missing
    if "timeout" not in kwargs:
        kwargs["timeout"] = 90

    # Obtain an OpenID Connect (OIDC) token from metadata server or using service
    # account.
    google_open_id_connect_token = id_token.fetch_id_token(Request(), client_id)

    # Fetch the Identity-Aware Proxy-protected URL, including an
    # Authorization header containing "Bearer " followed by a
    # Google-issued OpenID Connect token for the service account.
    resp = requests.request(
        method,
        url,
        headers={"Authorization": "Bearer {}".format(google_open_id_connect_token)},
        **kwargs,
    )
    if resp.status_code == 403:
        raise Exception(
            "Service account does not have permission to "
            "access the IAP-protected application."
        )
    elif resp.status_code != 200:
        raise Exception(
            "Bad response from application: {!r} / {!r} / {!r}".format(
                resp.status_code, resp.headers, resp.text
            )
        )
    else:
        return resp.text


# END COPIED IAP CODE

Testar a função

Para verificar se a função e o DAG funcionam conforme o esperado:

  1. Aguarde até que a função seja implantada.
  2. Acione a função de acordo com o gatilho especificado. Também é possível acionar a função manualmente selecionando a ação Testar a função em Google Cloud console.
  3. Verifique a página do DAG na interface da Web do Airflow. O DAG precisa ter uma execução ativa ou já concluída.
  4. Na IU do Airflow, verifique os registros de tarefas desta execução. Você verá que a tarefa print_gcs_info gera os dados recebidos da função para os registros:

Exemplo de saída:

[2021-04-04 18:25:44,778] {bash_operator.py:154} INFO - Output:
[2021-04-04 18:25:44,781] {bash_operator.py:158} INFO - Triggered from GCF:
    {bucket: example-storage-for-gcf-triggers, contentType: text/plain,
    crc32c: dldNmg==, etag: COW+26Sb5e8CEAE=, generation: 1617560727904101,
    ... }
[2021-04-04 18:25:44,781] {bash_operator.py:162} INFO - Command exited with
    return code 0h

A seguir