Issue while connecting to GCP using Secrets / Impersonation Chain in DataprocCreateClusterOperator

[abhishekshenoy]

We are using airflow 2.0.0b2 version and our airflow runs on GKE clusters.

The executor pods run with a service account that has permission to access secrets in Google Secret Manager.

      DataprocCreateClusterTask ------------------------>Gets Secret-----------------------> Used to Create Cluster 
        (svc-executor-account)                    (svc-hadoop-admin-account)
(Pods Secret Accessor role Account)      (Secret is a service account with Dataproc admin role)

When i pass the secret name (prefix and all correctly set) to gcp_conn_id , i see that though the secrets are correctly fetched it is not used to create a cluster in Dataproc. Rather the default account in the pod is used to create the cluser which does not have the dataproc admin role.

Task definition is below:

GCP_CONNECTION = "poc-dataproc"

create_cluster = DataprocCreateClusterOperator(
    dag=dag_dataproc,
    task_id="create_cluster",
    project_id=PROJECT_ID,
    cluster_config=CLUSTER_CONFIG,
    region=REGION,
    cluster_name=CLUSTER_NAME,
    labels={'product': 'sample-label'},
    # request_id="",
    delete_on_error=True,
    use_if_exists=True,
    retry=Retry(maximum=10.0),
    timeout=1800,
    gcp_conn_id=GCP_CONNECTION
)

I get the below exception:

AIRFLOW_CTX_DAG_RUN_ID=manual__2020-12-13T11:36:03.968854+00:00
[2020-12-13 11:36:27,927] {dataproc.py:603} INFO - Creating cluster: first-cluster-setup
[2020-12-13 11:36:28,365] {credentials_provider.py:300} INFO - Getting connection using `google.auth.default()` since no key file is defined for hook.
[2020-12-13 11:36:28,738] {taskinstance.py:1396} ERROR - 403 Not authorized to requested resource.
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/google/api_core/grpc_helpers.py", line 57, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/grpc/_channel.py", line 923, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/home/airflow/.local/lib/python3.8/site-packages/grpc/_channel.py", line 826, in _end_unary_response_blocking
    raise _InactiveRpcError(state)
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
	status = StatusCode.PERMISSION_DENIED
	details = "Not authorized to requested resource."
	debug_error_string = "{"created":"@1607859388.737130770","description":"Error received from peer ipv4:172.217.1.202:443","file":"src/core/lib/surface/call.cc","file_line":1061,"grpc_message":"Not authorized to requested resource.","grpc_status":7}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1086, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1260, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1300, in _execute_task
    result = task_copy.execute(context=context)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/cloud/operators/dataproc.py", line 607, in execute
    cluster = self._create_cluster(hook)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/cloud/operators/dataproc.py", line 535, in _create_cluster
    operation = hook.create_cluster(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/common/hooks/base_google.py", line 425, in inner_wrapper
    return func(self, *args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/dataproc.py", line 291, in create_cluster
    result = client.create_cluster(
  File "/home/airflow/.local/lib/python3.8/site-packages/google/cloud/dataproc_v1beta2/gapic/cluster_controller_client.py", line 296, in create_cluster
    operation = self._inner_api_calls["create_cluster"](
  File "/home/airflow/.local/lib/python3.8/site-packages/google/api_core/gapic_v1/method.py", line 145, in __call__
    return wrapped_func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/google/api_core/retry.py", line 281, in retry_wrapped_func
    return retry_target(
  File "/home/airflow/.local/lib/python3.8/site-packages/google/api_core/retry.py", line 184, in retry_target
    return target()
  File "/home/airflow/.local/lib/python3.8/site-packages/google/api_core/timeout.py", line 102, in func_with_timeout
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/google/api_core/grpc_helpers.py", line 59, in error_remapped_callable
    six.raise_from(exceptions.from_grpc_error(exc), exc)
  File "<string>", line 3, in raise_from

I know i am missing something here , i tried retrieving the connection and passing the connection.uri string to gcp_conn_id using templates but as gcp_conn_id is not a templated field the variable was not getting resolved.

def get_secrets(**kwargs):
    conn = BaseHook.get_connection(kwargs['my_conn_id'])
    return conn.get_uri()

check_connection_retrieval = PythonOperator(
    task_id='check_connection_retrieval',
    python_callable=get_secrets,
    op_kwargs={'my_conn_id': GCP_CONNECTION},
    dag=dag_dataproc
)

# [START how_to_cloud_dataproc_create_cluster_operator]
create_cluster = DataprocCreateClusterOperator(
    dag=dag_dataproc,
    task_id="create_cluster",
    project_id=PROJECT_ID,
    cluster_config=CLUSTER_CONFIG,
    region=REGION,
    cluster_name=CLUSTER_NAME,
    labels={'product': 'sample-label'},
    # request_id="",
    delete_on_error=True,
    use_if_exists=True,
    retry=Retry(maximum=10.0),
    timeout=1800,
    gcp_conn_id='{{ task_instance.xcom_pull(task_ids='check_connection_retrieval', key='return_value') }}'
)

I am still unable to establish a connection when i try gcp_conn_id with connections retrieved from secrets.

I moved from using secrets to using impersonation_chain by giving the service accounts with which my pods run serviceAccountTokenCreator role for .

The below Task definition works fine in my local Docker setup and i am able to create a cluster wherein my tasks are running as local executor.

create_cluster = DataprocCreateClusterOperator(
    dag=dag_dataproc,
    task_id="create_cluster",
    project_id=PROJECT_ID,
    cluster_config=CLUSTER_CONFIG,
    region=REGION,
    cluster_name=CLUSTER_NAME,
    labels={'product': 'sample-label'},
    # request_id="",
    delete_on_error=True,
    use_if_exists=True,
    retry=Retry(maximum=10.0),
    timeout=1800,
    impersonation_chain=IMPERSONATE_SERVICE_ACCOUNT
)

The same when run on my dev setup throws the below excpetion , i do not understand why it is looking into the secret manager to get google_cloud_default information ? Any help would unblock me in proceeding with setting up a dataproc flow using Airflow. Have attached the file which has the whole exception stack trace.
impersonation_chain_exception.log

---

[2020-12-13 07:31:00,542] {taskinstance.py:1018} INFO - Starting attempt 1 of 2
[2020-12-13 07:31:00,542] {taskinstance.py:1019} INFO -

[2020-12-13 07:31:00,640] {taskinstance.py:1038} INFO - Executing <Task(DataprocCreateClusterOperator): create_cluster> on 2020-12-13T07:30:40.273638+00:00
[2020-12-13 07:31:00,645] {standard_task_runner.py:50} INFO - Started process 14 to run task
[2020-12-13 07:31:00,650] {standard_task_runner.py:74} INFO - Running: ['airflow', 'tasks', 'run', '4_spark_submit_dataproc', 'create_cluster', '2020-12-13T07:30:40.273638+00:00', '--job-id', '203', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/templates/4_spark_submit_dataproc.py', '--cfg-path', '/tmp/tmpeqcdvbvb']
[2020-12-13 07:31:00,651] {standard_task_runner.py:75} INFO - Job 203: Subtask create_cluster
[2020-12-13 07:31:01,189] {logging_mixin.py:103} INFO - Running <TaskInstance: 4_spark_submit_dataproc.create_cluster 2020-12-13T07:30:40.273638+00:00 [running]> on host 4sparksubmitdataproccreatecluster-8d0cc477352847c6a93d4b2568051
[2020-12-13 07:31:01,550] {taskinstance.py:1230} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_EMAIL=airflow@example.com
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=4_spark_submit_dataproc
AIRFLOW_CTX_TASK_ID=create_cluster
AIRFLOW_CTX_EXECUTION_DATE=2020-12-13T07:30:40.273638+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2020-12-13T07:30:40.273638+00:00
[2020-12-13 07:31:01,550] {dataproc.py:603} INFO - Creating cluster: first-cluster-setup
[2020-12-13 07:31:02,030] {secret_manager_client.py:89} ERROR - Google Cloud API Call Error (PermissionDenied): No access for Secret ID airflow-connections-google_cloud_default.
Did you add 'secretmanager.versions.access' permission?
[2020-12-13 07:31:02,103] {taskinstance.py:1396} ERROR - (psycopg2.errors.UndefinedColumn) column connection.description does not exist
LINE 1: ...id, connection.conn_type AS connection_conn_type, connection...
^

[SQL: SELECT connection.password AS connection_password, connection.extra AS connection_extra, connection.id AS connection_id, connection.conn_id AS connection_conn_id, connection.conn_type AS connection_conn_type, connection.description AS connection_description, connection.host AS connection_host, connection.schema AS connection_schema, connection.login AS connection_login, connection.port AS connection_port, connection.is_encrypted AS connection_is_encrypted, connection.is_extra_encrypted AS connection_is_extra_encrypted
FROM connection
WHERE connection.conn_id = %(conn_id_1)s
LIMIT %(param_1)s]
[parameters: {'conn_id_1': 'google_cloud_default', 'param_1': 1}]
(Background on this error at: http://sqlalche.me/e/13/f405)
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1276, in _execute_context
self.dialect.do_execute(
File "/home/airflow/.local/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 593, in do_execute
cursor.execute(statement, parameters)
psycopg2.errors.UndefinedColumn: column connection.description does not exist
LINE 1: ...id, connection.conn_type AS connection_conn_type, connection...
^