環境に 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

Secret Manager API を有効にします。

API を有効にする

gcloud

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

次のステップ