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 as funções do Cloud Run para acionar DAGs do Serviço Gerenciado para Apache Airflow em resposta a eventos.

O Apache Airflow foi criado para executar DAGs regularmente, mas também é possível acionar DAGs em resposta a eventos. Uma forma de fazer isso é usando as funções do Cloud Run para acionar DAGs do Airflow gerenciado quando ocorre um evento especificado.

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 ambiente do Airflow gerenciado. A solicitação contém o identificador e o tipo do evento, além do payload dele.
  3. O Airflow processa essa solicitação e executa o DAG especificado nela. O DAG gera os dados 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 nas configurações de IP particular 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.

Funções necessárias para ativar APIs

Para ativar as APIs, é necessário ter a permissão serviceusage.services.enable. Se você criou o projeto, provavelmente já tem essa permissão pelo papel de proprietário (roles/owner). Caso contrário, você pode receber essa permissão pelo papel de administrador de uso do serviço (roles/serviceusage.serviceUsageAdmin). Saiba como conceder papéis.

Ativar as APIs

gcloud

Ative as APIs do Airflow gerenciado e do Cloud Run functions:

Funções necessárias para ativar APIs

Para ativar as APIs, é necessário ter a permissão serviceusage.services.enable. Se você criou o projeto, provavelmente já tem essa permissão pelo papel de proprietário (roles/owner). Caso contrário, você pode receber essa permissão pelo papel de administrador de uso do serviço (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

As funções do Cloud Run podem acessar a API REST do Airflow por 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 no Controle de acesso ao servidor da Web que seja All IP addresses have access (default) para não bloquear acidentalmente as funções do Cloud Run. É possível 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 Google Cloud console, 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 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 a linguagem preferida com suporte do Cloud Run functions ou do Cloud Run. Este tutorial demonstra uma função do Cloud implementada em Python e Java.

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

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

    Para mais informações sobre como criar acionadores, consulte Criar acionadores 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 acionador precisa ter permissões suficientes para acionar DAGs em ambientes do Airflow gerenciado.

    Recomendamos seguir o princípio de privilégio mínimo e conceder apenas o papel 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 o código deste exemplo, selecione o ambiente de execução Python 3.10 ou posterior 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 acionador especificado. Também é possível acionar a função manualmente selecionando a ação Testar a função no 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