google cloud storage - Accessing GOOGLE_APPLICATION_CREDENTIALS in a kubernetes operator

Question

I am adding a task to an airflow dag as follows:

examples_task = KubernetesPodOperator(
    task_id='examples_generation',
    dag=dag,
    namespace='test',
    image='test_amazon_image',
    name='pipe-labelled-examples-generation-tf-record-operator',
    env={
        'GOOGLE_APPLICATION_CREDENTIALS': Variable.get('google_cloud_credentials')
    },
    arguments=[
        "--assets_path", Variable.get('assets_path'),
        "--folder_source", Variable.get('folder_source'),
        "--folder_destination", Variable.get('folder_destination'),
        "--gcs_folder_destination", Variable.get('gcs_folder_destination'),
        "--aws_region", Variable.get('aws_region'),
        "--s3_endpoint", Variable.get('s3_endpoint')
    ],
    get_logs=True)

I thought I could paste the service account json file as a variable and call it but this doesn't work and airflow/google documentation isn't clear. How do you do this?

question from:https://stackoverflow.com/questions/65941224/accessing-google-application-credentials-in-a-kubernetes-operator

Answer 1

Solutions to port the json into an argument

examples_task = KubernetesPodOperator(
    task_id='examples_generation',
    dag=dag,
    namespace='test',
    image='test_amazon_image',
    name='pipe-labelled-examples-generation-tf-record-operator',
    arguments=[
        "--folder_source", Variable.get('folder_source'),
        "--folder_destination", Variable.get('folder_destination'),
        "--gcs_folder_destination", Variable.get('gcs_folder_destination'),
        "--aws_region", Variable.get('aws_region'),
        "--s3_endpoint", Variable.get('s3_endpoint')
        "--gcs_credentials", Variable.get('google_cloud_credentials')

    ],
    get_logs=True)

then in the cli set

import json
from google.cloud import storage
from google.oauth2 import service_account

credentials = service_account.Credentials.from_service_account_info(json.loads(gcs_credentials))
client = storage.Client(project='project_id', credentials=credentials)