환경에 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 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 직렬화 사용 설정

일반적으로 연산자의 execute() 메서드 또는 Jinja 템플릿으로만 Secret Manager 백엔드를 사용해야 합니다. 예를 들어 var.value.example_var을 사용하여 변수를 검색할 수 있습니다.

Airflow 웹 서버는 권한이 제한된 다른 서비스 계정으로 실행되므로 Secret Manager의 보안 비밀에 액세스할 수 없습니다. DAG 코드가 태스크뿐만 아니라 DAG 처리 중 보안 비밀에 액세스하고, execute() 메서드 내에서 보안 비밀에 액세스하도록 조정할 수 없으면 DAG 직렬화를 사용 설정합니다. 이렇게 한 후에는 Airflow 웹 서버에서 처리된 DAG가 사용되고 보안 비밀에 액세스할 필요가 없습니다.

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로 인코딩되어야 합니다(퍼센트 인코딩). 예를 들어 공백 기호가 있는 비밀번호는 다음과 같이 URL로 인코딩되어야 합니다. mysql://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()

다음 단계