環境に Secret Manager を構成する

Cloud Composer 1 | Cloud Composer 2

このページでは、Secret Manager を使用して Airflow の接続とシークレットを安全に格納する方法を説明します。

始める前に

  • Secret Manager を使用するには、Cloud Composer 環境で Airflow 1.10.10 以降と Python 3.6 以降を使用する必要があります。
  • Python 2 はサポートされていません。

環境に Secret Manager を構成する

このセクションでは、Cloud Composer 環境で Secret を使用できるように、Secret Manager を構成する方法について説明します。

Secret Manager API を有効にする

Console

Enable the Secret Manager API.

Enable the API

gcloud

Enable the Secret Manager API: 

gcloud services enable secretmanager.googleapis.com

アクセス制御の構成

Airflow が Secret Manager に保存されているシークレットにアクセスできるように、アクセス制御を構成する必要があります。

そのためには、シークレットにアクセスするサービス アカウントに secretmanager.versions.access 権限を持つロールが付与されている必要があります。たとえば、Secret Manager のシークレット アクセサーのロールには、この権限が含まれています。

このロールは、シークレット レベル、プロジェクト レベル、フォルダレベル、または組織レベルで付与できます。

以下のいずれかの方法を選択します。

DAG のシリアル化を有効にする

一般に、Secret Manager のバックエンドは、演算子の execute() メソッド内から、または Jinja テンプレートでのみ使用してください。たとえば、変数は var.value.example_var を使用して取得できます。

Airflow ウェブサーバーは、権限が制限された別のサービス アカウントで実行されるため、Secret Manager のシークレットにアクセスできません。DAG コードが（タスクだけでなく）DAG の処理中にシークレットにアクセスし、execute() メソッド内からシークレットにアクセスするように調整できない場合は、DAG のシリアル化を有効にします。これにより、Airflow ウェブサーバーは処理された DAG を受け取り、Secret にアクセスする必要がなくなります。

Secret Manager バックエンドを有効にして構成する

  1. 次の Airflow 構成オプションをオーバーライドします。

    セクション キー 値
    secrets backend airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend

  2. 次の Airflow 構成オプションをオーバーライドして、オプションの設定を追加します。

    セクション キー 値
    secrets backend_kwargs 以下の説明をご覧ください。

    backend_kwargs 値は次の項目を持つ backend_kwargs オブジェクトの JSON 表現です。

    • connections_prefix: 接続を取得するために読み取るシークレット名の接頭辞を指定します。デフォルト値は airflow-connections です。
    • variables_prefix: 変数を取得するために読み取るシークレット名の接頭辞を指定します。デフォルト値は airflow-variables です。
    • gcp_key_path: Google Cloud 認証情報の JSON ファイルへのパス（指定しない場合、デフォルトのサービス アカウントが使用されます）。
    • gcp_keyfile_dict: Google Cloud 認証情報の JSON 辞書。gcp_key_path とは相互に排他的です。
    • sep: connections_prefix と conn_id を連結する際に使用する区切り文字。デフォルト: -
    • project_id: シークレットが保存されている Google Cloud プロジェクト ID。

    たとえば、backend_kwargs の値は {"project_id": "<project id>", "connections_prefix":"example-connections", "variables_prefix":"example-variables", "sep":"-"} に設定できます。

Secret Manager で接続と変数を追加する

シークレットとバージョンの作成で説明されている手順に沿って、シークレットを作成します。

変数

  • [variable_prefix][sep][variable_name] 形式を使用する必要があります。
  • [variable_prefix] のデフォルト値は airflow-variables です。
  • デフォルトの区切り文字 [sep] は - です。

たとえば、変数名が example-var の場合、シークレット名は airflow-variables-example-var です。

接続名

  • [connection_prefix][sep][connection_name] 形式を使用する必要があります。
  • [connection_prefix] のデフォルト値は airflow-connections です。
  • デフォルトの区切り文字 [sep] は - です。

たとえば、接続名が exampleConnection の場合、シークレット名は airflow-connections-exampleConnection です。

接続値

  • URI 表現を使用する必要があります。例: mysql://login:password@examplehost:9000。

  • URI は URL エンコードされている必要があります（パーセントでエンコード）。たとえば、スペース記号を含むパスワードは、mysql://login:secret%20password@examplehost:9000 のように URL エンコードする必要があります。

Airflow には、接続 URI を生成するための便利なメソッドがあります。JSON エクストラで複雑な URL をエンコードする例については、Airflow ドキュメントをご覧ください。

Cloud Composer で Secret Manager を使用する

変数と接続を取得するとき、Cloud Composer は最初に Secret Manager をチェックします。リクエストされた変数または接続が見つからない場合、Cloud Composer は次に環境変数と Airflow データベースを確認します。

Jinja テンプレートを使用して変数を読み取る

Secret Manager では、テンプレート化された演算子フィールド（実行時に解決されます）について、JINJA テンプレートを使用して変数を読み取ることができます。

airflow-variables-secret_filename シークレットの場合:

file_name = '{{var.value.secret_filename}}'

カスタム演算子とコールバックを使用して変数の読み取る

Secret Manager を使用して、カスタム オペレータの変数やオペレータからのコールバック メソッドを読み取ることもできます。DAG 内から変数を読み取るとパフォーマンスが低下する可能性があるため、DAG で変数を使用する場合は Jinja テンプレートを使用してください。

たとえば、airflow-variables-secret_filename シークレットの場合:

from airflow.models.variable import Variable
file_name = Variable.get('secret_filename')

読み取り接続

カスタム演算子を作成する場合を除き、接続に直接アクセスする必要はほとんどありません。ほとんどのフックは、インスタンス化パラメータとして接続名を取得し、タスクの実行時にシークレット バックエンドから接続を自動的に取得する必要があります。

独自のフックを作成する場合には、接続を直接読み取ることが有用な可能性があります。

たとえば、airflow-connections-exampleConnection 接続の場合:

from airflow.hooks.base_hook import BaseHook
exampleConnection = BaseHook.get_connection('exampleConnection')

BaseHook.get_connection は Connection オブジェクトを返します。次のように、接続の URI 文字列表現を取得できます。

exampleConnectionUri = BaseHook.get_connection('exampleConnection').get_uri()

次のステップ