Memicu DAG Managed Service for Apache Airflow dengan fungsi Cloud Run dan Airflow REST API

Managed Airflow (Gen 3) | Managed Airflow (Gen 2) | Managed Airflow (Legacy Gen 1)

Halaman ini menjelaskan cara menggunakan fungsi Cloud Run untuk memicu DAG Managed Service untuk Apache Airflow sebagai respons terhadap peristiwa.

Apache Airflow dirancang untuk menjalankan DAG sesuai jadwal rutin, tetapi Anda juga dapat memicu DAG sebagai respons terhadap peristiwa. Salah satu cara untuk melakukannya adalah dengan menggunakan Cloud Run Functions untuk memicu DAG Managed Airflow saat peristiwa tertentu terjadi.

Anda juga dapat:

Contoh dalam panduan ini menunjukkan fungsi yang memicu DAG sebagai respons terhadap suatu peristiwa:

  1. Anda mengonfigurasi pemicu untuk fungsi di Cloud Run Functions.
  2. Saat dipicu, fungsi akan membuat permintaan untuk memicu DAG melalui Airflow REST API dari lingkungan Managed Airflow Anda. Permintaan berisi ID dan jenis peristiwa, serta payload peristiwa.
  3. Airflow memproses permintaan ini dan menjalankan DAG yang ditentukan dalam permintaan. DAG menampilkan data yang diteruskan kepadanya dari fungsi.

Sebelum memulai

Bagian ini mencantumkan langkah-langkah persiapan.

Periksa konfigurasi jaringan lingkungan Anda

Solusi ini tidak berfungsi dalam konfigurasi Kontrol Layanan VPC dan IP Pribadi karena tidak mungkin mengonfigurasi konektivitas dari fungsi Cloud Run ke server web Airflow dalam konfigurasi ini.

Mengaktifkan API untuk project Anda

Konsol

Aktifkan Managed Airflow dan Cloud Run Functions API.

Peran yang diperlukan untuk mengaktifkan API

Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Admin Penggunaan Layanan (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran.

Aktifkan API

gcloud

Aktifkan Managed Airflow dan Cloud Run Functions API:

Peran yang diperlukan untuk mengaktifkan API

Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Admin Penggunaan Layanan (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran. 

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

Mengaktifkan Airflow REST API

Bergantung pada versi Airflow Anda:

Mengizinkan panggilan API ke Airflow REST API menggunakan kontrol akses jaringan server web

Fungsi Cloud Run dapat menghubungi Airflow REST API melalui alamat IPv4 atau IPv6.

Jika Anda tidak yakin dengan rentang IP panggilan, gunakan opsi konfigurasi default di Kontrol Akses Webserver, yaitu All IP addresses have access (default), agar tidak memblokir Cloud Run Functions Anda secara tidak sengaja. Anda selalu dapat mengonfigurasi akses jaringan server web nanti.

Mendapatkan URL server web Airflow

Contoh ini membuat permintaan REST API ke endpoint server web Airflow. Anda menggunakan bagian URL antarmuka web Airflow sebelum .appspot.com dalam kode Cloud Function Anda.

Konsol

  1. Di konsol Google Cloud , buka halaman Environments.

    Buka Environments

  2. Klik nama lingkungan Anda.

  3. Di halaman Detail lingkungan, buka tab Konfigurasi lingkungan.

  4. URL server web Airflow tercantum di item Airflow web UI.

gcloud

Jalankan perintah berikut:

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

Ganti:

  • ENVIRONMENT_NAME dengan nama lingkungan.
  • LOCATION dengan region tempat lingkungan berada.

Mendapatkan client_id proxy IAM

Untuk membuat permintaan ke endpoint Airflow REST API, fungsi ini memerlukan ID klien proxy Identity and Access Management yang melindungi server web Airflow.

Managed Airflow tidak memberikan informasi ini secara langsung. Sebagai gantinya, buat permintaan yang tidak diautentikasi ke server web Airflow dan ambil ID klien dari URL pengalihan:

cURL 

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

Ganti AIRFLOW_URL dengan URL antarmuka web Airflow.

Pada output, telusuri string setelah client_id. Contoh:

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

Python

Simpan kode berikut dalam file bernama get_client_id.py. Isi nilai untuk project_id, location, dan composer_environment, lalu jalankan kode di Cloud Shell atau lingkungan lokal Anda. 

# 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])

Mengupload DAG ke lingkungan Anda

Mengupload DAG ke lingkungan Anda. DAG contoh berikut menampilkan output konfigurasi eksekusi DAG yang diterima. Anda memicu DAG ini dari fungsi, yang akan Anda buat nanti dalam panduan ini.

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 }}}}')

Men-deploy fungsi yang memicu DAG

Anda dapat men-deploy fungsi menggunakan bahasa pilihan yang didukung oleh fungsi Cloud Run atau Cloud Run. Tutorial ini menunjukkan Cloud Function yang diimplementasikan di Python dan Java.

Menentukan parameter konfigurasi fungsi

  • Pemicu: Pilih pemicu Eventarc atau beberapa pemicu untuk fungsi Anda.

    Untuk mengetahui informasi selengkapnya tentang cara membuat pemicu, lihat Membuat pemicu dengan Eventarc. Misalnya, Anda dapat memicu fungsi dari Cloud Storage menggunakan Eventarc.

  • Akun layanan: Akun layanan yang Anda tentukan untuk pemicu harus memiliki izin yang memadai untuk memicu DAG di lingkungan Managed Airflow.

    Sebaiknya ikuti prinsip hak istimewa minimum dan berikan hanya peran Pengguna Composer (composer.user) kepadanya. Untuk mengetahui informasi selengkapnya tentang mengonfigurasi izin, lihat Peran dan izin untuk target Cloud Run.

  • Titik entri fungsi:

    • (Python) Saat menambahkan kode untuk contoh ini, pilih runtime Python 3.10 atau yang lebih baru dan tentukan trigger_dag sebagai titik entri.

Menambahkan persyaratan

Tentukan dependensi dalam file requirements.txt:

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

Menambahkan kode fungsi

Masukkan kode berikut ke file main.py dan lakukan penggantian berikut:

  • Ganti nilai variabel client_id dengan nilai client_id yang Anda peroleh sebelumnya.

  • Ganti nilai variabel webserver_id dengan ID project tenant Anda, yang merupakan bagian dari URL antarmuka web Airflow sebelum .appspot.com. Anda telah mendapatkan URL antarmuka web Airflow sebelumnya.

  • Tentukan versi Airflow REST API yang Anda gunakan:

    • Jika Anda menggunakan Airflow REST API yang stabil, tetapkan variabel USE_EXPERIMENTAL_API ke False.
    • Jika Anda menggunakan Airflow REST API eksperimental, tidak ada perubahan yang diperlukan. Variabel USE_EXPERIMENTAL_API sudah disetel ke 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

Menguji fungsi

Untuk memeriksa apakah fungsi dan DAG Anda berfungsi sebagaimana mestinya:

  1. Tunggu hingga fungsi Anda di-deploy.
  2. Memicu fungsi sesuai dengan pemicu yang ditentukan. Anda juga dapat memicu fungsi secara manual dengan memilih tindakan Uji fungsi untuk fungsi tersebut di konsol Google Cloud .
  3. Periksa halaman DAG di antarmuka web Airflow. DAG harus memiliki satu proses DAG yang aktif atau sudah selesai.
  4. Di UI Airflow, periksa log tugas untuk proses ini. Anda akan melihat bahwa tugas print_gcs_info menampilkan data yang diterima dari fungsi ke log:

Contoh output:

[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

Langkah berikutnya