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.
gcloud
Enable the Secret Manager API:
gcloud services enable secretmanager.googleapis.com
アクセス制御の構成
Airflow が Secret Manager に保存されているシークレットにアクセスできるように、アクセス制御を構成する必要があります。
そのためには、シークレットにアクセスするサービス アカウントに secretmanager.versions.access
権限を持つロールが付与されている必要があります。たとえば、Secret Manager のシークレット アクセサーのロールには、この権限が含まれています。
このロールは、シークレット レベル、プロジェクト レベル、フォルダレベル、または組織レベルで付与できます。
以下のいずれかの方法を選択します。
(推奨)ご使用の環境のサービス アカウントにこのロールを付与します。
Airflow が Secret Manager にアクセスするサービス アカウントをオーバーライドします。
- サービス アカウントにこのロールを付与します。
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 バックエンドを有効にして構成する
次の Airflow 構成オプションをオーバーライドします。
セクション キー 値 secrets
backend
airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend
次の 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()