環境に Secret Manager を構成する

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

このページでは、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 を有効にする

gcloud services enable secretmanager.googleapis.com

アクセス制御の構成

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

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

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

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

• （推奨）環境のサービス アカウントにこのロールを付与します。

• Airflow が Secret Manager にアクセスするサービス アカウントをオーバーライドします。

  1. サービス アカウントにこのロールを付与します。
  2. backend_kwargs Airflow 構成オプションの gcp_key_path パラメータを設定して、サービス アカウントの認証情報を含む JSON ファイルを指すようにします。

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 で接続と変数を追加する

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

変数

• [variables_prefix][sep][variable_name] 形式を使用する必要があります。
• [variables_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 表現を使用する必要があります。例: postgresql://login:secret@examplehost:9000

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

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

次のステップ

• Airflow 構成オプションをオーバーライドする
• Airflow REST API にアクセスする