Mengakses Airflow REST API

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

Apache Airflow memiliki antarmuka REST API yang dapat Anda gunakan untuk melakukan tugas seperti mendapatkan informasi tentang tugas dan operasi DAG, memperbarui DAG, mendapatkan konfigurasi Airflow, menambahkan dan menghapus koneksi, serta mencantumkan pengguna.

Untuk mengetahui contoh penggunaan Airflow REST API dengan fungsi Cloud Run, lihat artikel Memicu DAG dengan fungsi Cloud Run.

Versi Airflow REST API

• Airflow 1 menggunakan REST API eksperimental.
• Airflow 2 menggunakan REST API stabil. REST API eksperimental tidak digunakan lagi oleh Airflow.
• Anda masih dapat menggunakan REST API eksperimental di Airflow 2 jika mengaktifkannya melalui penggantian konfigurasi Airflow, seperti yang dijelaskan lebih lanjut.

Mengonfigurasi REST API Airflow yang stabil

Airflow 2

REST API stabil diaktifkan secara default di Airflow 2. Cloud Composer menggunakan backend autentikasi API-nya sendiri, yang terintegrasi dengan Identity-Aware Proxy.

Otorisasi berfungsi dengan cara standar yang disediakan oleh Airflow. Saat pengguna baru memberikan otorisasi melalui API, akun pengguna tersebut akan mendapatkan peran Op secara default.

Anda dapat mengaktifkan atau menonaktifkan REST API stabil, atau mengubah peran pengguna default dengan mengganti opsi konfigurasi Airflow berikut:

Bagian	Kunci	Nilai	Catatan
api	(Airflow 2.2.5 dan yang lebih lama) auth_backend (Airflow 2.3.0 dan yang lebih baru) auth_backends	airflow.composer.api.backend.composer_auth	Untuk menonaktifkan REST API stabil, ubah ke airflow.api.auth.backend.deny_all
api	composer_auth_user_registration_role	Op	Anda dapat menentukan peran lainnya.

Aliran udara 1

REST API stabil tidak tersedia di Airflow 1. Sebagai gantinya, Anda dapat menggunakan REST API eksperimental.

Mengonfigurasi REST API Airflow eksperimental

Airflow 2

Secara default, fitur autentikasi API dinonaktifkan di API eksperimental. Server web Airflow menolak semua permintaan ke server tersebut. Untuk mengaktifkan fitur autentikasi API dan API eksperimental Airflow 2, ganti opsi konfigurasi Airflow berikut:

Bagian	Kunci	Nilai	Catatan
api	(Airflow 2.2.5 dan yang lebih lama) auth_backend (Airflow 2.3.0 dan yang lebih baru) auth_backends	airflow.api.auth.backend.default	Nilai defaultnya adalah airflow.composer.api.backend.composer_auth.
api	enable_experimental_api	True	Nilai defaultnya adalah False.

Aliran udara 1

Catatan: Bagian ini berlaku untuk versi Cloud Composer yang menggunakan Airflow 1.10.12 dan yang lebih baru. Jika lingkungan Anda menggunakan Airflow 1.10.10 dan versi sebelumnya, REST API eksperimental akan diaktifkan secara default.

Secara default, fitur autentikasi API dinonaktifkan di Airflow 1.10.11 dan versi yang lebih baru. Server web Airflow menolak semua permintaan yang Anda buat. Anda menggunakan permintaan untuk memicu DAG, jadi aktifkan fitur ini.

Untuk mengaktifkan fitur autentikasi API di Airflow 1, ganti opsi konfigurasi Airflow berikut:

Bagian	Kunci	Nilai	Catatan
api	auth_backend	airflow.api.auth.backend.default	Defaultnya adalah airflow.api.auth.backend.deny_all

Setelah Anda menetapkan opsi konfigurasi ini ke airflow.api.auth.backend.default, server web Airflow akan menerima semua permintaan API tanpa autentikasi.

Meskipun server web Airflow itu sendiri tidak memerlukan autentikasi, Cloud Composer menggunakan lapisan autentikasi sendiri untuk melindunginya, yang terintegrasi dengan Identity-Aware Proxy.

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

Bergantung pada metode yang digunakan untuk memanggil Airflow REST API, metode pemanggil dapat menggunakan alamat IPv4 atau IPv6. Jangan lupa untuk berhenti memblokir traffic IP ke Airflow REST API menggunakan kontrol akses server web.

Gunakan opsi konfigurasi default yang merupakan All IP addresses have access (default) jika Anda tidak yakin dari alamat IP mana panggilan Anda ke Airflow REST API akan dikirim.

Melakukan panggilan ke Airflow REST API

Mendapatkan client_id proxy IAM

Untuk membuat permintaan ke endpoint REST API Airflow, fungsi ini memerlukan client ID proxy IAM yang melindungi server web Airflow.

Cloud Composer tidak memberikan informasi ini secara langsung. Sebagai gantinya, buat permintaan yang tidak diautentikasi ke server web Airflow dan ambil client ID dari URL alihan:

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

Memanggil Airflow REST API menggunakan client_id

Lakukan penggantian berikut:

• Ganti nilai variabel client_id dengan nilai client_id yang diperoleh pada langkah 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 pada langkah 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 diperlukan perubahan. Variabel USE_EXPERIMENTAL_API sudah ditetapkan 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

Mengakses Airflow REST API menggunakan akun layanan

Catatan: Jika lingkungan Anda menggunakan Airflow 1, bagian ini hanya berlaku jika Kontrol Akses UI Airflow diaktifkan di lingkungan Anda. Jika tidak, Airflow tidak akan melakukan pengelolaan penggunanya sendiri.

Database Airflow membatasi panjang kolom email hingga 64 karakter. Akun layanan terkadang memiliki alamat email yang lebih panjang dari 64 karakter. Anda tidak dapat membuat pengguna Airflow untuk akun layanan tersebut dengan cara biasa. Jika tidak ada pengguna Airflow untuk akun layanan tersebut, mengakses Airflow REST API akan menghasilkan error HTTP 401 dan 403.

Sebagai solusi, Anda dapat melakukan prapendaftaran pengguna Airflow untuk akun layanan. Untuk melakukannya, gunakan accounts.google.com:NUMERIC_USER_ID sebagai nama pengguna, dan string unik apa pun sebagai email.

1. Untuk mendapatkan NUMERIC_USER_ID untuk akun layanan, jalankan:

   gcloud iam service-accounts describe \
     SA_NAME@PROJECT_ID. \
     --format="value(oauth2ClientId)"
   

   Ganti:

   • SA_NAME dengan nama akun layanan.
   • PROJECT_ID dengan Project ID.

2. Buat pengguna Airflow dengan peran Op untuk akun layanan:

   UI Airflow

   1. Buka UI Airflow.

   2. Buka Admin > Pengguna, lalu klik Buat. Pengguna Airflow Anda harus memiliki peran Admin untuk membuka halaman ini.

   3. Tentukan accounts.google.com:NUMERIC_USER_ID sebagai nama pengguna. Ganti NUMERIC_USER_ID dengan ID pengguna yang diperoleh di langkah sebelumnya.

   4. Tentukan ID unik sebagai email. Anda dapat menggunakan string unik apa pun.

   5. Tentukan peran untuk pengguna. Misalnya, Op.

   6. Pastikan kotak centang Aktif? dicentang.

   7. Tentukan nama depan dan nama belakang untuk pengguna. Anda dapat menggunakan string apa pun.

   8. Klik Simpan.

   gcloud

   Catatan: Di lingkungan Cloud Composer dengan Airflow 1, Anda tidak dapat menjalankan perintah Airflow CLI yang membuat pengguna melalui gcloud. Anda dapat menggunakan UI Airflow.

   Di Airflow 2, jalankan perintah Airflow CLI berikut:

   gcloud composer environments run ENVIRONMENT_NAME \
       --location LOCATION \
       users create -- \
       -u accounts.google.com:NUMERIC_USER_ID \
       -e UNIQUE_ID  \
       -f UNIQUE_ID \
       -l - -r Op --use-random-password
   

   Ganti:

   • ENVIRONMENT_NAME dengan nama lingkungan.
   • LOCATION dengan region tempat lingkungan tersebut berada.
   • NUMERIC_USER_ID dengan ID pengguna yang diperoleh di langkah sebelumnya.
   • UNIQUE_ID dengan ID untuk pengguna Airflow. Anda dapat menggunakan string unik apa pun.

3. Setelah Anda membuat pengguna Airflow untuk akun layanan, pemanggil yang diautentikasi sebagai akun layanan akan dikenali sebagai pengguna yang terdaftar sebelumnya, dan login ke Airflow.

Menskalakan komponen Airflow REST API

Endpoint Airflow REST API dan UI Airflow dijalankan dalam server web Airflow. Jika Anda menggunakan REST API secara intensif, pertimbangkan untuk meningkatkan jumlah CPU dan memori yang tersedia untuk server web Airflow, berdasarkan beban yang diharapkan.

Langkah selanjutnya

• Memicu DAG dengan fungsi Cloud Run
• Antarmuka command line Airflow