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

일반적으로 연산자의 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 표현을 사용해야 합니다. 예를 들면 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()

다음 단계