Secret Manager로 환경 구성

이 페이지에서는 Secret Manager를 사용하여 Airflow 연결 및 보안 비밀을 안전하게 저장하는 방법을 보여줍니다.

시작하기 전에

Secret Manager를 사용하려면 Cloud Composer 환경에서 Airflow 1.10.10 이상과 Python 3.6 이상을 사용해야 합니다(Python 2는 지원되지 않음).

Secret Manager로 환경 구성

  1. Secret Manager API를 사용 설정합니다.

    Secret Manager API

  2. 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 문서를 참조하세요.

    [variable_prefix], [connection_prefix] 또는 [sep]에 다른 값을 사용하려면 Secret Manager 백엔드 사용 설정 및 구성 섹션에서 선택적 설정을 사용하세요.

  3. 액세스 제어 구성

    Airflow에 Secret Manager에 저장된 보안 비밀에 대한 액세스 권한을 부여해야 합니다. 이렇게 하려면 secretmanager.versions.access 권한(예: Secret Manager 보안 비밀 접근자)이 포함된 역할을 Cloud Composer 환경이 Secret Manager에 액세스하기 위해 사용하는 서비스 계정에 부여합니다.

    기본적으로 이 계정은 Compute Engine 기본 서비스 계정이지만 환경을 만드는 동안 커스텀 서비스 계정을 지정할 수 있습니다.

    보안 비밀, 프로젝트, 폴더, 조직 수준에서 이 역할을 부여할 수 있습니다.

    액세스 제어

  4. DAG 직렬화 사용 설정

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

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

  5. 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_prefixconn_id를 연결하기 위해 사용되는 구분 기호입니다. 기본값: -.
      • project_id: 보안 비밀이 저장된 Google Cloud 프로젝트 ID입니다.

      예를 들어 backend_kwargs 값은 {"connections_prefix":"example-connections","variables_prefix":"example-variables", "sep":"-"}일 수 있습니다.

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_connectionConnection 객체를 반환합니다. 다음과 같이 연결의 URI 문자열 표현을 가져올 수 있습니다.

exampleConnectionUri = BaseHook.get_connection('exampleConnection').get_uri()